PySpark按数据源类型条件选列生成指定结构JSON如何实现

解决思路 #

• 首先定义不同数据源对应的需保留字段映射，避免输出无关空字段
• 遍历原始全量结果，按数据源类型过滤只保留指定字段，同时自动剔除空值字段
• 最终将处理后的列表套入外层ingestion_connection_details结构后输出到文件

完整实现代码 #

import json

# 定义各数据源对应的保留字段映射，可根据业务需求自行调整
source_field_map = {
    "ORACLE": ["ingestion_connection_name", "ingestion_source", "hostname", "port", "database", "schema", "username"],
    "KAFKA": ["ingestion_connection_name", "ingestion_source", "topicname", "brokers", "startingoffsets"],
    "HDFS": ["ingestion_connection_name", "ingestion_source"]
}

# 原始数据读取逻辑保持不变
results = df.toJSON().map(lambda j: json.loads(j)).collect()

# 逐条处理记录
processed_list = []
for item in results:
    source_type = item.get("ingestion_source", "").strip()
    keep_fields = source_field_map.get(source_type, source_field_map["HDFS"])
    processed_item = {}
    for k in keep_fields:
        val = item.get(k)
        # 仅保留非空值字段，过滤空字符串、None等无效值
        if val not in (None, "", " "):
            # 适配期望输出中kafka字段驼峰命名要求
            if k == "startingoffsets":
                processed_item["startingOffsets"] = val
            else:
                processed_item[k] = val
    # 适配期望输出中hdfs、kafka数据源名称小写的要求
    if source_type == "KAFKA":
        processed_item["ingestion_source"] = "kafka"
    elif source_type == "HDFS":
        processed_item["ingestion_source"] = "hdfs"
    processed_list.append(processed_item)

# 外层套要求的结构
output_data = {
    "ingestion_connection_details": processed_list
}

# 写入文件
with open('/hadoopData/bdipoc/poc/logs/data.json', 'w', encoding='utf-8') as f:
    json.dump(output_data, f, ensure_ascii=False, indent=4)

代码说明 #

• 新增数据源时只需在source_field_map中补充对应保留字段即可，扩展性强
• 自动过滤所有空值字段，不会输出冗余的空字段内容
• 已适配你给出的期望输出中的命名格式要求，如有其他格式调整可直接修改对应转换逻辑即可

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