Kafka开源项目指南

Kafka开源项目指南 实战SparkStream+Kafka+Redis实时计算商品销售额
cover

实战SparkStream+Kafka+Redis实时计算商品销售额

写在前面2016年天猫双十一当天，零点的倒计时话音未落，52秒交易额冲破10亿。随后，又迅速在0时6分28秒，达到100亿！每一秒开猫大屏上的交易额都在刷新，这种时实刷新的大屏看着感觉超爽。天猫这个大屏后面的技术应该是使用流计算，阿里使用Java将Storm重写了，叫JStrom(https://github.com/a...

徐令予

31人浏览 · 2024-07-26 02:21:49
徐令予  ·  2024-07-26 02:21:49 发布


写在前面

2016年天猫双十一当天，零点的倒计时话音未落，52秒交易额冲破10亿。随后，又迅速在0时6分28秒，达到100亿！每一秒开猫大屏上的交易额都在刷新，这种时实刷新的大屏看着感觉超爽。天猫这个大屏后面的技术应该是使用流计算，阿里使用Java将Storm重写了，叫JStrom( https://github.com/alibaba/jstorm)，最近学习SparkStream和Kafka,可以简单模仿一下这个时实计算成交额的过程，主要目的是实际运用这些技术，也了解一下技术的运用场景，加深对技术的理解。

实战SparkStream+Kafka+Redis实时计算商品销售额_时实计算

实时计算模型

下图所示为通用SparkStream时实计算模型，主要分为三部分

  1. 数据源
    我们这里的数据源选用了Kafka，关于Kafka的安装与使用说明可以参考这里 https://kafkadoc.beanmr.com/
  2. SparkStream计算
    SparkStream是实时计算的核心，这们这里也是近时实计算，选择一个时间窗口，对时间窗口中的数据做离线计算。
  3. 数据落地
    SparkStream算好的结果可以存HDFS/Mysql/Redis等等，我们这里对商品销售额计算过程有涉及累加，所以选择了Redis

实战SparkStream+Kafka+Redis实时计算商品销售额_kakfa_02

业务模型介绍

我们模仿一个电商系统，每时每刻都有订单成交，每一笔成交的数据以一个事件发送到Kafka中，SparkStream每一分中从Kafka中读取一次数据，计算一分钟内每个商品的销售额，然而写入Redis，并在Redis中累加每分钟的数据，Redis中主要存三种结果数量，从开始到当前总销售额、从开始到当前每个商品销售额、上一分钟每个商品的销售额

Kafka生产者，模拟每时每刻订单交易
object OrderProducer {


  def main(args: Array[String]): Unit = {

    //Kafka参数设置
    val topic = "order"
    val brokers = "127.0.0.1:9092"
    val props = new Properties()
    props.put("metadata.broker.list", brokers)
    props.put("serializer.class", "kafka.serializer.StringEncoder")
    val kafkaConfig = new ProducerConfig(props)
    //创建生产者
    val producer = new Producer[String, String](kafkaConfig)

    while (true) {
      //随机生成10以内ID
      val id = Random.nextInt(10)
      //创建订单成交事件
      val event = new JSONObject();
      //商品ID
      event.put("id", id)
      //商品成交价格
      event.put("price", Random.nextInt(10000))

      //发送信息
      producer.send(new KeyedMessage[String, String](topic, event.toString))
      println("Message sent: " + event)
      //随机暂停一段时间
      Thread.sleep(Random.nextInt(100))
    }
  }

}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.

生产者输出结果：

Message sent: {"price":3959,"id":6}
Message sent: {"price":1579,"id":0}
Message sent: {"price":857,"id":6}
Message sent: {"price":8440,"id":1}
Message sent: {"price":6873,"id":6}
Message sent: {"price":6202,"id":2}
Message sent: {"price":8403,"id":6}
Message sent: {"price":7866,"id":2}
Message sent: {"price":9441,"id":5}
Message sent: {"price":6880,"id":4}
Message sent: {"price":4572,"id":5}
Message sent: {"price":509,"id":3}
Message sent: {"price":7526,"id":0}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.

上述代码主要模拟一家店铺有十件商品，ID从0到9，每隔一小段随机时间成交一单，成交价格以分为单位，每成交一笔就像Kafka中发送一个消息，用这个生产者模拟线上的真实交易，在实际生产中成交数据可以从日志中获取。

Kafka消费者，SparkStream时实计算
object OrderConsumer {
  //Redis配置
  val dbIndex = 0
  //每件商品总销售额
  val orderTotalKey = "app::order::total"
  //每件商品上一分钟销售额
  val oneMinTotalKey = "app::order::product"
  //总销售额
  val totalKey = "app::order::all"


  def main(args: Array[String]): Unit = {

    // 创建 StreamingContext 时间片为1秒
    val conf = new SparkConf().setMaster("local").setAppName("UserClickCountStat")
    val ssc = new StreamingContext(conf, Seconds(1))

    // Kafka 配置
    val topics = Set("order")
    val brokers = "127.0.0.1:9092"
    val kafkaParams = Map[String, String](
      "metadata.broker.list" -> brokers,
      "serializer.class" -> "kafka.serializer.StringEncoder")


    // 创建一个 direct stream
    val kafkaStream = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, topics)

    //解析JSON
    val events = kafkaStream.flatMap(line => Some(JSON.parseObject(line._2)))

    // 按ID分组统计个数与价格总合
    val orders = events.map(x => (x.getString("id"), x.getLong("price"))).groupByKey().map(x => (x._1, x._2.size, x._2.reduceLeft(_ + _)))

    //输出
    orders.foreachRDD(x =>
      x.foreachPartition(partition =>
        partition.foreach(x => {


          println("id=" + x._1 + " count=" + x._2 + " price=" + x._3)

          //保存到Redis中
          val jedis = RedisClient.pool.getResource
          jedis.select(dbIndex)
          //每个商品销售额累加
          jedis.hincrBy(orderTotalKey, x._1, x._3)
          //上一分钟第每个商品销售额
          jedis.hset(oneMinTotalKey, x._1.toString, x._3.toString)
          //总销售额累加
          jedis.incrBy(totalKey, x._3)
          RedisClient.pool.returnResource(jedis)


        })
      ))


    ssc.start()
    ssc.awaitTermination()
  }

}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
  • 30.
  • 31.
  • 32.
  • 33.
  • 34.
  • 35.
  • 36.
  • 37.
  • 38.
  • 39.
  • 40.
  • 41.
  • 42.
  • 43.
  • 44.
  • 45.
  • 46.
  • 47.
  • 48.
  • 49.
  • 50.
  • 51.
  • 52.
  • 53.
  • 54.
  • 55.
  • 56.
  • 57.
  • 58.
  • 59.
  • 60.
  • 61.
  • 62.
  • 63.

消费者每分钟输出

id=4 count=3 price=7208
id=8 count=2 price=10152
id=7 count=1 price=6928
id=5 count=1 price=3327
id=6 count=3 price=20483
id=0 count=2 price=9882
id=2 count=2 price=9191
id=3 count=2 price=8211
id=1 count=3 price=9906
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
Redis客户端
object RedisClient extends Serializable {

  val redisHost = "127.0.0.1"
  val redisPort = 6379
  val redisTimeout = 30000
  lazy val pool = new JedisPool(new GenericObjectPoolConfig(), redisHost, redisPort, redisTimeout)

  lazy val hook = new Thread {
    override def run = {
      println("Execute hook thread: " + this)
      pool.destroy()
    }
  }
  sys.addShutdownHook(hook.run)


  def main(args: Array[String]): Unit = {
    val dbIndex = 0

    val jedis = RedisClient.pool.getResource
    jedis.select(dbIndex)
    jedis.set("test", "1")
    println(jedis.get("test"))
    RedisClient.pool.returnResource(jedis)

  }


}
  • 1.
  • 2.
  • 3.
  • 4.
  • 5.
  • 6.
  • 7.
  • 8.
  • 9.
  • 10.
  • 11.
  • 12.
  • 13.
  • 14.
  • 15.
  • 16.
  • 17.
  • 18.
  • 19.
  • 20.
  • 21.
  • 22.
  • 23.
  • 24.
  • 25.
  • 26.
  • 27.
  • 28.
  • 29.
Redis结果

上一分钟商品销售额，有了这个数据就可以做成动态的图表展示时实交易额了

实战SparkStream+Kafka+Redis实时计算商品销售额_redis_03

每件商品总销售额

实战SparkStream+Kafka+Redis实时计算商品销售额_时实计算_04

总销售额，这就是天猫大屏上的1111亿了

实战SparkStream+Kafka+Redis实时计算商品销售额_kafka_05


原创作者: u_4937244 转载于: https://blog.51cto.com/u_4937244/11567160
# kafka # redis # 分布式 # 数据库 # 缓存
Logo
Kafka开源项目指南

Kafka开源项目指南提供详尽教程，助开发者掌握其架构、配置和使用，实现高效数据流管理和实时处理。它高性能、可扩展，适合日志收集和实时数据处理，通过持久化保障数据安全，是企业大数据生态系统的核心。

更多推荐

Kafka入门（一） 概述、部署与API的简单使用

Kafka概述、部署与API的简单使用

avatar Kafka开源项目指南

基于canal和kafka同步，实现binlog同步ElasticSearch

文章目录前言elasticsearch 安装canal安装canal-adapter 安装及配置mysql 安装zk及kafaka安装查看效果注意事项前言中间件版本elasticsearch7.5.2canal1.1.4client-adapter1.1.5-alpha-1zookeeper3.4.13kafka2.6.0mysql5.7.31elasticsearch 安装{"settings"

avatar Kafka开源项目指南

基于 Iceberg 的湖仓一体架构在 B 站的实践

背景在B站，每天都有PB级的数据注入到大数据平台，经过离线或实时的ETL建模后，提供给下游的分析、推荐及预测等场景使用。面对如此大规模的数据，如何高效低成本地满足下游数据的分析需求，一直是我们重点的工作方向。我们之前的数据处理流程基本上是这样的：采集端将客户端埋点、服务端埋点、日志、业务数据库等数据收集到HDFS、Kafka等存储系统中，然后通过Hive、Spark、Fl...

avatar Kafka开源项目指南
  • 浏览量 31
  • 收藏 0
  • 0

所有评论(0)

查看更多评论 

徐令予

@weixin_34650199
已为社区贡献1条内容