使用Python在Lambda中解析字段顺序可变JSON适配Redshift COPY

核心问题分析 #

你当前用字符串替换的方式处理JSON对象间的逗号，本质是靠字段开头的特征匹配，一旦JSON字段顺序变化就失效，而且还存在误修改JSON值的风险（比如字段值里包含,{"这类字符）。转CSV时列不匹配，也是因为JSON字段顺序不固定，没有强制按表结构的列顺序生成数据。

可靠的JSON处理方案（推荐） #

直接用Python的json模块解析再重新序列化，完全不受字段顺序影响，还能避免字符串替换的潜在bug：

import json

def process_json_for_redshift(raw_json_str):
    # 1. 把原始的逗号分隔JSON处理成合法的JSON数组
    cleaned_str = raw_json_str.strip()
    # 移除末尾多余的逗号（如果存在）
    if cleaned_str.endswith(','):
        cleaned_str = cleaned_str[:-1]
    # 包裹成JSON数组格式
    json_array = f"[{cleaned_str}]"
    
    # 2. 解析为Python对象列表
    json_objects = json.loads(json_array)
    
    # 3. 生成Redshift需要的无逗号连续JSON格式
    # 或者换成'\n'.join生成JSON Lines格式（每行一个JSON，Redshift COPY更推荐）
    return ''.join(json.dumps(obj) for obj in json_objects)

# 测试示例
raw_data = '''{"person":"Alice", "address":"123 St","state":"CA"},
{"person":"Bob", "address":"456 Ave","state":"NY"},
{"address":"789 Rd", "state":"TX","person":"Charlie"},
'''

processed_data = process_json_for_redshift(raw_data)
print(processed_data)

为什么推荐JSON Lines格式？

如果把代码里的''.join换成'\n'.join，生成每行一个JSON的格式，Redshift COPY时用FORMAT AS JSON 'auto'就能自动匹配表的字段，比连续JSON更易排查问题，也符合JSON Lines的标准规范。

解决CSV列不匹配的问题 #

如果一定要转CSV，必须强制按Redshift表的列顺序提取字段，确保每行数据和列名一一对应：

import csv
from io import StringIO

def json_to_fixed_csv(json_objects, table_columns):
    output = StringIO()
    # 按指定列顺序创建CSV写入器
    writer = csv.DictWriter(output, fieldnames=table_columns)
    # 可选：写入表头，Redshift COPY时可加SKIPHEADER 1跳过
    writer.writeheader()
    
    for obj in json_objects:
        # 按列顺序提取字段，缺失字段用空字符串填充
        row = {col: obj.get(col, '') for col in table_columns}
        writer.writerow(row)
    
    return output.getvalue()

# 调用示例：假设Redshift表列顺序是person, address, state
table_columns = ['person', 'address', 'state']
csv_data = json_to_fixed_csv(json_objects, table_columns)
print(csv_data)

关键注意事项 #

• 永远不要用字符串替换处理JSON结构，json模块是最安全的方式，能避免值里的特殊字符导致的解析错误
• Redshift COPY对JSON的支持很灵活，优先用JSON Lines格式，配置简单且容错性高
• 转CSV时必须绑定固定列顺序，不能依赖JSON的字段顺序

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