如何使用pandas.to_sql将dict列写入Redshift的SUPER类型列

问题根因 #

你使用sqlalchemy.types.JSON做类型映射时，SQLAlchemy默认会将JSON类数据序列化为带转义符的普通字符串传入INSERT语句，未触发Redshift要求的JSON_PARSE函数调用，因此SUPER列会存储带转义的字符串而非原生JSON结构。

解决方案1：自定义SQLAlchemy类型适配（无需手动编写全量INSERT语句） #

你可以通过自定义SQLAlchemy类型装饰器，在插入数据时自动调用JSON_PARSE函数处理目标列的值，完全兼容pandas.DataFrame.to_sql的用法，示例代码如下：

1. 首先定义Redshift SUPER类型的自定义适配类

from sqlalchemy import TypeDecorator, String, func

class RedshiftSUPER(TypeDecorator):
    impl = String
    cache_ok = True

    def bind_expression(self, bindvalue):
        # 插入时自动对值套JSON_PARSE函数
        return func.JSON_PARSE(bindvalue)

    def process_result_value(self, value, dialect):
        # 查询时自动将SUPER类型返回值转成Python原生dict/list
        import json
        if value is not None:
            return json.loads(value)
        return value

2. 调整类型映射和DataFrame预处理逻辑

# 预处理：将目标列的JSON对象转成未转义的JSON字符串
import json
df['my_json_column'] = df['my_json_column'].apply(lambda x: json.dumps(x, ensure_ascii=False))

# 替换类型映射为自定义的RedshiftSUPER
type_dict = {
    'my_json_column': RedshiftSUPER
}

# 正常调用to_sql即可
df.to_sql('table', connection, schema='my_schema', if_exists='append', dtype=type_dict)

解决方案2：大数据量场景推荐使用COPY方式 #

如果插入的数据量较大，COPY命令的性能远高于逐行INSERT，你可以先将DataFrame导出为jsonlines格式文件，上传到S3后直接调用Redshift COPY命令导入SUPER列，原生支持JSON结构解析，无需额外处理转义问题。

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