摘要foreachRDD 是 Spark Streaming 里最常用、也最容易写挂的输出算子。这篇讲清它和 transform 的区别——foreachRDD 在 Driver 端拿到 RDD、真正执行在 Executor 端以及由此带来的一个高频坑连接到底该建在哪。用三种连接方式的性能对比和一段完整的写 MySQL 代码把 foreachRDD 的正确姿势讲透。关键词Spark Streaming, foreachRDD, foreachPartition, 连接管理, 事务, exactly-once一、foreachRDD 是什么先分清它和 transformDStream 的算子分两类转换操作返回新 DStream和输出操作不返回触发计算。foreachRDD属于后者是 output 操作。它和transform长得像本质区别就一条// transform拿到 RDD返回新 RDD继续流式计算valnewDsds.transform(rddrdd.map(...))// foreachRDD拿到 RDD做输出到此为止无返回值ds.foreachRDD(rdd{/* 写外部存储 */})foreachRDD是 Action 语义——一旦调用前面所有 DStream 转换才会真正执行。所以一个流里如果没有 foreachRDD或 print、saveAsTextFiles 这类 output 操作整个流是不会动的。二、最核心的坑连接建在哪这是 foreachRDD 用法的分水岭。关键要理解执行位置foreachRDD的闭包在 Driver 端定义、在 Driver 端拿到 RDD但它内部的foreachPartition/foreach真正跑在Executor端。而闭包里引用的外部变量会被序列化发送到 Executor。连接对象Connection不可序列化所以在foreachRDD这一层建连接要么报NotSerializableException要么每个 Driver 只建一个连接却要发给所有 Executor完全不对。结论一句话连接必须在 Executor 端创建也就是放进foreachPartition或foreach里面而不是foreachRDD这一层。三、三种连接方式性能差三个数量级假设要写 100 万条记录到 MySQL看三种写法的差别。方式一foreach 每条建连接反模式rdd.foreach{rowvalconnDriverManager.getConnection(url,user,pwd)save(conn,row)conn.close()}100 万条记录 100 万次 TCP 握手 认证 建连接。连接开销远大于写入本身吞吐直接崩。这是最常见的新手错误要极力避免。方式二foreachPartition 每分区建连接常用rdd.foreachPartition{itervalconnDriverManager.getConnection(url,user,pwd)try{iter.foreach(rowsave(conn,row))}finally{conn.close()}}foreachPartition的闭包在每个分区的第一条记录上执行一次所以一个分区只建一个连接分区内所有记录复用。假设每分区 1000 条连接次数就从 100 万降到 1000。这是大多数场景够用的写法。注意finally里关连接防止中途异常导致连接泄漏。方式三连接池生产最优方式二的问题在于连接数 分区数分区一多连接就多。生产上用连接池跨分区复用// Executor 端懒加载单例连接池objectConnectionPool{lazyvalds:HikariDataSource{valcfgnewHikariConfig()cfg.setJdbcUrl(url);cfg.setUsername(user);cfg.setPassword(pwd)cfg.setMaximumPoolSize(10)newHikariDataSource(cfg)}}rdd.foreachPartition{itervalconnConnectionPool.ds.getConnectiontry{iter.foreach(rowsave(conn,row))}finally{conn.close()}}连接池用lazy val单例 静态对象保证每个 Executor 只初始化一次池子里的连接跨分区复用连接数可控。四、写外部存储的事务问题foreachRDD 写外部存储默认是 at-least-once 语义处理到一半 Executor 挂了重算时会重复写入已经写过的记录。要往 exactly-once 靠需要三件事配合幂等写给记录设计唯一键用 upsert 替代 insert重复写同一行结果不变。每分区一个事务一个分区的写入放在一个事务里全部成功才 commit失败整体回滚。offset 与结果绑定只有写入成功才提交 offset这在 Direct 模式下天然支持offset 自管理。纯靠 foreachRDD 单算子做不到 exactly-once它只能做到配合外部系统幂等 事务之后的效果。这点别被网上foreachRDD 保证 exactly-once的说法误导。五、完整示例写 MySQLimportorg.apache.spark.streaming.{Seconds,StreamingContext}importcom.zaxxer.hikari.{HikariConfig,HikariDataSource}objectStreamingToMySQL{// Executor 端懒加载连接池单例objectConnectionPool{lazyvalds:HikariDataSource{valcfgnewHikariConfig()cfg.setJdbcUrl(jdbc:mysql://host:3306/db)cfg.setUsername(root)cfg.setPassword(password)cfg.setMaximumPoolSize(10)newHikariDataSource(cfg)}}defmain(args:Array[String]):Unit{valsscnewStreamingContext(local[2],stream-to-mysql,Seconds(2))vallinesssc.socketTextStream(localhost,9999)lines.foreachRDD{rddif(!rdd.isEmpty){rdd.foreachPartition{itervalconnConnectionPool.ds.getConnection conn.setAutoCommit(false)try{valpsconn.prepareStatement(INSERT INTO result(word, cnt) VALUES(?, 1) ON DUPLICATE KEY UPDATE cnt cnt 1)iter.foreach{wordps.setString(1,word)ps.addBatch()}ps.executeBatch()conn.commit()// 分区内一个事务成功才提交}catch{casee:Exceptionconn.rollback();throwe}finally{conn.close()// 归还连接池}}}}ssc.start();ssc.awaitTermination()}}这段代码把前四节的点都串起来了连接池单例、分区级连接、批处理 幂等 upsert、事务 commit/rollback。六、总结foreachRDD 是 output 操作Driver 端拿 RDD、Executor 端执行和 transform 的区别是有无返回值。连接必须建在 Executor 端foreachPartition/foreach 内foreachRDD 这层建连接会踩序列化坑。连接方式从差到好foreach 每条建连接 → foreachPartition 每分区建连接 → 连接池复用。默认 at-least-once要 exactly-once 得靠幂等写 分区事务 offset 绑定三件套。作者大数据技术实践者博客blog.starzy.cnGitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
摘要foreachRDD 是 Spark Streaming 里最常用、也最容易写挂的输出算子。这篇讲清它和 transform 的区别——foreachRDD 在 Driver 端拿到 RDD、真正执行在 Executor 端以及由此带来的一个高频坑连接到底该建在哪。用三种连接方式的性能对比和一段完整的写 MySQL 代码把 foreachRDD 的正确姿势讲透。关键词Spark Streaming, foreachRDD, foreachPartition, 连接管理, 事务, exactly-once一、foreachRDD 是什么先分清它和 transformDStream 的算子分两类转换操作返回新 DStream和输出操作不返回触发计算。foreachRDD属于后者是 output 操作。它和transform长得像本质区别就一条// transform拿到 RDD返回新 RDD继续流式计算valnewDsds.transform(rddrdd.map(...))// foreachRDD拿到 RDD做输出到此为止无返回值ds.foreachRDD(rdd{/* 写外部存储 */})foreachRDD是 Action 语义——一旦调用前面所有 DStream 转换才会真正执行。所以一个流里如果没有 foreachRDD或 print、saveAsTextFiles 这类 output 操作整个流是不会动的。二、最核心的坑连接建在哪这是 foreachRDD 用法的分水岭。关键要理解执行位置foreachRDD的闭包在 Driver 端定义、在 Driver 端拿到 RDD但它内部的foreachPartition/foreach真正跑在Executor端。而闭包里引用的外部变量会被序列化发送到 Executor。连接对象Connection不可序列化所以在foreachRDD这一层建连接要么报NotSerializableException要么每个 Driver 只建一个连接却要发给所有 Executor完全不对。结论一句话连接必须在 Executor 端创建也就是放进foreachPartition或foreach里面而不是foreachRDD这一层。三、三种连接方式性能差三个数量级假设要写 100 万条记录到 MySQL看三种写法的差别。方式一foreach 每条建连接反模式rdd.foreach{rowvalconnDriverManager.getConnection(url,user,pwd)save(conn,row)conn.close()}100 万条记录 100 万次 TCP 握手 认证 建连接。连接开销远大于写入本身吞吐直接崩。这是最常见的新手错误要极力避免。方式二foreachPartition 每分区建连接常用rdd.foreachPartition{itervalconnDriverManager.getConnection(url,user,pwd)try{iter.foreach(rowsave(conn,row))}finally{conn.close()}}foreachPartition的闭包在每个分区的第一条记录上执行一次所以一个分区只建一个连接分区内所有记录复用。假设每分区 1000 条连接次数就从 100 万降到 1000。这是大多数场景够用的写法。注意finally里关连接防止中途异常导致连接泄漏。方式三连接池生产最优方式二的问题在于连接数 分区数分区一多连接就多。生产上用连接池跨分区复用// Executor 端懒加载单例连接池objectConnectionPool{lazyvalds:HikariDataSource{valcfgnewHikariConfig()cfg.setJdbcUrl(url);cfg.setUsername(user);cfg.setPassword(pwd)cfg.setMaximumPoolSize(10)newHikariDataSource(cfg)}}rdd.foreachPartition{itervalconnConnectionPool.ds.getConnectiontry{iter.foreach(rowsave(conn,row))}finally{conn.close()}}连接池用lazy val单例 静态对象保证每个 Executor 只初始化一次池子里的连接跨分区复用连接数可控。四、写外部存储的事务问题foreachRDD 写外部存储默认是 at-least-once 语义处理到一半 Executor 挂了重算时会重复写入已经写过的记录。要往 exactly-once 靠需要三件事配合幂等写给记录设计唯一键用 upsert 替代 insert重复写同一行结果不变。每分区一个事务一个分区的写入放在一个事务里全部成功才 commit失败整体回滚。offset 与结果绑定只有写入成功才提交 offset这在 Direct 模式下天然支持offset 自管理。纯靠 foreachRDD 单算子做不到 exactly-once它只能做到配合外部系统幂等 事务之后的效果。这点别被网上foreachRDD 保证 exactly-once的说法误导。五、完整示例写 MySQLimportorg.apache.spark.streaming.{Seconds,StreamingContext}importcom.zaxxer.hikari.{HikariConfig,HikariDataSource}objectStreamingToMySQL{// Executor 端懒加载连接池单例objectConnectionPool{lazyvalds:HikariDataSource{valcfgnewHikariConfig()cfg.setJdbcUrl(jdbc:mysql://host:3306/db)cfg.setUsername(root)cfg.setPassword(password)cfg.setMaximumPoolSize(10)newHikariDataSource(cfg)}}defmain(args:Array[String]):Unit{valsscnewStreamingContext(local[2],stream-to-mysql,Seconds(2))vallinesssc.socketTextStream(localhost,9999)lines.foreachRDD{rddif(!rdd.isEmpty){rdd.foreachPartition{itervalconnConnectionPool.ds.getConnection conn.setAutoCommit(false)try{valpsconn.prepareStatement(INSERT INTO result(word, cnt) VALUES(?, 1) ON DUPLICATE KEY UPDATE cnt cnt 1)iter.foreach{wordps.setString(1,word)ps.addBatch()}ps.executeBatch()conn.commit()// 分区内一个事务成功才提交}catch{casee:Exceptionconn.rollback();throwe}finally{conn.close()// 归还连接池}}}}ssc.start();ssc.awaitTermination()}}这段代码把前四节的点都串起来了连接池单例、分区级连接、批处理 幂等 upsert、事务 commit/rollback。六、总结foreachRDD 是 output 操作Driver 端拿 RDD、Executor 端执行和 transform 的区别是有无返回值。连接必须建在 Executor 端foreachPartition/foreach 内foreachRDD 这层建连接会踩序列化坑。连接方式从差到好foreach 每条建连接 → foreachPartition 每分区建连接 → 连接池复用。默认 at-least-once要 exactly-once 得靠幂等写 分区事务 offset 绑定三件套。作者大数据技术实践者博客blog.starzy.cnGitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
