如何在PySpark中将ADLS中多个JSON文件转为单个CSV（不使用Pandas）

完全可行，PySpark天生适合处理分布式存储（如ADLS）中的批量文件，225000条记录的规模对PySpark来说毫无压力。以下是具体实现步骤和代码示例：

核心步骤 #

• 配置Spark与ADLS的访问权限
• 批量读取ADLS中的JSON文件
• （可选）处理嵌套JSON结构（API返回的JSON常存在嵌套）
• 将数据合并为单个CSV文件写入ADLS

代码示例 #

from pyspark.sql import SparkSession

# 初始化SparkSession，配置ADLS访问密钥
spark = SparkSession.builder \
    .appName("JSON-to-CSV") \
    .config("fs.azure.account.key.<你的存储账户名>.dfs.core.windows.net", "<你的访问密钥>") \
    .getOrCreate()

# 批量读取ADLS路径下的所有JSON文件（支持通配符匹配）
json_df = spark.read.json("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/json_files_dir/*.json")

# 若JSON存在嵌套结构，可通过selectExpr展开字段（示例）
# json_df = json_df.selectExpr(
#     "id as record_id",
#     "api_data.user.name as user_name",
#     "api_data.timestamp as request_time"
# )

# 合并为单个分区并写入CSV（coalesce(1)避免小文件，适合数据量不大的场景）
json_df.coalesce(1) \
    .write \
    .mode("overwrite") \
    .option("header", "true") \
    .csv("abfss://<容器名>@<存储账户名>.dfs.core.windows.net/output_csv_dir/")

注意事项 #

1. 权限配置：确保Spark环境已正确配置ADLS的访问凭证（密钥、SAS或服务主体均可），否则会出现访问拒绝错误。
2. JSON结构一致性：若多个JSON文件结构不一致，可添加mode="PERMISSIVE"参数让Spark兼容读取，缺失字段会填充为null。
3. 输出文件名：Spark写入CSV时会生成part-xxxx.csv格式的文件及少量元数据文件，可手动将part文件重命名为目标文件名，或通过ADLS的文件操作API批量处理。
4. 性能优化：22万条记录用coalesce(1)足够高效，若数据量超千万级，可考虑先分区处理再合并，避免单分区压力过大。

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