摘要上一篇讲了 foreachRDD 是输出操作这篇讲它的姊妹算子 transform——一个转换操作拿到 RDD 处理后返回新 RDD让流继续往下算。它的真正价值在于DStream 只有几十个算子而 transform 让你能直接用 RDD 全套 API。这篇用黑名单过滤、实时流 join 维度表、用 RDD 独有算子三个场景把 transform 的用法和坑讲清楚。关键词Spark Streaming, transform, 广播变量, 实时流 join 维度表, RDD API一、transform 和 foreachRDD 是一对先分清两者都让你在 Driver 端直接拿到 RDD但方向相反// transform转换操作拿到 RDD → 返回新 RDD → 流继续往下算valnewDsds.transform(rddrdd.filter(...))// foreachRDD输出操作拿到 RDD → 做输出 → 到此为止ds.foreachRDD(rddrdd.foreachPartition(...))判断依据就一条有没有返回值。transform 返回新 DStream所以它是惰性的、可链式的foreachRDD 不返回是 Action 语义会触发前面所有转换真正执行。一个流里 transform 和 foreachRDD 通常是配合着用的transform 负责加工foreachRDD 负责落地。二、transform 的定位DStream 和 RDD 之间的桥这是理解 transform 为什么存在的关键。DStream 的算子只有几十个而 RDD 有上百个。很多 RDD 上的能力——mapPartitions、sortBy、distinct、sample、subtract、intersection——DStream 根本不提供。transform 就是那道桥它把 RDD 交到你手上你可以在里面用任何 RDD 算子再把结果包回 DStream。valsortedds.transform(rddrdd.sortBy(_.ts,ascendingfalse))这个sortBy是 DStream 没有的不借 transform 根本写不出来。三、场景一黑名单过滤transform 广播变量实时日志里要过滤掉一批黑名单用户黑名单在外部库里、会定期更新。这是 transform 最典型的用法。// 黑名单加载一次广播出去每个 Executor 一份副本valblacklistssc.sparkContext.broadcast(loadBlacklist())valfilteredlogDStream.transform{rddrdd.filter(record!blacklist.value.contains(record.userId))}为什么要广播变量如果不广播直接在filter闭包里引用blacklist这个集合会被序列化后随闭包发给每个 Task——每个 Task 都带一份完整黑名单网络和内存开销翻倍。广播变量让每个 Executor 只持有一份所有 Task 共享。为什么要用 transform黑名单是 RDD 层面的集合运算DStream 的filter只能传函数没法方便地引用一个外部集合做contains判断。放进 transform 里你就拿到了 RDD可以自由地用广播变量做过滤。四、场景二实时流 join 维度表交易流里只有商品 ID要关联商品维度表补上名称和类目。维度表通常不大适合广播 join。// 维度表加载成 Map广播出去valdimssc.sparkContext.broadcast(loadDimTable().collectAsMap())valenrichedorderDStream.transform{rddrdd.map{ordervalnamedim.value.getOrElse(order.productId,unknown)(order,name)}}这里的关键点小表广播 join维度表几百 MB 以内用collectAsMap拉到 Driver、广播到各 Executorjoin 时纯内存查 Map不用 shuffle。维度表会变怎么办用transform每次都从 Driver 侧重新读维度表或者维护一个定时刷新的广播变量Spark 1.6 的spark.streaming.unpersist配合定时任务。维度表很大的时候广播就不合适了得换外部 KV 存储HBase/Redis做关联。五、场景三用 RDD 独有的算子有些需求 DStream 直接写不了借 transform 就能写。举两个// distinct 去重DStream 没有valdedupedds.transform(_.distinct())// mapPartitions分区级复用重对象连接等valprocessedds.transform{rddrdd.mapPartitions{iter// 每分区初始化一次比如建连接、加载模型valhelpernewExpensiveHelper()iter.map(helper.process)}}mapPartitions这个场景和上一篇 foreachRDD 里讲的连接管理是同一个道理——重量级对象放分区级初始化而不是每条记录 new 一个。六、闭包序列化的坑和 foreachRDD 一样transform 的闭包在 Driver 端定义、内部对 RDD 的操作在 Executor 端执行闭包引用的外部变量同样会被序列化发送。所以连接、文件句柄这类不可序列化的对象不能直接写在 transform 闭包里引用要放进mapPartitions里。大对象用广播变量别让每个 Task 都序列化一份。这两条和 foreachRDD 完全一致写 transform 时同样要盯紧。七、总结transform 是转换操作返回新 RDD 让流继续算foreachRDD 是输出操作到此为止。判断依据是有无返回值。transform 是 DStream 到 RDD 的桥让你能用 RDD 全套 API突破 DStream 算子限制。三大场景黑名单过滤广播变量、实时流 join 维度表小表广播、RDD 独有算子mapPartitions/distinct/sortBy。闭包序列化的坑和 foreachRDD 一样重对象放分区级初始化大对象用广播变量。作者大数据技术实践者博客blog.starzy.cnGitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
摘要上一篇讲了 foreachRDD 是输出操作这篇讲它的姊妹算子 transform——一个转换操作拿到 RDD 处理后返回新 RDD让流继续往下算。它的真正价值在于DStream 只有几十个算子而 transform 让你能直接用 RDD 全套 API。这篇用黑名单过滤、实时流 join 维度表、用 RDD 独有算子三个场景把 transform 的用法和坑讲清楚。关键词Spark Streaming, transform, 广播变量, 实时流 join 维度表, RDD API一、transform 和 foreachRDD 是一对先分清两者都让你在 Driver 端直接拿到 RDD但方向相反// transform转换操作拿到 RDD → 返回新 RDD → 流继续往下算valnewDsds.transform(rddrdd.filter(...))// foreachRDD输出操作拿到 RDD → 做输出 → 到此为止ds.foreachRDD(rddrdd.foreachPartition(...))判断依据就一条有没有返回值。transform 返回新 DStream所以它是惰性的、可链式的foreachRDD 不返回是 Action 语义会触发前面所有转换真正执行。一个流里 transform 和 foreachRDD 通常是配合着用的transform 负责加工foreachRDD 负责落地。二、transform 的定位DStream 和 RDD 之间的桥这是理解 transform 为什么存在的关键。DStream 的算子只有几十个而 RDD 有上百个。很多 RDD 上的能力——mapPartitions、sortBy、distinct、sample、subtract、intersection——DStream 根本不提供。transform 就是那道桥它把 RDD 交到你手上你可以在里面用任何 RDD 算子再把结果包回 DStream。valsortedds.transform(rddrdd.sortBy(_.ts,ascendingfalse))这个sortBy是 DStream 没有的不借 transform 根本写不出来。三、场景一黑名单过滤transform 广播变量实时日志里要过滤掉一批黑名单用户黑名单在外部库里、会定期更新。这是 transform 最典型的用法。// 黑名单加载一次广播出去每个 Executor 一份副本valblacklistssc.sparkContext.broadcast(loadBlacklist())valfilteredlogDStream.transform{rddrdd.filter(record!blacklist.value.contains(record.userId))}为什么要广播变量如果不广播直接在filter闭包里引用blacklist这个集合会被序列化后随闭包发给每个 Task——每个 Task 都带一份完整黑名单网络和内存开销翻倍。广播变量让每个 Executor 只持有一份所有 Task 共享。为什么要用 transform黑名单是 RDD 层面的集合运算DStream 的filter只能传函数没法方便地引用一个外部集合做contains判断。放进 transform 里你就拿到了 RDD可以自由地用广播变量做过滤。四、场景二实时流 join 维度表交易流里只有商品 ID要关联商品维度表补上名称和类目。维度表通常不大适合广播 join。// 维度表加载成 Map广播出去valdimssc.sparkContext.broadcast(loadDimTable().collectAsMap())valenrichedorderDStream.transform{rddrdd.map{ordervalnamedim.value.getOrElse(order.productId,unknown)(order,name)}}这里的关键点小表广播 join维度表几百 MB 以内用collectAsMap拉到 Driver、广播到各 Executorjoin 时纯内存查 Map不用 shuffle。维度表会变怎么办用transform每次都从 Driver 侧重新读维度表或者维护一个定时刷新的广播变量Spark 1.6 的spark.streaming.unpersist配合定时任务。维度表很大的时候广播就不合适了得换外部 KV 存储HBase/Redis做关联。五、场景三用 RDD 独有的算子有些需求 DStream 直接写不了借 transform 就能写。举两个// distinct 去重DStream 没有valdedupedds.transform(_.distinct())// mapPartitions分区级复用重对象连接等valprocessedds.transform{rddrdd.mapPartitions{iter// 每分区初始化一次比如建连接、加载模型valhelpernewExpensiveHelper()iter.map(helper.process)}}mapPartitions这个场景和上一篇 foreachRDD 里讲的连接管理是同一个道理——重量级对象放分区级初始化而不是每条记录 new 一个。六、闭包序列化的坑和 foreachRDD 一样transform 的闭包在 Driver 端定义、内部对 RDD 的操作在 Executor 端执行闭包引用的外部变量同样会被序列化发送。所以连接、文件句柄这类不可序列化的对象不能直接写在 transform 闭包里引用要放进mapPartitions里。大对象用广播变量别让每个 Task 都序列化一份。这两条和 foreachRDD 完全一致写 transform 时同样要盯紧。七、总结transform 是转换操作返回新 RDD 让流继续算foreachRDD 是输出操作到此为止。判断依据是有无返回值。transform 是 DStream 到 RDD 的桥让你能用 RDD 全套 API突破 DStream 算子限制。三大场景黑名单过滤广播变量、实时流 join 维度表小表广播、RDD 独有算子mapPartitions/distinct/sortBy。闭包序列化的坑和 foreachRDD 一样重对象放分区级初始化大对象用广播变量。作者大数据技术实践者博客blog.starzy.cnGitHubstarzy1990.github.io专注 AI Agent · LangGraph · RAG · 大数据架构 · 数据工程实践
