如何在PySpark Streaming中按首列值读取不同Schema的CSV文件

当然可以搞定！这种每行结构跟着首列关键字变化的CSV流数据，完全可以通过分阶段处理来实现按不同Schema解析，最后输出到Parquet。下面我给你详细拆解实现步骤和代码示例：

核心思路 #

我们的处理流程分为四步，确保所有行都能被正确读取并按对应规则解析：

• 宽松读取原始流：先把所有行读成无类型的原始字段，避免因Schema不匹配丢失数据
• 按首列拆分数据流：根据left/right/center三个关键字把原始流拆分成三个子流
• 子流单独应用Schema：给每个子流做类型转换和字段映射，匹配对应的结构化Schema
• 合并或分别输出：可以把结构化后的子流对齐Schema合并后统一输出，也可以直接分别写入不同路径

代码示例（Python版） #

1. 初始化SparkSession并定义各类型Schema #

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DoubleType
from pyspark.sql.functions import lit, try_cast

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("DynamicSchemaCSVStreaming") \
    .getOrCreate()

# 定义三种行类型对应的Schema
left_schema = StructType([
    StructField("direction", StringType(), nullable=False),
    StructField("value1", IntegerType(), nullable=False),
    StructField("code", StringType(), nullable=False),
    StructField("id", IntegerType(), nullable=False),
    StructField("ratio", DoubleType(), nullable=False)
])

right_schema = StructType([
    StructField("direction", StringType(), nullable=False),
    StructField("value1", IntegerType(), nullable=False),
    StructField("code", StringType(), nullable=False),
    StructField("id", IntegerType(), nullable=False),
    StructField("ratio", DoubleType(), nullable=False),
    StructField("tag", StringType(), nullable=False),
    StructField("param1", IntegerType(), nullable=False),
    StructField("param2", IntegerType(), nullable=False)
])

center_schema = StructType([
    StructField("direction", StringType(), nullable=False),
    StructField("id", IntegerType(), nullable=False),
    StructField("ratio", DoubleType(), nullable=False),
    StructField("param1", IntegerType(), nullable=False),
    StructField("param2", IntegerType(), nullable=False)
])

2. 读取原始CSV流 #

我们用最宽松的方式读取，不指定Schema，把所有字段读成字符串，同时忽略逗号后的空格：

raw_stream = spark.readStream \
    .option("header", "false") \  # 你的CSV没有表头
    .option("inferSchema", "false") \  # 不自动推断Schema，避免出错
    .option("delimiter", ",") \
    .option("ignoreLeadingWhiteSpace", "true") \  # 去掉逗号后的空格，比如" 10"变成"10"
    .csv("/path/to/your/input/directory")

此时raw_stream的列名为_c0、_c1、_c2...对应每行的各个字段。

3. 拆分并解析每个子流 #

针对每个关键字对应的行，我们做类型转换和字段映射：

# 处理left类型的行，用try_cast避免类型转换失败导致报错
left_stream = raw_stream.filter(raw_stream._c0 == "left") \
    .select(
        "_c0".alias("direction"),
        try_cast("_c1", IntegerType()).alias("value1"),
        "_c2".alias("code"),
        try_cast("_c3", IntegerType()).alias("id"),
        try_cast("_c4", DoubleType()).alias("ratio")
    )

# 处理right类型的行
right_stream = raw_stream.filter(raw_stream._c0 == "right") \
    .select(
        "_c0".alias("direction"),
        try_cast("_c1", IntegerType()).alias("value1"),
        "_c2".alias("code"),
        try_cast("_c3", IntegerType()).alias("id"),
        try_cast("_c4", DoubleType()).alias("ratio"),
        "_c5".alias("tag"),
        try_cast("_c6", IntegerType()).alias("param1"),
        try_cast("_c7", IntegerType()).alias("param2")
    )

# 处理center类型的行
center_stream = raw_stream.filter(raw_stream._c0 == "center") \
    .select(
        "_c0".alias("direction"),
        try_cast("_c1", IntegerType()).alias("id"),
        try_cast("_c2", DoubleType()).alias("ratio"),
        try_cast("_c3", IntegerType()).alias("param1"),
        try_cast("_c4", IntegerType()).alias("param2")
    )

4. 对齐Schema并合并输出（可选） #

如果需要把所有数据写入同一个Parquet路径，需要先对齐三个流的Schema，补全缺失字段为null：

# 给left_stream补全right和center的额外字段
left_aligned = left_stream \
    .withColumn("tag", lit(None).cast(StringType())) \
    .withColumn("param1", lit(None).cast(IntegerType())) \
    .withColumn("param2", lit(None).cast(IntegerType()))

# 给center_stream补全left和right的额外字段
center_aligned = center_stream \
    .withColumn("value1", lit(None).cast(IntegerType())) \
    .withColumn("code", lit(None).cast(StringType()))

# 合并三个流
unified_stream = left_aligned.unionByName(right_stream).unionByName(center_aligned)

5. 写入Parquet流 #

最后把处理好的流写入Parquet，记得设置checkpoint路径用于故障恢复：

# 写入统一的Parquet路径
query = unified_stream.writeStream \
    .format("parquet") \
    .option("path", "/path/to/your/output/parquet") \
    .option("checkpointLocation", "/path/to/your/checkpoint/directory") \  # 必须指定，用于恢复
    .outputMode("append") \  # 流式数据常用append模式
    .start()

# 如果是分别输出三个子流，直接各自调用writeStream即可
# left_query = left_stream.writeStream(...).start()
# right_query = right_stream.writeStream(...).start()
# center_query = center_stream.writeStream(...).start()

query.awaitTermination()

关键注意事项 #

• Checkpoint路径：Structured Streaming必须指定checkpointLocation，建议用分布式存储（比如HDFS、S3），不要用本地路径，否则集群环境下会出问题。
• 类型转换容错：用try_cast替代普通cast，这样当字段无法转换成目标类型时会返回null，而不是直接导致任务失败。
• 空格处理：CSV中逗号后有空格，一定要开启ignoreLeadingWhiteSpace，避免字段值带前导空格影响后续处理。
• 输出模式选择：如果是追加新数据，用append模式；如果需要更新已有数据，根据业务场景选择update或complete模式。

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