如何在PySpark中获取Structured Streaming DataFrame的单行数据

关于DStream是否适用的结论 #

DStream在目前最新的Spark 3.x版本中仍然可用，但属于处于维护模式的遗留API，Spark官方自2.3版本起就已将Structured Streaming作为流处理的首推实现，不建议新业务使用DStream开发，因此它不是你当前场景的最优选择。

Structured Streaming下逐行处理的可选方案 #

你完全不需要切换到DStream，现有Structured Streaming的原生能力就能满足逐行获取、处理数据的需求，常用方案如下：

• 方案1：直接使用map/foreach算子逐行处理
  不需要将DataFrame转为RDD就能直接访问每一行数据，对读取到的流DataFrame直接调用map算子即可拿到每行的Row对象，在算子内部实现和外部数据库的比对逻辑即可。示例代码（Scala）：

  import org.apache.spark.sql.Row
  
  // 读取Kafka流数据
  val kafkaStreamDF = spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "你的Kafka broker地址")
    .option("subscribe", "你的Topic名称")
    .load()
  
  // 逐行处理逻辑
  val resultDF = kafkaStreamDF.map(row => {
    // 提取当前行的Kafka数据，例如获取消息内容
    val msgContent = row.getAs[Array[Byte]]("value").toString
    // 此处实现和外部数据库的比对逻辑
    // 返回自定义的处理结果
    (msgContent, "比对结果")
  })
  
  // 启动流任务
  val query = resultDF.writeStream
    .format("console")
    .start()
  query.awaitTermination()
  

• 方案2：使用foreachBatch兼容RDD处理逻辑
  如果你已经有成熟的RDD侧比对逻辑，可以通过foreachBatch拿到每批次的静态DataFrame，直接转为RDD复用原有逻辑：

  val query = kafkaStreamDF.writeStream
    .foreachBatch { (batchDF: org.apache.spark.sql.DataFrame, batchId: Long) =>
      // 批次DataFrame转RDD，沿用原有处理逻辑
      batchDF.rdd.foreach(row => {
        // 逐行比对逻辑
      })
    }
    .start()
  query.awaitTermination()
  

• 方案3：流静态JOIN优化比对性能
  如果你的外部比对库数据量不大、支持JDBC读取，可以先将外部库数据读为静态DataFrame，直接和流DataFrame做JOIN操作，性能远高于逐行请求外部数据库，还能享受Spark的查询优化能力。

总结：现有Structured Streaming的原生API就能完全覆盖你的需求，相比DStream性能更优、维护成本更低，不需要切换技术栈。

内容的提问来源于stack exchange，提问作者Donsitoz