Synapse Notebook中PySpark保存DataFrame到ADLS动态路径遇错

错误原因 #

你遇到的TypeError: Column is not iterable是因为save()方法仅接受静态字符串路径，不能直接传入Spark的Column对象或concat()这类列表达式——这些是分布式计算中针对数据行的操作，无法作为写入路径的参数传入。

解决方案 #

要实现YYYY/MM/YYYY-MM-DD.parquet的文件夹结构，需要先从created_date提取日期维度字段，再通过按日期分组写入的方式实现：

步骤1：提取日期维度列 #

先从created_date字段中解析出年、月、完整日期的字符串格式（假设created_date是date或timestamp类型）：

from pyspark.sql.functions import date_format

# 添加年、月、日期字符串列
df_with_date_parts = df1.withColumn("year", date_format("created_date", "yyyy")) \
                        .withColumn("month", date_format("created_date", "MM")) \
                        .withColumn("date_str", date_format("created_date", "yyyy-MM-dd"))

步骤2：按日期分组写入目标路径 #

收集所有唯一日期，循环过滤对应日期的数据并写入指定路径：

# 收集所有唯一的日期字符串
unique_dates = [row.date_str for row in df_with_date_parts.select("date_str").distinct().collect()]

# 定义ADLS根路径
root_path = "<file path to ADLS>"

for date_str in unique_dates:
    # 过滤当前日期的数据
    daily_data = df_with_date_parts.filter(df_with_date_parts.date_str == date_str)
    # 拆分年和月
    year_part, month_part = date_str.split("-")[0], date_str.split("-")[1]
    # 构建最终路径
    target_path = f"{root_path}/{year_part}/{month_part}/{date_str}.parquet"
    # 写入Parquet文件
    daily_data.write.format("parquet").mode("overwrite").save(target_path)

替代方案（批量分区写入，需调整目录结构） #

如果希望用Spark的批量分区写入优化性能，可先按年、月分区生成year=YYYY/month=MM的结构，再通过ADLS操作重命名目录去掉前缀：

# 关闭分区列类型推断，让分区目录直接使用值（需配合特定输出提交器）
spark.conf.set("spark.sql.sources.partitionColumnTypeInference.enabled", "false")
spark.conf.set("spark.sql.parquet.output.committer.class", "org.apache.spark.sql.parquet.DirectParquetOutputCommitter")

# 按年、月分区写入根路径
df_with_date_parts.write.format("parquet") \
                  .mode("overwrite") \
                  .partitionBy("year", "month") \
                  .save(root_path)

之后可通过Synapse的ADLS文件管理或脚本，将year=2024重命名为2024，month=05重命名为05，再将每个分区下的文件统一命名为对应日期的YYYY-MM-DD.parquet。

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