尧图网站建设 
简介这份资源是面向大数据方向学习者与开发者的Spark 2.X新闻话题实时统计分析项目实战包适合计算机相关专业学生、教师及企业员工用于毕业设计、课程设计、项目立项演示或技能进阶。项目已通过导师评审并获95分答辩成绩代码经测试可正常运行。压缩包共499个文件约6.2MB以400个xml配置、47个class字节码、18个jar依赖包为主另含9个scala源码、6个properties配置、4个js与2个html页面、2个java文件及iml、zip、txt等辅助资料覆盖Structured Streaming对接Kafka、JDBCSink写入、Weblog服务与MySQL连接池等核心模块。已有63人学习下载。读者可获得完整项目源码、可运行环境配置、实时统计流程实现思路与排错参考既能直接用于毕设课设也可在此基础上二次开发扩展功能。1. 从一份 Spark2.X 新闻话题实时统计源码包说起它到底能跑出什么新闻资讯类业务有个绕不开的诉求一条热点从出现到爆发往往只有几十分钟等 T1 离线跑完话题早就凉了。这份基于 Spark2.X 的新闻话题实时统计分析项目解决的正是这个窗口期问题——用 Structured Streaming 消费 Kafka 里的新闻流做窗口聚合、话题热度排序再把结果落到 MySQL 供前端展示。压缩包里既有 Scala 源码JDBCSink、StreamingKafka8/10、StructuredStreamingKafka、WeblogService 等也有编译后的 class 文件和一份实战文档属于那种「拿来就能对着跑、跑完能改」的完整工程。它适合三类人做大数据课程设计或毕设的学生需要一套能讲清架构又能演示的完整链路刚转实时计算方向的工程师想找一个比官方 wordcount 更接近生产的例子还有要快速搭演示原型的开发者。下面按「资源里有什么 → 环境怎么搭 → 代码怎么跑 → 坑在哪 → 怎么改」的顺序拆开讲。2. 拆开压缩包先看结构Spark2.X 实时链路的模块划分与选型理由拿到一个源码包我习惯先不急着跑而是把目录和类名过一遍判断作者的技术选型和链路设计。这份资源的类名信息量很大基本能反推出整套架构。2.1 从类名反推架构Kafka Structured Streaming MySQL 三段式包里出现的核心类可以分成三组。第一组是数据接入层StreamingKafka8和StreamingKafka10两个版本并存说明作者兼容了 Kafka 0.8 和 0.10 两代消费者 API这是 Spark2.X 时代的典型做法——Spark 2.0 到 2.2 主要对接 Kafka 0.8 的 Receiver/Direct 模式2.3 之后才稳定支持 Kafka 0.10 的 Structured Streaming 直连。第二组是计算层StructuredStreamingKafka及其$Weblog变体走的是 Structured Streaming 的 DataFrame/Dataset 编程模型而不是老的 DStream。第三组是输出层JDBCSink负责把结果写进 MySQLMySqlPool管连接池WeblogService封装业务查询。这个三段式Kafka 接入 → Structured Streaming 计算 → JDBC 落地是新闻话题实时统计最主流的骨架。选 Structured Streaming 而不是 DStream 的理由很实在它基于 Catalyst 优化器窗口聚合写起来是 SQL 风格的增量输出模式update/append/complete能直接控制结果怎么吐给下游而 DStream 要自己维护状态、自己处理乱序代码量和出错概率都高一个量级。2.2 环境与依赖版本对齐别让版本错配毁掉一整天Spark2.X 项目最怕的就是版本错配。这份资源没有在正文里写死具体小版本但根据类名和 API 用法可以确定一个能跑通的组合区间。下面这张表是我按经验给的对照实际以你本地已有环境为准能对齐就对齐对不齐优先改 Spark 而不是改代码逻辑。组件建议版本说明JDK1.8Spark2.X 对 JDK9 支持差别用高版本Scala2.11.xSpark2.X 默认编译版本2.12 需重新编译Spark2.2.0 ~ 2.4.x2.3 以下用 Kafka0.8 那套类2.3 用 0.10Kafka0.10.x与 StreamingKafka10 对应MySQL5.78.0 驱动类名和时区参数有变化构建工具Maven 3.5依赖坐标按 pom 走环境变量这块SPARK_HOME、JAVA_HOME、SCALA_HOME三个必须配好spark-submit才能找到对应 jar。我一般会在spark-env.sh里显式指定SPARK_EXECUTOR_MEMORY和SPARK_DRIVER_MEMORY因为实时任务默认内存偏小窗口聚合一堆积就容易 OOM。2.3 导入工程与依赖检查先让 IDE 不报红把源码导入 IntelliJ IDEA 后第一步是确认 Maven 依赖能全部拉下来。Structured Streaming 对接 Kafka 需要spark-sql-kafka-0-10这个包JDBC 落地需要mysql-connector-java。如果 pom 里用的是 provided 作用域本地跑要临时改成 compile否则运行时报 ClassNotFound。# 检查关键依赖是否在本地仓库 ls ~/.m2/repository/org/apache/spark/spark-sql-kafka-0-10_2.11/ ls ~/.m2/repository/mysql/mysql-connector-java/ # 如果缺失手动拉一次版本按 pom 里的来 mvn dependency:resolve -DincludeArtifactIdsspark-sql-kafka-0-10_2.11,mysql-connector-java上面第一条命令确认 Kafka 连接器和 MySQL 驱动是否已下载第二条在依赖缺失时触发 Maven 重新解析。参数-DincludeArtifactIds用来只拉指定构件避免全量下载拖时间。如果拉取一直失败先看settings.xml里的镜像地址是否可达这是国内环境最常见的卡点。3. 把实时链路跑起来Kafka 生产、Structured Streaming 消费与 JDBC 落地环境就绪后真正的工作量在「让数据从 Kafka 流进来、算完、写出去」这条链路上。这一章按数据流向拆成三步每步都给可抄的命令和代码骨架。3.1 启动 Kafka 并灌入新闻测试数据实时任务没有数据源就是空转。先起一个单节点 Kafka测试够用建 topic再用控制台生产者灌几条模拟新闻。# 启动 ZooKeeper 和 Kafka版本按你本地0.10 时代还需 ZK bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties # 建 topic3 分区 1 副本测试环境够用 bin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 3 --topic news_topic # 灌入模拟新闻字段用逗号分隔话题,标题,时间戳 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic news_topic--partitions 3是为了让后续 Spark 消费时能并行起 3 个 task单分区的话并行度上不去窗口聚合会成瓶颈。生产消息时每条一行格式要和消费端解析逻辑对齐——这是后面最容易翻车的地方先记住字段顺序。3.2 Structured Streaming 消费 Kafka 的核心代码骨架消费端的核心是把 Kafka 的 value 字段解析成结构化列再按窗口聚合。下面这段是这类项目的通用骨架字段名和窗口时长按你实际业务改。// 从 Kafka 读取流注意 bootstrapServers 和 subscribe 参数 val raw spark.readStream .format(kafka) .option(kafka.bootstrap.servers, localhost:9092) .option(subscribe, news_topic) .option(startingOffsets, latest) .load() // Kafka 的 value 是二进制先转字符串再按逗号切分 val parsed raw.selectExpr(CAST(value AS STRING) as line) .select( split($line, ,)(0).as(topic_name), split($line, ,)(1).as(title), to_timestamp(split($line, ,)(2)).as(event_time) ) // 按 1 分钟滚动窗口统计每个话题的出现次数 val windowed parsed .withWatermark(event_time, 2 minutes) .groupBy(window($event_time, 1 minute), $topic_name) .count() // 输出到 MySQLforeachBatch 里调 JDBCSink val query windowed.writeStream .outputMode(update) .foreachBatch { (batchDF: Dataset[Row], batchId: Long) JDBCSink.write(batchDF, batchId) } .trigger(ProcessingTime(10 seconds)) .start()逐段说明startingOffsets设成latest表示只消费启动后的新消息调试时想重放历史数据就改成earliestwithWatermark是处理乱序的关键设 2 分钟意味着迟到超过 2 分钟的数据会被丢弃这个值要按你数据源的延迟分布调outputMode(update)让每个批次只输出有更新的窗口比complete省资源trigger设 10 秒是微批的间隔太短会增加调度开销太长则实时性下降。3.3 JDBCSink 写 MySQL连接池与幂等写入JDBCSink和MySqlPool这两个类解决的是「每个批次都要写库不能每次新建连接」的问题。连接池复用连接写入时用INSERT ... ON DUPLICATE KEY UPDATE保证同一窗口重复输出不会产生脏数据。object JDBCSink { def write(df: Dataset[Row], batchId: Long): Unit { df.foreachPartition { partition val conn MySqlPool.getConnection() // 从池里取不是新建 val stmt conn.prepareStatement( INSERT INTO topic_stat(window_start, topic_name, cnt) VALUES(?,?,?) ON DUPLICATE KEY UPDATE cnt VALUES(cnt)) partition.foreach { row stmt.setTimestamp(1, row.getTimestamp(0)) stmt.setString(2, row.getString(1)) stmt.setLong(3, row.getLong(2)) stmt.addBatch() } stmt.executeBatch() MySqlPool.release(conn) // 用完归还别 close } } }foreachPartition而不是foreach是因为前者每个分区只建一次连接后者每行都建性能差几十倍。ON DUPLICATE KEY UPDATE要求 MySQL 表上对(window_start, topic_name)建唯一索引否则幂等不生效。MySqlPool.release归还连接而不是关闭池子大小在MySqlPool初始化时配一般设 10 到 20 够用。3.4 提交任务与验证结果代码打包后用spark-submit提交。本地模式先验证逻辑再上集群。# 本地模式跑通逻辑 spark-submit --class com.xxx.StructuredStreamingKafka \ --master local[4] \ --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0 \ target/news-stat.jar # 验证 MySQL 里有没有数据 mysql -uroot -p -e SELECT * FROM news_db.topic_stat ORDER BY window_start DESC LIMIT 10;--master local[4]用 4 个线程模拟并行--packages在线拉取 Kafka 连接器生产环境建议打进 jar 或用--jars指定本地路径避免每次联网。验证时如果 MySQL 表是空的先看 Spark 控制台有没有批次输出再查 Kafka 里是否真有数据逐段排查。4. 避坑与排查这套 Spark2.X 实时链路最容易翻车的五个地方实时任务的坑大多不在业务逻辑而在版本、序列化、状态和资源这些「环境层」。下面五条是我拆这类项目时反复遇到的每条按现象、原因、解决写。4.1 现象任务启动即报 NoSuchMethodError 或 ClassNotFoundException原因基本是版本错配。Spark2.X 的spark-sql-kafka-0-10必须和 Spark 主版本严格对应2.4 的 Spark 配 2.3 的连接器就会在运行时找不到方法。另外 Scala 2.11 和 2.12 的构件后缀不同拉错后缀直接类加载失败。解决先spark-submit --version确认 Spark 版本再按_2.11或_2.12后缀选连接器版本号与 Spark 保持一致。pom 里把spark-sql-kafka-0-10的版本写成和spark-core同一个属性变量避免手改漏改。4.2 现象窗口聚合结果一直不输出或者输出后又被覆盖原因通常是 watermark 和输出模式配合错了。update模式下窗口只有在 watermark 超过窗口结束时间后才会被最终确定如果 watermark 设得比数据延迟还小迟到数据被丢结果就偏。complete模式则会每次输出全量窗口数据量大时直接把 driver 内存打满。解决先统计数据源的实际延迟分布watermark 设成 P99 延迟再加一点余量。调试阶段用complete看全貌上线切update。如果结果被覆盖检查 MySQL 的唯一索引是否建对。4.3 现象MySQL 连接数暴涨最后报 Too many connections原因在JDBCSink里用了foreach而不是foreachPartition或者用完连接没归还而是close了池里的连接。前者每行建一次连接后者让池子以为连接还在、实际已断反复新建。解决统一用foreachPartition连接从池取、用完release归还。池大小按分区数乘以并发批次估算别盲目设大。MySQL 侧max_connections也要留余量。4.4 现象Kafka 消费 lag 持续增长任务越跑越慢原因是消费并行度和 Kafka 分区数不匹配。topic 3 个分区Spark 只起了 1 个 task吞吐上不去。或者trigger间隔太短批次还没处理完下一个就来了任务排队堆积。解决让 Spark 消费的并行度等于分区数Structured Streaming 默认按分区并行确认没被repartition(1)之类操作破坏。trigger间隔按单批次处理耗时来定一般设成处理耗时的 1.5 到 2 倍。4.5 现象本地跑通上集群后报序列化异常原因是JDBCSink或MySqlPool里引用了不可序列化的对象比如直接持有 Connection 的成员变量driver 往 executor 分发任务时序列化失败。解决把连接相关逻辑全部收进foreachPartition内部对象在 executor 侧创建不跨节点传递。所有要在闭包里用的外部变量确认实现了Serializable。5. 在源码基础上做二次开发换数据源、调窗口、加指标的具体手法跑通只是起点这份资源真正的价值在于它是个可改的骨架。下面说三个我实际改过的方向都是新闻话题统计场景下最常被要求加的功能。5.1 换数据源从 Kafka 切到 Socket 或文件流调试阶段不想起 Kafka可以把readStream.format(kafka)换成 socket 源代码改动很小。// 用 socket 源替代 Kafka方便本地调试 val raw spark.readStream .format(socket) .option(host, localhost) .option(port, 9999) .load() // 后续解析逻辑不变只是 value 列名从 kafka 的 value 变成 socket 的 valuesocket 源只有单分区、不支持 offset 管理只适合验证逻辑别用在生产。文件流则适合回放历史新闻做回归测试把format换成text或json指向一个目录Spark 会自动监听新文件。5.2 调窗口与指标从「计数」扩展到「热度分」原始代码只统计话题出现次数实际业务往往要一个加权热度分。改法是在groupBy后加聚合表达式把不同字段按权重算进去。指标计算方式权重建议出现次数count0.5独立来源数countDistinct(source)0.3时间衰减按窗口距当前时间加权0.2窗口时长也要按业务调突发新闻用 1 分钟滚动窗口话题趋势用 5 到 10 分钟滑动窗口。滑动窗口的slideDuration小于windowDuration时同一数据会进多个窗口计算量成倍增加资源要提前估。5.3 加状态与去重用 dropDuplicates 处理重复新闻新闻源经常重复推送同一条直接计数会虚高。Structured Streaming 支持在流上做去重但要注意它会维护状态状态大小随时间增长。// 按标题去重配合 watermark 限制状态保留时间 val deduped parsed .withWatermark(event_time, 10 minutes) .dropDuplicates(title)dropDuplicates必须配合 watermark 用否则状态无限增长跑几天就 OOM。watermark 设 10 分钟意味着只对 10 分钟内的重复做去重更早的状态被清理。这个取舍要看业务对重复的容忍度。5.4 验证改动是否生效三个必看的观测点改完代码别急着上线先看三个地方。一是 Spark UI 的 Streaming 页确认批次处理耗时稳定、没有堆积二是 MySQL 里的结果表确认窗口边界和计数符合预期三是 Kafka 的 consumer lag确认消费跟得上生产。这三个点任何一个异常都说明改动引入了新问题。我自己的习惯是每次改完窗口或 watermark都先用一小批带时间戳的测试数据跑一遍手动核对窗口划分对不对再放真实流量。从那以后我每次调实时任务的窗口参数都强制走一遍「小数据验证 → 看 UI → 查落库」这三步省下过好几次半夜爬起来排查的功夫。希望帮到你。本文还有配套的精品资源点击获取 Spark2.X新闻话题实时统计：Structured Streaming+Kafka+MySQL源码实战
简介这份资源是面向大数据方向学习者与开发者的Spark 2.X新闻话题实时统计分析项目实战包适合计算机相关专业学生、教师及企业员工用于毕业设计、课程设计、项目立项演示或技能进阶。项目已通过导师评审并获95分答辩成绩代码经测试可正常运行。压缩包共499个文件约6.2MB以400个xml配置、47个class字节码、18个jar依赖包为主另含9个scala源码、6个properties配置、4个js与2个html页面、2个java文件及iml、zip、txt等辅助资料覆盖Structured Streaming对接Kafka、JDBCSink写入、Weblog服务与MySQL连接池等核心模块。已有63人学习下载。读者可获得完整项目源码、可运行环境配置、实时统计流程实现思路与排错参考既能直接用于毕设课设也可在此基础上二次开发扩展功能。1. 从一份 Spark2.X 新闻话题实时统计源码包说起它到底能跑出什么新闻资讯类业务有个绕不开的诉求一条热点从出现到爆发往往只有几十分钟等 T1 离线跑完话题早就凉了。这份基于 Spark2.X 的新闻话题实时统计分析项目解决的正是这个窗口期问题——用 Structured Streaming 消费 Kafka 里的新闻流做窗口聚合、话题热度排序再把结果落到 MySQL 供前端展示。压缩包里既有 Scala 源码JDBCSink、StreamingKafka8/10、StructuredStreamingKafka、WeblogService 等也有编译后的 class 文件和一份实战文档属于那种「拿来就能对着跑、跑完能改」的完整工程。它适合三类人做大数据课程设计或毕设的学生需要一套能讲清架构又能演示的完整链路刚转实时计算方向的工程师想找一个比官方 wordcount 更接近生产的例子还有要快速搭演示原型的开发者。下面按「资源里有什么 → 环境怎么搭 → 代码怎么跑 → 坑在哪 → 怎么改」的顺序拆开讲。2. 拆开压缩包先看结构Spark2.X 实时链路的模块划分与选型理由拿到一个源码包我习惯先不急着跑而是把目录和类名过一遍判断作者的技术选型和链路设计。这份资源的类名信息量很大基本能反推出整套架构。2.1 从类名反推架构Kafka Structured Streaming MySQL 三段式包里出现的核心类可以分成三组。第一组是数据接入层StreamingKafka8和StreamingKafka10两个版本并存说明作者兼容了 Kafka 0.8 和 0.10 两代消费者 API这是 Spark2.X 时代的典型做法——Spark 2.0 到 2.2 主要对接 Kafka 0.8 的 Receiver/Direct 模式2.3 之后才稳定支持 Kafka 0.10 的 Structured Streaming 直连。第二组是计算层StructuredStreamingKafka及其$Weblog变体走的是 Structured Streaming 的 DataFrame/Dataset 编程模型而不是老的 DStream。第三组是输出层JDBCSink负责把结果写进 MySQLMySqlPool管连接池WeblogService封装业务查询。这个三段式Kafka 接入 → Structured Streaming 计算 → JDBC 落地是新闻话题实时统计最主流的骨架。选 Structured Streaming 而不是 DStream 的理由很实在它基于 Catalyst 优化器窗口聚合写起来是 SQL 风格的增量输出模式update/append/complete能直接控制结果怎么吐给下游而 DStream 要自己维护状态、自己处理乱序代码量和出错概率都高一个量级。2.2 环境与依赖版本对齐别让版本错配毁掉一整天Spark2.X 项目最怕的就是版本错配。这份资源没有在正文里写死具体小版本但根据类名和 API 用法可以确定一个能跑通的组合区间。下面这张表是我按经验给的对照实际以你本地已有环境为准能对齐就对齐对不齐优先改 Spark 而不是改代码逻辑。组件建议版本说明JDK1.8Spark2.X 对 JDK9 支持差别用高版本Scala2.11.xSpark2.X 默认编译版本2.12 需重新编译Spark2.2.0 ~ 2.4.x2.3 以下用 Kafka0.8 那套类2.3 用 0.10Kafka0.10.x与 StreamingKafka10 对应MySQL5.78.0 驱动类名和时区参数有变化构建工具Maven 3.5依赖坐标按 pom 走环境变量这块SPARK_HOME、JAVA_HOME、SCALA_HOME三个必须配好spark-submit才能找到对应 jar。我一般会在spark-env.sh里显式指定SPARK_EXECUTOR_MEMORY和SPARK_DRIVER_MEMORY因为实时任务默认内存偏小窗口聚合一堆积就容易 OOM。2.3 导入工程与依赖检查先让 IDE 不报红把源码导入 IntelliJ IDEA 后第一步是确认 Maven 依赖能全部拉下来。Structured Streaming 对接 Kafka 需要spark-sql-kafka-0-10这个包JDBC 落地需要mysql-connector-java。如果 pom 里用的是 provided 作用域本地跑要临时改成 compile否则运行时报 ClassNotFound。# 检查关键依赖是否在本地仓库 ls ~/.m2/repository/org/apache/spark/spark-sql-kafka-0-10_2.11/ ls ~/.m2/repository/mysql/mysql-connector-java/ # 如果缺失手动拉一次版本按 pom 里的来 mvn dependency:resolve -DincludeArtifactIdsspark-sql-kafka-0-10_2.11,mysql-connector-java上面第一条命令确认 Kafka 连接器和 MySQL 驱动是否已下载第二条在依赖缺失时触发 Maven 重新解析。参数-DincludeArtifactIds用来只拉指定构件避免全量下载拖时间。如果拉取一直失败先看settings.xml里的镜像地址是否可达这是国内环境最常见的卡点。3. 把实时链路跑起来Kafka 生产、Structured Streaming 消费与 JDBC 落地环境就绪后真正的工作量在「让数据从 Kafka 流进来、算完、写出去」这条链路上。这一章按数据流向拆成三步每步都给可抄的命令和代码骨架。3.1 启动 Kafka 并灌入新闻测试数据实时任务没有数据源就是空转。先起一个单节点 Kafka测试够用建 topic再用控制台生产者灌几条模拟新闻。# 启动 ZooKeeper 和 Kafka版本按你本地0.10 时代还需 ZK bin/zookeeper-server-start.sh -daemon config/zookeeper.properties bin/kafka-server-start.sh -daemon config/server.properties # 建 topic3 分区 1 副本测试环境够用 bin/kafka-topics.sh --create --zookeeper localhost:2181 \ --replication-factor 1 --partitions 3 --topic news_topic # 灌入模拟新闻字段用逗号分隔话题,标题,时间戳 bin/kafka-console-producer.sh --broker-list localhost:9092 --topic news_topic--partitions 3是为了让后续 Spark 消费时能并行起 3 个 task单分区的话并行度上不去窗口聚合会成瓶颈。生产消息时每条一行格式要和消费端解析逻辑对齐——这是后面最容易翻车的地方先记住字段顺序。3.2 Structured Streaming 消费 Kafka 的核心代码骨架消费端的核心是把 Kafka 的 value 字段解析成结构化列再按窗口聚合。下面这段是这类项目的通用骨架字段名和窗口时长按你实际业务改。// 从 Kafka 读取流注意 bootstrapServers 和 subscribe 参数 val raw spark.readStream .format(kafka) .option(kafka.bootstrap.servers, localhost:9092) .option(subscribe, news_topic) .option(startingOffsets, latest) .load() // Kafka 的 value 是二进制先转字符串再按逗号切分 val parsed raw.selectExpr(CAST(value AS STRING) as line) .select( split($line, ,)(0).as(topic_name), split($line, ,)(1).as(title), to_timestamp(split($line, ,)(2)).as(event_time) ) // 按 1 分钟滚动窗口统计每个话题的出现次数 val windowed parsed .withWatermark(event_time, 2 minutes) .groupBy(window($event_time, 1 minute), $topic_name) .count() // 输出到 MySQLforeachBatch 里调 JDBCSink val query windowed.writeStream .outputMode(update) .foreachBatch { (batchDF: Dataset[Row], batchId: Long) JDBCSink.write(batchDF, batchId) } .trigger(ProcessingTime(10 seconds)) .start()逐段说明startingOffsets设成latest表示只消费启动后的新消息调试时想重放历史数据就改成earliestwithWatermark是处理乱序的关键设 2 分钟意味着迟到超过 2 分钟的数据会被丢弃这个值要按你数据源的延迟分布调outputMode(update)让每个批次只输出有更新的窗口比complete省资源trigger设 10 秒是微批的间隔太短会增加调度开销太长则实时性下降。3.3 JDBCSink 写 MySQL连接池与幂等写入JDBCSink和MySqlPool这两个类解决的是「每个批次都要写库不能每次新建连接」的问题。连接池复用连接写入时用INSERT ... ON DUPLICATE KEY UPDATE保证同一窗口重复输出不会产生脏数据。object JDBCSink { def write(df: Dataset[Row], batchId: Long): Unit { df.foreachPartition { partition val conn MySqlPool.getConnection() // 从池里取不是新建 val stmt conn.prepareStatement( INSERT INTO topic_stat(window_start, topic_name, cnt) VALUES(?,?,?) ON DUPLICATE KEY UPDATE cnt VALUES(cnt)) partition.foreach { row stmt.setTimestamp(1, row.getTimestamp(0)) stmt.setString(2, row.getString(1)) stmt.setLong(3, row.getLong(2)) stmt.addBatch() } stmt.executeBatch() MySqlPool.release(conn) // 用完归还别 close } } }foreachPartition而不是foreach是因为前者每个分区只建一次连接后者每行都建性能差几十倍。ON DUPLICATE KEY UPDATE要求 MySQL 表上对(window_start, topic_name)建唯一索引否则幂等不生效。MySqlPool.release归还连接而不是关闭池子大小在MySqlPool初始化时配一般设 10 到 20 够用。3.4 提交任务与验证结果代码打包后用spark-submit提交。本地模式先验证逻辑再上集群。# 本地模式跑通逻辑 spark-submit --class com.xxx.StructuredStreamingKafka \ --master local[4] \ --packages org.apache.spark:spark-sql-kafka-0-10_2.11:2.4.0 \ target/news-stat.jar # 验证 MySQL 里有没有数据 mysql -uroot -p -e SELECT * FROM news_db.topic_stat ORDER BY window_start DESC LIMIT 10;--master local[4]用 4 个线程模拟并行--packages在线拉取 Kafka 连接器生产环境建议打进 jar 或用--jars指定本地路径避免每次联网。验证时如果 MySQL 表是空的先看 Spark 控制台有没有批次输出再查 Kafka 里是否真有数据逐段排查。4. 避坑与排查这套 Spark2.X 实时链路最容易翻车的五个地方实时任务的坑大多不在业务逻辑而在版本、序列化、状态和资源这些「环境层」。下面五条是我拆这类项目时反复遇到的每条按现象、原因、解决写。4.1 现象任务启动即报 NoSuchMethodError 或 ClassNotFoundException原因基本是版本错配。Spark2.X 的spark-sql-kafka-0-10必须和 Spark 主版本严格对应2.4 的 Spark 配 2.3 的连接器就会在运行时找不到方法。另外 Scala 2.11 和 2.12 的构件后缀不同拉错后缀直接类加载失败。解决先spark-submit --version确认 Spark 版本再按_2.11或_2.12后缀选连接器版本号与 Spark 保持一致。pom 里把spark-sql-kafka-0-10的版本写成和spark-core同一个属性变量避免手改漏改。4.2 现象窗口聚合结果一直不输出或者输出后又被覆盖原因通常是 watermark 和输出模式配合错了。update模式下窗口只有在 watermark 超过窗口结束时间后才会被最终确定如果 watermark 设得比数据延迟还小迟到数据被丢结果就偏。complete模式则会每次输出全量窗口数据量大时直接把 driver 内存打满。解决先统计数据源的实际延迟分布watermark 设成 P99 延迟再加一点余量。调试阶段用complete看全貌上线切update。如果结果被覆盖检查 MySQL 的唯一索引是否建对。4.3 现象MySQL 连接数暴涨最后报 Too many connections原因在JDBCSink里用了foreach而不是foreachPartition或者用完连接没归还而是close了池里的连接。前者每行建一次连接后者让池子以为连接还在、实际已断反复新建。解决统一用foreachPartition连接从池取、用完release归还。池大小按分区数乘以并发批次估算别盲目设大。MySQL 侧max_connections也要留余量。4.4 现象Kafka 消费 lag 持续增长任务越跑越慢原因是消费并行度和 Kafka 分区数不匹配。topic 3 个分区Spark 只起了 1 个 task吞吐上不去。或者trigger间隔太短批次还没处理完下一个就来了任务排队堆积。解决让 Spark 消费的并行度等于分区数Structured Streaming 默认按分区并行确认没被repartition(1)之类操作破坏。trigger间隔按单批次处理耗时来定一般设成处理耗时的 1.5 到 2 倍。4.5 现象本地跑通上集群后报序列化异常原因是JDBCSink或MySqlPool里引用了不可序列化的对象比如直接持有 Connection 的成员变量driver 往 executor 分发任务时序列化失败。解决把连接相关逻辑全部收进foreachPartition内部对象在 executor 侧创建不跨节点传递。所有要在闭包里用的外部变量确认实现了Serializable。5. 在源码基础上做二次开发换数据源、调窗口、加指标的具体手法跑通只是起点这份资源真正的价值在于它是个可改的骨架。下面说三个我实际改过的方向都是新闻话题统计场景下最常被要求加的功能。5.1 换数据源从 Kafka 切到 Socket 或文件流调试阶段不想起 Kafka可以把readStream.format(kafka)换成 socket 源代码改动很小。// 用 socket 源替代 Kafka方便本地调试 val raw spark.readStream .format(socket) .option(host, localhost) .option(port, 9999) .load() // 后续解析逻辑不变只是 value 列名从 kafka 的 value 变成 socket 的 valuesocket 源只有单分区、不支持 offset 管理只适合验证逻辑别用在生产。文件流则适合回放历史新闻做回归测试把format换成text或json指向一个目录Spark 会自动监听新文件。5.2 调窗口与指标从「计数」扩展到「热度分」原始代码只统计话题出现次数实际业务往往要一个加权热度分。改法是在groupBy后加聚合表达式把不同字段按权重算进去。指标计算方式权重建议出现次数count0.5独立来源数countDistinct(source)0.3时间衰减按窗口距当前时间加权0.2窗口时长也要按业务调突发新闻用 1 分钟滚动窗口话题趋势用 5 到 10 分钟滑动窗口。滑动窗口的slideDuration小于windowDuration时同一数据会进多个窗口计算量成倍增加资源要提前估。5.3 加状态与去重用 dropDuplicates 处理重复新闻新闻源经常重复推送同一条直接计数会虚高。Structured Streaming 支持在流上做去重但要注意它会维护状态状态大小随时间增长。// 按标题去重配合 watermark 限制状态保留时间 val deduped parsed .withWatermark(event_time, 10 minutes) .dropDuplicates(title)dropDuplicates必须配合 watermark 用否则状态无限增长跑几天就 OOM。watermark 设 10 分钟意味着只对 10 分钟内的重复做去重更早的状态被清理。这个取舍要看业务对重复的容忍度。5.4 验证改动是否生效三个必看的观测点改完代码别急着上线先看三个地方。一是 Spark UI 的 Streaming 页确认批次处理耗时稳定、没有堆积二是 MySQL 里的结果表确认窗口边界和计数符合预期三是 Kafka 的 consumer lag确认消费跟得上生产。这三个点任何一个异常都说明改动引入了新问题。我自己的习惯是每次改完窗口或 watermark都先用一小批带时间戳的测试数据跑一遍手动核对窗口划分对不对再放真实流量。从那以后我每次调实时任务的窗口参数都强制走一遍「小数据验证 → 看 UI → 查落库」这三步省下过好几次半夜爬起来排查的功夫。希望帮到你。本文还有配套的精品资源点击获取