PostgreSQL超万列CSV数据高效管理：JSONB方案详解

CREATE TABLE large_csv_data (
    record_id SERIAL PRIMARY KEY,
    site_id VARCHAR(50) NOT NULL,
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    -- 其他核心或频繁查询的列
    -- ...
    -- 存储所有次要或不常用列的JSONB字段
    additional_attributes JSONB
);

import csv
import json
import psycopg2

# 假设的核心列和次要列列表
CORE_COLUMNS = ['record_id', 'site_id', 'col_core_1', 'col_core_2']
# 假设的CSV文件路径
CSV_FILE_PATH = 'your_large_data.csv'

def import_csv_to_postgresql(csv_file_path, db_connection_string):
    conn = psycopg2.connect(db_connection_string)
    cur = conn.cursor()

    with open(csv_file_path, 'r', encoding='utf-8') as f:
        reader = csv.reader(f)
        header = next(reader) # 读取CSV头部

        # 确定次要列的索引
        additional_col_indices = [i for i, col_name in enumerate(header) if col_name not in CORE_COLUMNS]
        # 确定核心列的索引
        core_col_indices = [i for i, col_name in enumerate(header) if col_name in CORE_COLUMNS]
        core_col_names_ordered = [col_name for col_name in header if col_name in CORE_COLUMNS]


        for row in reader:
            core_data = {header[i]: row[i] for i in core_col_indices}
            additional_data = {header[i]: row[i] for i in additional_col_indices if row[i]} # 只存储非空值

            # 准备SQL插入语句
            # 注意：record_id通常由数据库序列生成，这里假设不从CSV直接取
            # 实际情况可能需要调整SQL语句和core_data的结构
            insert_sql = f"""
                INSERT INTO large_csv_data ({', '.join(core_col_names_ordered)}, additional_attributes)
                VALUES ({', '.join(['%s'] * len(core_col_names_ordered))}, %s);
            """

            # 准备插入值
            values = [core_data[col_name] for col_name in core_col_names_ordered] + [json.dumps(additional_data)]

            try:
                cur.execute(insert_sql, values)
            except Exception as e:
                print(f"Error inserting row: {row}. Error: {e}")
                conn.rollback()
                continue

    conn.commit()
    cur.close()
    conn.close()
    print("Data import completed.")

# 示例调用
# db_conn_str = "dbname=your_db user=your_user password=your_password host=your_host port=your_port"
# import_csv_to_postgresql(CSV_FILE_PATH, db_conn_str)

-- 查询 record_id, site_id 和 additional_attributes 中名为 'col_A' 的值
SELECT
    record_id,
    site_id,
    additional_attributes ->> 'col_A' AS column_A_value,
    additional_attributes -> 'col_B' AS column_B_raw_json -- 获取原始JSON值，可能包含嵌套
FROM
    large_csv_data
WHERE
    site_id = 'site_X';

-- 查找 additional_attributes 中 'col_C' 值为 'target_value' 的记录
SELECT
    record_id,
    site_id,
    additional_attributes ->> 'col_C' AS column_C_value
FROM
    large_csv_data
WHERE
    additional_attributes ->> 'col_C' = 'target_value';

-- 查找 additional_attributes 中包含键 'col_D' 的记录
SELECT
    record_id,
    site_id
FROM
    large_csv_data
WHERE
    additional_attributes ? 'col_D'; -- 检查键是否存在

-- 假设 additional_attributes 结构为 {"settings": {"theme": "dark"}}
SELECT
    record_id,
    additional_attributes -> 'settings' ->> 'theme' AS user_theme
FROM
    large_csv_data
WHERE
    additional_attributes -> 'settings' ->> 'theme' = 'dark';

-- 创建一个用于查找键或键值对的GIN索引
CREATE INDEX idx_large_csv_data_additional_attributes_gin
ON large_csv_data USING GIN (additional_attributes);

-- 如果需要更精确地匹配包含特定键值对的JSON对象，可以使用 jsonb_path_ops 操作符类
-- CREATE INDEX idx_large_csv_data_additional_attributes_path_ops
-- ON large_csv_data USING GIN (additional_attributes jsonb_path_ops);