PySpark Kafka批处理API：多转换是否读取同一份数据？

我刚接触Spark，对Kafka批处理API的load行为有疑问。我通过以下方式读取DataFrame：

df = spark.read \
        .format("kafka") \
        .option("subscribe", topic_name) \ # 其他配置项省略
        .load()

之后我想对这个DataFrame执行不同操作：比如提取各分区读取的最后偏移量，同时过滤数据并生成Parquet文件：

aggregates = df.groupBy("partition").max("offset").collect()

# 对df执行一些过滤操作

df.write.parquet(target_path)

我的理解是load调用时仅读取一次数据，后续聚合和写入操作都基于同一份数据，处理期间Kafka新增的事件会被忽略。请问这个理解是否正确？还是collect和write.parquet会重新触发加载，导致读取到不同的Kafka事件？

你的理解不正确，Spark的DataFrame是懒加载的：调用load()的时候并没有真正从Kafka读取数据，只是定义了数据源的元信息和读取逻辑。只有当你执行collect()、write.parquet()这类行动操作时，才会触发实际的数据读取。

在你的代码里，collect()和write.parquet()是两个独立的行动操作，这意味着Spark会触发两次完全独立的作业，也就是会两次去Kafka读取数据。如果这期间Kafka有新的消息写入，两次读取的数据集就会不一样——比如聚合得到的最大偏移量是第一次读取的结果，而写入Parquet的是第二次读取的包含新消息的数据，两者就不对应了。

要避免这种情况，确保两次操作基于同一份数据，有几种常用方法：

• 缓存DataFrame：在执行行动操作前调用cache()或persist()，这样第一次行动操作读取的数据会被缓存到内存/磁盘，第二次行动操作直接使用缓存的数据，不会再去Kafka读取。示例代码：

  df = spark.read \
          .format("kafka") \
          .option("subscribe", topic_name) \
          # 其他配置项
          .load()
  
  # 缓存数据
  df.cache()
  
  # 第一次行动操作，触发读取并缓存
  aggregates = df.groupBy("partition").max("offset").collect()
  
  # 过滤后写入Parquet，使用缓存的数据
  df.filter(...).write.parquet(target_path)
  
  # 不再使用时释放缓存资源
  df.unpersist()
  

• 先写入临时存储：先把读取到的Kafka数据写入临时表或文件，之后的聚合和写入操作都基于这个临时存储的数据，确保数据源唯一。

• 合并行动操作：通过自定义逻辑，在一次行动操作中同时完成聚合和写入，避免多次触发读取。

内容的提问来源于stack exchange，提问作者G. Bricconi