PySpark中writeStream无法写入指定路径的技术问询

问题背景 #

环境：Spark 3.3.2、Python 3.10
从Kafka主题读取流式数据，转换后尝试用writeStream写入Hudi到磁盘，初始代码无数据输出；改用foreachBatch内部调用批量写入后可以正常输出，但质疑这种方式是否属于真正的流式写入。

初始无输出代码 #

# read data from topic
df = spark \
     .readStream \
     .format("kafka") \
     .option("kafka.bootstrap.servers", "my_broker:9092") \
     .option("subscribe", "my_topic") \
     .option("starting_offsets", "earliest")  # 修正原代码的语法错误
     .options(**{  # 修正原代码的options用法错误
                "maxOffsetsPerTrigger": 10000,
                "failOnDataLoss": "false"}) \
     .load()

# convert the byte array into a dict/Map object
df1 = df.selectExpr("cast(value as string) as jsonString").select(from_json(col("jsonString"), schema).alias("json"))

# write dataframe to disk
df1.writeStream \
   .foreachBatch(my_func)\
   .format("hudi")\
   .option("hoodie.table.name", "my_table")\
   .option("hoodie.datasource.write.precombine.field", "precombine_key")\
   .option("hoodie.datasource.write.recordkey.field", "record_key")\
   .option("path", "/path/to/output/")\
   .option("checkpointLocation", "/path/to/checkpoint/")\
   .trigger(processingTime="1 minutes")\
   .start()\
   .awaitTermination() 


def my_func(batch_df, batch_id):
    # process the JSON into tabular format and persist to disk
    processed_df = batch_df.select(
            col("json").getItem("payload").getItem("record_key").alias("id"),
            col("json").getItem("payload").getItem("precombine_key").alias("name"),
            explode(col("json").getItem("payload")).alias("keys")
        )
    return processed_df

修改后可正常输出代码 #

# write dataframe to disk
df1.writeStream \
   .foreachBatch(my_func)\
   .trigger(processingTime="1 minutes")\
   .start()\
   .awaitTermination() 


def my_func(batch_df, batch_id):
    # process the JSON into tabular format and persist to disk
    processed_df = batch_df.select(
            col("json").getItem("payload").getItem("id").alias("id"),
            col("json").getItem("payload").getItem("name").alias("name"),
            explode(col("json").getItem("payload")).alias("keys")
        )
    processed_df.write\
                .format("hudi")\
                .option("hoodie.table.name", "my_table")\
                .option("hoodie.datasource.write.precombine.field", "precombine_key")\
                .option("hoodie.datasource.write.recordkey.field", "record_key")\
                .option("path", "/path/to/output/")\
                .option("checkpointLocation", "/path/to/checkpoint/")\
                .mode("append")\
                .save()

核心误区解析 #

• foreachBatch与format配置的冲突：一旦调用.foreachBatch()，后续的.format()、写入相关option都会被Spark忽略。foreachBatch是自定义微批处理逻辑的入口，Spark会完全按照该函数内的逻辑处理每个微批，不会再执行后续指定的流式写入流程。初始代码中my_func仅返回转换后的DataFrame但未执行写入，自然无数据输出。
• 对"流式写入"的误解：Structured Streaming默认采用微批处理模型，所谓的流式写入本质是Spark按触发间隔自动处理每个微批数据，并自动管理偏移量、检查点以保证语义一致性。foreachBatch是官方推荐的、针对不支持直接流式Sink的数据源的适配方案，完全属于Structured Streaming的流式处理体系，并非"违背初衷"。
• foreachBatch函数的逻辑错误：该函数不需要返回值，而是要在函数内部完成数据落地操作。初始代码中仅返回处理后的DataFrame，Spark不会自动处理这个返回结果。

正确的实现方式 #

方式1：直接使用Hudi流式Sink（推荐，需Hudi版本支持） #

如果使用的Hudi版本（如0.10+）支持Spark 3.x的Structured Streaming直接写入，可省略foreachBatch，直接将转换后的DataFrame通过writeStream写入：

# 先完成数据转换
processed_df = df1.select(
    col("json").getItem("payload").getItem("record_key").alias("id"),
    col("json").getItem("payload").getItem("precombine_key").alias("name"),
    explode(col("json").getItem("payload")).alias("keys")
)

# 直接流式写入Hudi
processed_df.writeStream \
   .format("hudi") \
   .option("hoodie.table.name", "my_table") \
   .option("hoodie.datasource.write.precombine.field", "precombine_key") \
   .option("hoodie.datasource.write.recordkey.field", "record_key") \
   .option("path", "/path/to/output/") \
   .option("checkpointLocation", "/path/to/checkpoint/") \
   .trigger(processingTime="1 minutes") \
   .start() \
   .awaitTermination()

注意：需确保Spark环境引入了对应版本的Hudi依赖包，例如通过spark-submit添加：

--packages org.apache.hudi:hudi-spark3.3-bundle_2.12:0.13.1

方式2：正确使用foreachBatch（兼容旧版Hudi） #

修改后的代码是合理的流式实现，因为Spark会自动管理每个微批的偏移量和检查点，保证Exactly-Once语义，区别于手动批量写入的核心点在于：Structured Streaming会自动处理触发、偏移量提交、故障恢复等流式生命周期管理，而foreachBatch只是将每个微批的写入逻辑交给用户自定义。

补充语法修正 #

初始代码中的两处语法错误也会导致运行问题：

• 原代码.option("starting_offsets": "earliest") 应改为 .option("starting_offsets", "earliest")（参数传递用逗号而非冒号）
• 原代码.options("additional_options": {...}) 应改为 .options(**{"maxOffsetsPerTrigger": 10000, "failOnDataLoss": "false"})（使用关键字参数展开字典）

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