ToB企服应用市场:ToB评测及商务社交产业平台

标题: 大数据-156 Apache Druid 案例实战 Scala Kafka 订单统计 [打印本页]

作者: 刘俊凯 时间: 2024-10-1 17:54
标题: 大数据-156 Apache Druid 案例实战 Scala Kafka 订单统计
点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

Hadoop（已更完）
HDFS（已更完）
MapReduce（已更完）
Hive（已更完）
Flume（已更完）
Sqoop（已更完）
Zookeeper（已更完）
HBase（已更完）
Redis （已更完）
Kafka（已更完）
Spark（已更完）
Flink（已更完）
ClickHouse（已更完）
Kudu（已更完）
Druid（正在更新…）

章节内容

上节我们完成了如下的内容：

Apache Druid 数据存储
Apache Druid 数据分区
索引服务
压缩机制
数据聚合

整体流程

Kafka 数据源: Kafka 是一个分布式流处理平台，负责接收、存储并传输数据。它支持从各类应用、日记、传感器等装备收罗实时数据，将数据分别为多个主题（Topic），并将消息分发给消耗者。在这个案例中，Kafka 是 Druid 的数据源。
Kafka Producer: 数据生产者（Producer）负责将数据发送到 Kafka 的主题中。例如，应用步伐可以向 Kafka 写入日记、用户举动数据、传感器数据等。每条消息可以是 JSON、Avro 等格式的数据记载。
Druid Kafka Ingestion: Druid 提供了对 Kafka 的原生支持。通过 Kafka Indexing Service，Druid 可以持续从 Kafka 的某个主题中消耗数据，实时地将这些数据摄取到 Druid 中。摄取过程中，Druid 会将数据拆解为小的段（Segment），并将这些段存储在 Druid 集群的深度存储中（如 HDFS、S3 等）。
实时数据摄取和索引: Druid 的 Kafka 摄取任务会监听 Kafka 的分区，按照流数据的到达顺序消耗数据，并在内部创建索引。这些索引结构化存储了数据，并通过分片和分区机制，保证了查询的高效性和程度扩展能力。
Druid 查询层: Druid 提供了非常强大的查询能力，可以通过 SQL 查询方式进行交互，也支持多维查询、聚合查询等。这些查询可以是低延迟的实时查询，也可以对历史数据进行复杂的分析。用户通过 Druid 查询接口或 BI 工具（如 Apache Superset、Tableau 等）向集群发送查询。
Kafka 消耗者 Offset 管理: Druid 使用 Kafka 消耗者模型，实时消耗消息并管理 Offset（偏移量），确保数据不丢失或重复摄取。Offset 会被定期提交到 Kafka 中，保证纵然任务重启，摄取进度也能从上一次的位置继续。
持久化和数据存储: 数据在颠末摄取和索引后，Druid 会定期将数据段（Segment）持久化到深度存储中，并对旧数据进行归并和压缩，减少存储空间的占用。Druid 的集群架构支持分布式存储和查询，并能根据数据规模进行自动扩展。

案例假设

假设我们在构建一个用户举动分析体系，通过 Kafka 收罗用户点击日记，并通过 Druid 实时分析用户举动。

Kafka 数据生产: 电商平台的应用步伐会将每次用户点击产生的日记记载（例如点击商品、页面欣赏等）发送到 Kafka 中的 user-clicks 主题。每条记载都包含用户ID、商品ID、时间戳、页面信息等。
Druid 数据摄取: 设置 Druid 的 Kafka Indexing Service，从 user-clicks 主题消耗数据。数据会实时流入 Druid 中，Druid 将数据按照时间范围切分为段，并存储到其深度存储中。
实时数据查询与分析: 业务方可以通过 SQL 查询或多维查询接口，实时分析用户的点击举动。查询的例子可能是统计每个小时的页面欣赏量、分析差别商品的受接待程度等。这些查询可以直接反映用户的当前举动，资助业务方做出快速决定。
可视化和报表: Druid 的查询结果可以通过 Apache Superset 等工具进行可视化展示，创建实时仪表盘，展示用户举动的各种关键指标。数据分析师和运营人员可以在可视化平台上直观地看到当前体系的运营状态。

需求分析

场景分析

数据量大，需要在这些数据中根据业务需要灵活查询
实时性要求高
数据实时的推过来，要在秒级对数据进行分析并查询出结果

数据描述

{"ts":1607499629841,"orderId":"1009388","userId":"807134","orderStatusId":1,"orderStatus":"已支付","payModeId":0,"payMode":"微信","payment":"933.90","products":
[{"productId":"102163","productName":"贝合xxx+粉","price":18.7,"productNum":3,"categoryid":"10360","catname1":"厨卫清洁、纸制用品","catname2":"生活日用","catname3":"浴室用品"},{"productId":"100349","productName":"COxxx0C","price":877.8,"productNum":1,"categoryid":"10302","catname1":"母婴、玩具乐器","catname2":"西洋弦乐器","catname3":"吉他"}]}

复制代码

ts 生意业务时间
orderId 订单编号
userId 用户id
orderStatusId 订单状态Id
orderStatus 订单状态 0-11：未付出,已付出,发货中,已发货,发货失败,已退款,已关单,订单过期,订单已失效,产品已失效,代付拒绝,付出中
payModelId 付出方式id
payMode 付出方式：0-6：微信,付出宝,信用卡,银联,货到付款,现金,其他
payment：付出金额
products：购买商品（一个订单可能包含多个商品，这里是嵌套结构）
productId 商品Id
productName 商品名称
price 单价
productNum 购买数量
categoryid 商品分类Id
catname1 商品一级分类名称
catname2 商品二级分类名称
catname3 商品三级分类名称

以上的嵌套的json数据格式，Druid不利益理，需要对数据进行预处理，将数据拉平，处理后的数据格式：

{"ts":1607499629841,"orderId":"1009388","userId":"807134","orderStatusId":1,"orderStatus":"已支付","payModeId":0,"payMode":"微信","payment":"933.90","product":
{"productId":"102163","productName":"贝合xxx+粉","price":18.7,"productNum":3,"categoryid":"10360","catname1":"厨卫清洁、纸制用品","catname2":"生活日用","catname3":"浴室用品"}}
{"ts":1607499629841,"orderId":"1009388","userId":"807134","orderStatusId":1,"orderStatus":"已支付","payModeId":0,"payMode":"微信","payment":"933.90","product":
{"productId":"100349","productName":"COxxx0C","price":877.8,"productNum":1,"categoryid":"10302","catname1":"母婴、玩具乐器","catname2":"西洋弦乐器","catname3":"吉他"}}

复制代码

Kafka生产者

很久没用Scala了，用Scala写一个：

package icu.wzk.kafka
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer
import java.util.Properties
import scala.io.BufferedSource
object KafkaProducerForDruid {
def main(args: Array[String]): Unit = {
val brokers = "h121.wzk.icu:9092"
val topic = "druid2"
val prop = new Properties()
prop.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
prop.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
prop.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer])
val producer = new KafkaProducer[String, String](prop);
val source: BufferedSource = scala.io.Source.fromFile("orders1.json")
val iter: Iterator[String] = source.getLines();
iter.foreach {
line => val msg = new ProducerRecord[String, String](topic, line);
producer.send(msg)
println(msg)
Thread.sleep(10)
}
producer.close()
source.close()
}
}