企业版SparkDatabricks+企业版KafkaConfluent联合高效挖掘数据价值

前提条件

• 已注册阿里云账号&＃xff0c;详情请参见阿里云账号注册流程
• 已开通 Databricks 数据洞察服务
• 已开通 OSS 对象存储服务
• 已开通 Confluent 流数据服务

创建Databricks集群 & Confluent集群

1. 登录Confluent管理控制台&＃xff0c;创建Confluent集群&＃xff0c;并开启公网服务
2. 登录Databricks管理控制台&＃xff0c;创建Databricks集群

Databricks Worker节点公网访问

Databricks的worker节点暂时不支持公网访问&＃xff0c;为了能访问Confluent的公网地址&＃xff0c;请联系Databricks的开发人员添加NAT网关。

案例&＃xff1a;出租车数据入湖及分析

出租车和网约车在每天的运行中持续产生行驶轨迹和交易数据&＃xff0c;这些数据对于车辆调度&＃xff0c;流量预测&＃xff0c;安全监控等场景有着极大的价值。

本案例中我们使用纽约市的出租车数据来模拟网约车数据从产生&＃xff0c;发布到流数据服务Confluent&＃xff0c;通过Databricks Structured Streaming进行实时数据处理&＃xff0c;并存储到LakeHouse的整个流程。数据存储到LakeHouse后&＃xff0c;我们使用spark和spark sql对数据进行分析&＃xff0c;并使用Spark的MLlib进行机器学习训练。

前置准备&＃xff1a;

1. 创建topic&＃xff1a;
   登录Confluent的control center&＃xff0c;在左侧选中Topics&＃xff0c;点击Add a topic按钮&＃xff0c;创建一个名为nyc_taxi_data的topic&＃xff0c;将partition设置为3&＃xff0c;其他配置保持默认。
2. 创建OSS bucket&＃xff1a;
   在和Databricks同一Region的OSS中&＃xff0c;创建bucket&＃xff0c;bucket命名为&＃xff1a;databricks-confluent-integration进入到Bucket列表页&＃xff0c;点击创建bucket按钮创建好bucket之后&＃xff0c;在该bucket创建目录&＃xff1a;checkpoint_dir和data/nyc_taxi_data两个目录
3. 收集url&＃xff0c;用户名&＃xff0c;密码&＃xff0c;路径等以便后续使用
   1. confluent集群ID&＃xff1a;在csp的管控界面&＃xff0c;集群详情页获取
   2. Confluent Control Center的用户名和密码
   3. 路径&＃xff1a;

• Databricks Structured Streaming的checkpoint存储目录
• 采集的数据的存储目录

下面是我们后续会使用到的一些变量&＃xff1a;

# 集群管控界面获取
confluent_cluster_id &＃61; "your_confluent_cluster_id"
# 使用confluent集群ID拼接得到
confluent_server &＃61; "rb-{confluent_cluster_id}.csp.aliyuncs.com:9092"
control_center_username &＃61; "your_confluent_control_center_username"
control_center_password &＃61; "your_confluent_control_center_password"topic &＃61; "nyc_taxi_data"checkpoint_location &＃61; "oss://databricks-confluent-integration/checkpoint_dir"
taxi_data_delta_lake &＃61; "oss://databricks-confluent-integration/data/nyc_taxi_data"

数据的产生

在本案例中&＃xff0c;我们使用Kaggle上的NYC出租车数据集来模拟数据产生。

• 我们先安装confluent的python客户端&＃xff0c;其他语言的客户端参考confluent官网

pip install confluent_kafka

• 构造用于创建Kafka Producer的基础信息&＃xff0c;如&＃xff1a;bootstrap-server&＃xff0c;control center的username&＃xff0c;password等

conf &＃61; {&＃39;bootstrap.servers&＃39;: confluent_server,&＃39;key.serializer&＃39;: StringSerializer(&＃39;utf_8&＃39;),&＃39;value.serializer&＃39;: StringSerializer(&＃39;utf_8&＃39;),&＃39;client.id&＃39;: socket.gethostname(),&＃39;security.protocol&＃39;: &＃39;SASL_SSL&＃39;,&＃39;sasl.mechanism&＃39;: &＃39;PLAIN&＃39;,&＃39;sasl.username&＃39;: control_center_username,&＃39;sasl.password&＃39;: control_center_password
}

• 创建Producer&＃xff1a;

producer &＃61; Producer(conf)

• 向Kafka中发送消息&＃xff08;模拟数据的产生&＃xff09;&＃xff1a;

with open("/Path/To/train.csv", "rt") as f:float_field &＃61; [&＃39;fare_amount&＃39;, &＃39;pickup_longitude&＃39;, &＃39;pickup_latitude&＃39;, &＃39;dropoff_longitude&＃39;, &＃39;dropoff_latitude&＃39;]for row in reader:i &＃43;&＃61; 1try:for field in float_field:row[field] &＃61; float(row[field])row[&＃39;passenger_count&＃39;] &＃61; int(row[&＃39;passenger_count&＃39;])producer.produce(topic&＃61;topic, value&＃61;json.dumps(row))if i % 1000 &＃61;&＃61; 0:producer.flush()if i &＃61;&＃61; 200000:breakexcept ValueError: # discard null/NAN datacontinue

Kafka中的partition和offset

在使用spark读取Kafka中的数据之前&＃xff0c;我们回顾一下Kafka中的概念&＃xff1a;partition和offset

• partition&＃xff1a;kafka为了能并行进行数据的写入&＃xff0c;将每个topic的数据分为多个partition&＃xff0c;每个partition由一个Broker负责&＃xff0c;向partition写入数据时&＃xff0c;负责该partition的Broker将消息复制给它的follower
• offset&＃xff1a;Kafka会为每条写入partition里的消息进行编号&＃xff0c;消息的编号即为offset

我们在读取Kafka中的数据时&＃xff0c;需要指定我们想要读取的数据&＃xff0c;该指定需要从两个维度&＃xff1a;partition的维度 &＃43; offset的维度。

• Earliest&＃xff1a;从每个partition的offset 0开始读取和加载
• Latest&＃xff1a;从每个partition最新的数据开始读取
• 自定义&＃xff1a;指定每个partition的开始offset和结束offset
  • 读取topic1 partition 0 offset 23和partition 0 offset -2之后的数据&＃xff1a;"""{"topic1":{"0":23,"1":-2}}"""

除了指定start offset&＃xff0c;我们还可以通过endingOffsets参数指定读取到什么位置为止。

将数据存储到LakeHouse&＃xff1a;Spark集成Confluent

理解上述概念后&＃xff0c;Databricks和Confluent的集成非常简单&＃xff0c;只需要对spark session的readStream参数进行简单的设置就可以将Kafka中的实时流数据转换为Spark中的Dataframe&＃xff1a;

lines &＃61; (spark.readStream# 指定数据源: kafka.format("kafka")# 指定kafka bootstrap server的URL.option("kafka.bootstrap.servers", confluent_server)# 指定订阅的topic.option("subscribe", topic)# 指定想要读取的数据的offset&＃xff0c;earliest表示从每个partition的起始点开始读取.option("startingOffsets", "earliest")# 指定认证协议.option("kafka.security.protocol", "SASL_SSL").option("kafka.sasl.mechanism", "PLAIN")# 指定confluent的用户名和密码.option("kafka.sasl.jaas.config",f"""org.apache.kafka.common.security.plain.PlainLoginModule required username&＃61;"{control_center_username}" password&＃61;"{control_center_password}";""").load())

从kafka中读取的数据格式如下&＃xff1a;

root|-- key: binary (nullable &＃61; true)|-- value: binary (nullable &＃61; true)|-- topic: string (nullable &＃61; true)|-- partition: integer (nullable &＃61; true)|-- offset: long (nullable &＃61; true)|-- timestamp: timestamp (nullable &＃61; true)|-- timestampType: integer (nullable &＃61; true)

由于key和value都是binary格式的&＃xff0c;我们需要将value&＃xff08;json&＃xff09;由binary转换为string格式&＃xff0c;并定义schema&＃xff0c;提取出Json中的数据&＃xff0c;并转换为对应的格式&＃xff1a;

schema &＃61; (StructType().add(&＃39;key&＃39;, TimestampType()).add(&＃39;fare_amount&＃39;, FloatType()).add(&＃39;pickup_datetime&＃39;, TimestampType()).add(&＃39;pickup_longitude&＃39;, FloatType()).add(&＃39;pickup_latitude&＃39;, FloatType()).add(&＃39;dropoff_longitude&＃39;, FloatType()).add(&＃39;dropoff_latitude&＃39;, FloatType()).add(&＃39;passenger_count&＃39;, IntegerType()))# 将json中的列提取出来
lines &＃61; (lines.withColumn(&＃39;data&＃39;, from_json(col(&＃39;value&＃39;).cast(&＃39;string&＃39;), # binary 转 stringschema)) # 解析为schema.select(col(&＃39;data.*&＃39;))) # select value中的所有列

过滤掉错误&＃xff0c;为空&＃xff0c;NaN的数据&＃xff1a;

lines &＃61; (lines.filter(col(&＃39;pickup_longitude&＃39;) !&＃61; 0).filter(col(&＃39;pickup_latitude&＃39;) !&＃61; 0).filter(col(&＃39;dropoff_longitude&＃39;) !&＃61; 0).filter(col(&＃39;dropoff_latitude&＃39;) !&＃61; 0).filter(col(&＃39;fare_amount&＃39;) !&＃61; 0).filter(col(&＃39;passenger_count&＃39;) !&＃61; 0))

最后&＃xff0c;我们将解析出来的数据输出到LakeHouse中&＃xff0c;以进行后续的分析和机器学习模型训练&＃xff1a;

# lakehouse 的存储格式为 delta
query &＃61; (lines.writeStream.format(&＃39;delta&＃39;).option(&＃39;checkpointLocation&＃39;, checkpoint_location).option(&＃39;path&＃39;, taxi_data_delta_lake).start())
# 执行job&＃xff0c;直到出现异常&＃xff08;如果只想执行该Job一段时间&＃xff0c;可以指定timeout参数&＃xff09;
query.awaitTermination()

数据分析

我们先将LakeHouse中的数据使用Spark加载进来&＃xff1a;

然后&＃xff0c;我们对该Dataframe创建一个Table View&＃xff0c;并探索fare_amount的分布&＃xff1a;

可以看到fare_amount的最小值是负数&＃xff0c;这显然是一条错误的数据&＃xff0c;我们将这些错误的数据过滤&＃xff0c;并探索fare_amount的分布&＃xff1a;

然后我们探索价格和年份&＃xff0c;月份&＃xff0c;星期&＃xff0c;打车时间的关系&＃xff1a;

从上面可以看出两点&＃xff1a;

• 出租车的价格和年份有很大关系&＃xff0c;从09年到15年呈不断增长的态势
• 在中午和凌晨打车比上午和下午打车更贵一些。

我们再进一步探索价格和乘客数量的关系&＃xff1a;

此外&＃xff0c;出租车价格的另一个影响因素就是距离&＃xff0c;这里我们借助python的geopy包和Spark的UDF来计算给定两个位置的距离&＃xff0c;然后再分析费用和距离的关系。

经纬度的范围为[-90, 90]&＃xff0c;因此&＃xff0c;我们第一步是清除错误的数据&＃xff1a;

然后&＃xff0c;我们增加一列数据&＃xff1a;出租车行驶的距离&＃xff0c;并将距离离散化&＃xff0c;进行后续的分析&＃xff1a;

统计打车距离的分布&＃xff1a;

从上图可以看出&＃xff1a;打车距离分布在区间[0, 15]miles内&＃xff0c;我们继续统计在该区间内&＃xff0c;打车价格和打车距离的关系&＃xff1a;

如上图所示&＃xff1a;打车价格和打车距离呈现出线性增长的趋势。

机器学习建模

在上一小节的数据分析中&＃xff0c;我们已经提取了和出租车相关联的一些特征&＃xff0c;根据这些特征&＃xff0c;我们建立一个简单的线性回归模型&＃xff1a;

打车费用 ~ (年份&＃xff0c;打车时间&＃xff0c;乘客数&＃xff0c;距离)

先将特征和目标值提取出来&＃xff1a;

对特征做归一化&＃xff1a;

分割训练集和测试集&＃xff1a;

建立线性回归模型进行训练&＃xff1a;

训练结果统计&＃xff1a;

使用Evaluator对模型进行评价&＃xff1a;

总结

我们在本文中介绍了如何使用阿里云的Confluent Cloud和Databricks来构建您的数据流和LakeHouse&＃xff0c;并介绍了如何使用Databricks提供的能力来挖掘数据价值&＃xff0c;使用Spark MLlib构建您的机器学习模型。有了Confluent Cloud和Databricks&＃xff0c;您可以轻松实现数据入湖&＃xff0c;及时在最新的数据上进行探索&＃xff0c;挖掘您的数据价值。欢迎您试用阿里云Confluent和Databricks。

原文链接

本文为阿里云原创内容&＃xff0c;未经允许不得转载。