PySpark 流式 DataFrame 转换为 JSON 格式的实践指南

from pyspark.sql import functions as F
# ... 其他初始化代码

items = df.select('*')

# 错误示范：DataFrameWriter.json() 缺少 'path' 参数
query = (items.writeStream.outputMode("append").foreachBatch(lambda items, epoch_id: items.write.json()).start())

TypeError: DataFrameWriter.json() missing 1 required positional argument: 'path'

import os
from pyspark.sql import DataFrame

def write_batch_to_json(batch_df: DataFrame, batch_id: int, output_base_path: str):
    """
    将每个微批次的 DataFrame 写入到 JSON 文件。
    每个批次会写入到一个独立的子目录中，以避免文件冲突。
    """
    # 构建当前批次的唯一输出路径
    current_batch_output_path = os.path.join(output_base_path, f"batch_{batch_id}")
    print(f"Processing batch {batch_id}, writing to: {current_batch_output_path}")

    # 检查批次是否为空，避免写入空目录或空文件
    if not batch_df.isEmpty():
        # 使用 append 模式，因为每个批次写入的是不同的目录
        batch_df.write.json(current_batch_output_path, mode="append")
    else:
        print(f"Batch {batch_id} is empty, skipping write.")

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.streaming import DataStreamWriter
import os

# 1. 初始化 SparkSession (如果不在 Databricks 环境中，需要手动创建)
# 在 Databricks 环境中，'spark' 对象通常是预先配置好的。
# 如果在本地或其他非 Databricks 环境运行，请取消注释以下行：
# spark = SparkSession.builder \
#     .appName("StreamingToJsonTutorial") \
#     .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
#     .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
#     .getOrCreate()

# 2. 定义流式 DataFrame
# 原始问题中，df 是从 Delta 表读取的流
# table_name = "dev.emp.master_events"
# df = (
#     spark.readStream.format("delta")
#     .option("readChangeFeed", "true") # 如果需要读取 Delta Change Data Feed
#     .option("startingVersion", 2) # 从指定版本开始读取
#     .table(table_name)
# )

# 为了演示和本地测试，我们创建一个模拟的流式 DataFrame
# 它每秒生成一条记录
df = spark.readStream.format("rate").option("rowsPerSecond", 1).load()
items = df.selectExpr("CAST(value AS INT) as id", 
                      "CAST(value % 10 AS STRING) as name", 
                      "CAST(value * 1.0 AS DOUBLE) as value")

# 3. 定义输出基础路径和检查点路径
output_base_path = "/tmp/streaming_json_output" # 请根据实际环境修改
checkpoint_path = os.path.join(output_base_path, "checkpoint")

# 确保输出目录存在 (在实际生产中，通常由 Spark 自动创建或由外部系统管理)
# 但对于本地测试，手动创建可以避免一些权限问题
# import shutil
# if os.path.exists(output_base_path):
#     shutil.rmtree(output_base_path)
# os.makedirs(output_base_path, exist_ok=True)

# 4. 配置并启动流式查询
query = (
    items.writeStream
    .outputMode("append") # 对于 foreachBatch，通常使用 append 模式
    # 使用 functools.partial 传递额外的参数给 write_batch_to_json 函数
    .foreachBatch(lambda batch_df, batch_id: write_batch_to_json(batch_df, batch_id, output_base_path))
    .trigger(processingTime="5 seconds") # 每5秒处理一次微批次
    .option("checkpointLocation", checkpoint_path) # 必须指定检查点目录，用于恢复和容错
    .start()
)

print(f"Streaming query started. Output will be written to: {output_base_path}")
print(f"Checkpoint location: {checkpoint_path}")

# 等待查询终止（例如，按下 Ctrl+C）
query.awaitTermination()

# 如果需要在代码中停止流，可以使用 query.stop()
# query.stop()
# spark.stop() # 停止 SparkSession