# Spark Streaming处理Kafka数据

$ spark-submit –class com.fgedu.streaming.KafkaStreamProcessor \

–master yarn \

–deploy-mode cluster \

–packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.2.1 \

/bigdata/app/jars/kafka-stream-processor-1.0.jar

23/07/25 10:35:00 INFO SparkContext: Running Spark version 3.2.1

23/07/25 10:35:01 INFO SparkContext: Submitted application: Kafka Stream Processor

23/07/25 10:35:02 INFO SparkContext: Created Spark context with app ID
application_1564034400001_0002

23/07/25 10:35:03 INFO StreamingContext: Streaming context started

23/07/25 10:35:05 INFO KafkaSource: Initial offsets: {fgedu_events-0=0}

23/07/25 10:35:10 INFO KafkaSource: Processing 3 records from offset 0 to 2 for partition
fgedu_events-0

23/07/25 10:35:10 INFO KafkaStreamProcessor: Processed event: e001, user: u001, type:
click

23/07/25 10:35:10 INFO KafkaStreamProcessor: Processed event: e002, user: u002, type: view

23/07/25 10:35:10 INFO KafkaStreamProcessor: Processed event: e003, user: u001, type:
purchase

# Spark SQL融合多源数据

$ spark-shell

scala> // 读取MySQL数据

scala> val mysqlDF = spark.read.format(“jdbc”).option(“url”,
“jdbc:mysql://192.168.1.100:3306/fgedudb”).option(“dbtable”, “fgedu_customers”).option(“user”,
“fgedu”).option(“password”, “fgedu123”).load()

mysqlDF: org.apache.spark.sql.DataFrame = [customer_id: string, name: string, email:
string, phone: string]

scala> // 读取MongoDB数据

scala> val mongoDF = spark.read.format(“mongo”).option(“uri”,
“mongodb://192.168.1.101:27017/fgedudb.orders”).load()

mongoDF: org.apache.spark.sql.DataFrame = [_id: struct, customerId: string, amount:
double, orderDate: timestamp, status: string]

scala> // 读取Kafka数据

scala> val kafkaDF = spark.read.format(“kafka”).option(“kafka.bootstrap.servers”,
“fgedu.net.cn:9092”).option(“subscribe”, “fgedu_events”).load()

kafkaDF: org.apache.spark.sql.DataFrame = [key: binary, value: binary, topic: string,
partition: int, offset: long, timestamp: timestamp, timestampType: int]

scala> // 数据融合分析

scala> val mergedDF = mysqlDF.join(mongoDF, mysqlDF(“customer_id”) ===
mongoDF(“customerId”), “left”).select(mysqlDF(“customer_id”), mysqlDF(“name”), mysqlDF(“email”),
mongoDF(“amount”), mongoDF(“orderDate”), mongoDF(“status”))

mergedDF: org.apache.spark.sql.DataFrame = [customer_id: string, name: string, email:
string, amount: double, orderDate: timestamp, status: string]

scala> mergedDF.show(5)

+———–+———-+——————+——+——————-+———+

|customer_id| name| email|amount| orderDate| status|

+———–+———-+——————+——+——————-+———+

| 101|John Smith|john@example.com|99.99|2023-07-25 10:00:00|completed|

| 102|Jane Doe|jane@example.com|199.99|2023-07-25 10:05:00|completed|

| 103|Bob Brown|bob@example.com|299.99|2023-07-25 10:10:00| pending|

| 104|Alice Lee|alice@example.com|399.99|2023-07-25 10:15:00|completed|

| 105|Tom Wang|tom@example.com|499.99|2023-07-25 10:20:00|cancelled|

+———–+———-+——————+——+——————-+———+

Part05-风哥经验总结与分享

5.1 多源数据融合最佳实践

• 数据标准化：建立统一的数据标准和规范，确保不同来源的数据能够正确融合
• 元数据管理：建立完善的元数据管理系统，记录数据来源、格式和转换规则
• 增量处理：对于大规模数据，采用增量处理方式，减少处理时间和资源消耗
• 数据质量监控：建立数据质量监控体系，及时发现和处理数据质量问题
• 弹性架构：设计弹性的架构，能够适应数据源的变化和业务需求的调整

5.2 常见问题与解决方案

问题	解决方案
数据格式不一致	建立数据转换层，统一数据格式
数据重复	使用去重算法，如基于唯一ID的去重
数据延迟	优化数据处理流程，使用实时处理技术
数据丢失	使用事务机制和数据校验，确保数据完整性
性能瓶颈	优化硬件资源，使用并行处理和缓存技术

更多视频教程www.fgedu.net.cn

5.3 性能优化建议

• 数据分区：根据数据特点进行合理分区，提高查询性能
• 缓存策略：使用缓存技术，减少重复计算和数据读取
• 并行处理：充分利用集群资源，进行并行处理
• 压缩存储：使用数据压缩技术，减少存储空间和网络传输时间
• 索引优化：为常用查询建立适当的索引
• 资源调度：合理调度集群资源，避免资源争用

学习交流加群风哥微信: itpux-com

更多学习教程公众号风哥教程itpux_com
from bigdata视频:www.itpux.com