PostgreSQL(Timescale)多表关联动态行转列生成CSV方案咨询

核心方案 #

用PostgreSQL内置的crosstab行转列函数结合动态SQL自动生成所有饮品列，完全替代手动编写大量UNION语句，同时适配Timescale超表的时序特性。

步骤1：确保依赖扩展已安装 #

crosstab属于tablefunc扩展，先执行以下SQL启用：

CREATE EXTENSION IF NOT EXISTS tablefunc;

步骤2：编写动态交叉表查询 #

通过SQL自动拼接所有饮品名称作为列，生成完整的行转列查询语句：

WITH drink_list AS (
    -- 自动拼接所有饮品名称，用quote_ident处理特殊字符（空格/引号等）
    SELECT string_agg(DISTINCT quote_ident(name), ', ') AS cols
    FROM drinks
),
dynamic_crosstab AS (
    SELECT format(
        'SELECT * FROM crosstab(
            ''-- 基础查询：关联消费数据与饮品表，按时间+饮品排序
             SELECT cd.time, d.name, cd.consumption_value
             FROM consumption_data cd
             JOIN drinks d ON cd.drink_id = d.id
             ORDER BY 1, 2'',
            ''-- 指定所有饮品列的顺序
             SELECT DISTINCT name FROM drinks ORDER BY 1''
        ) AS ct(time TIMESTAMPTZ, %s)',
        cols
    ) AS query
    FROM drink_list
)
-- 输出最终可执行的交叉表SQL
SELECT query FROM dynamic_crosstab;

适配Timescale时序聚合（可选）

如果需要按时间桶（小时/天）聚合消费数据，修改基础查询部分：

SELECT time_bucket('1 hour', cd.time) AS bucket_time, d.name, SUM(cd.consumption_value) AS total_consumption
FROM consumption_data cd
JOIN drinks d ON cd.drink_id = d.id
GROUP BY bucket_time, d.name
ORDER BY bucket_time, d.name

步骤3：执行查询并导出CSV #

方式1：服务器端直接导出（需PostgreSQL有权限访问路径）

用动态SQL生成COPY命令并执行：

DO $$
DECLARE
    dynamic_copy text;
BEGIN
    WITH drink_list AS (
        SELECT string_agg(DISTINCT quote_ident(name), ', ') AS cols
        FROM drinks
    )
    SELECT format(
        'COPY (
            SELECT * FROM crosstab(
                ''SELECT cd.time, d.name, cd.consumption_value
                 FROM consumption_data cd
                 JOIN drinks d ON cd.drink_id = d.id
                 ORDER BY 1, 2'',
                ''SELECT DISTINCT name FROM drinks ORDER BY 1''
            ) AS ct(time TIMESTAMPTZ, %s)
        ) TO ''/path/to/your/output.csv'' WITH CSV HEADER;',
        cols
    ) INTO dynamic_copy
    FROM drink_list;

    EXECUTE dynamic_copy;
END $$;

方式2：Django客户端导出（避免服务器路径权限问题）

通过Django原生游标执行动态查询，在本地生成CSV：

from django.db import connection
import csv

def export_drink_consumption():
    with connection.cursor() as cursor:
        # 启用tablefunc扩展（仅首次执行需运行）
        cursor.execute("CREATE EXTENSION IF NOT EXISTS tablefunc;")

        # 获取动态交叉表查询语句
        cursor.execute("""
            WITH drink_list AS (
                SELECT string_agg(DISTINCT quote_ident(name), ', ') AS cols
                FROM drinks
            )
            SELECT format(
                'SELECT * FROM crosstab(
                    ''SELECT cd.time, d.name, cd.consumption_value
                     FROM consumption_data cd
                     JOIN drinks d ON cd.drink_id = d.id
                     ORDER BY 1, 2'',
                    ''SELECT DISTINCT name FROM drinks ORDER BY 1''
                ) AS ct(time TIMESTAMPTZ, %s)',
                cols
            ) AS query
            FROM drink_list;
        """)
        dynamic_query = cursor.fetchone()[0]

        # 执行查询并获取结果
        cursor.execute(dynamic_query)
        rows = cursor.fetchall()
        columns = [desc[0] for desc in cursor.description]

        # 写入本地CSV
        with open('/your/local/path/output.csv', 'w', newline='', encoding='utf-8') as f:
            writer = csv.writer(f)
            writer.writerow(columns)
            writer.writerows(rows)

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