如何在Redshift的SUPER列插入无转义字符的JSON数据？

我尝试将API返回的数据以JSON格式存储到Redshift表的SUPER类型列中。

我的请求代码：

data = requests.get(url=f'https://economia.awesomeapi.com.br/json/last/USD-BRL')

我的插入函数如下：

QUERY = f"""INSERT INTO {schema}.{table}(data) VALUES (%s)"""

conn = self.__get_conn()
cursor = conn.cursor()

print('INSERT DATA')
cursor.execute(query=QUERY, vars=([data, ]))
conn.commit()

当使用data.json()插入时，出现错误：psycopg2.ProgrammingError: can't adapt type 'dict'。于是我使用json.dumps(data.json())序列化后插入，但数据库中的数据带有转义字符，示例如下：

"{\"code\": \"USD\", \"codein\": \"BRL\", \"name\": \"Dólar Americano/Real Brasileiro\", \"high\": \"5.2768\", \"low\": \"5.1848\", \"varBid\": \"-0.0264\", \"pctChange\": \"-0.5\", \"bid\":"

我希望在Redshift中使用DBT通过JSON_PARSE()和CTE来结构化该数据集，但这些转义字符造成了阻碍。我忽略了什么？有没有其他实现方式？

表的DDL：

CREATE TABLE IF NOT EXISTS public.raw_currency
(
    id BIGINT  DEFAULT "identity"(105800, 0, '1,1'::text) ENCODE az64
    ,"data" SUPER   ENCODE zstd
    ,stored_at TIMESTAMP WITHOUT TIME ZONE   ENCODE az64
    ,error_log VARCHAR(65535)   ENCODE lzo
)

核心问题 #

Redshift的SUPER类型支持直接存储原生JSON结构，但psycopg2默认无法直接适配Python字典。用json.dumps()插入时，相当于把JSON字符串作为普通文本存入SUPER列，导致出现转义字符，后续JSON_PARSE()无法正确解析。

正确实现方式 #

方式1：在SQL中用JSON_PARSE()转换 #

修改插入语句，将序列化后的JSON字符串传入，并用JSON_PARSE()转换为SUPER类型：

import json

# 获取API返回的字典数据
json_data = data.json()
# 序列化为JSON字符串
json_str = json.dumps(json_data)

QUERY = f"""INSERT INTO {schema}.{table}(data) VALUES (JSON_PARSE(%s))"""

conn = self.__get_conn()
cursor = conn.cursor()
cursor.execute(query=QUERY, vars=(json_str,))
conn.commit()

插入后SUPER列会存储原生JSON结构，无转义字符，后续用DBT处理时可直接访问SUPER属性，无需额外解析。

方式2：使用psycopg2的Json适配器 #

psycopg2提供psycopg2.extras.Json类，可直接将Python字典适配为Redshift可识别的JSON格式：

from psycopg2.extras import Json

# 获取API返回的字典数据
json_data = data.json()

QUERY = f"""INSERT INTO {schema}.{table}(data) VALUES (%s)"""

conn = self.__get_conn()
cursor = conn.cursor()
cursor.execute(query=QUERY, vars=(Json(json_data),))
conn.commit()

这种方式无需手动序列化，适配器自动处理字典到SUPER类型的转换，插入数据为原生JSON结构。

额外优化：填充stored_at字段 #

表中包含stored_at字段，建议插入时自动填充当前时间，可在SQL中使用GETDATE()：

# 配合上面任意一种方式修改插入语句
QUERY = f"""INSERT INTO {schema}.{table}(data, stored_at) VALUES (%s, GETDATE())"""

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