PostgreSQL 18 如何导入大型 CSV 并自动生成列名、识别类型？

2026-11-04数据阿明

可以用脚本读取 CSV 表头并创建表，免去手写上百个列名的工作。PostgreSQL 18 的 COPY 要求目标表已经存在，不会根据 CSV 自动建表、推断列类型或建立两表关系。对于类型未知的大文件，可以先自动创建列类型全部为 text 的暂存表，导入后再确认类型和关联键。

为什么 `HEADER` 不能自动建表？

COPY ... WITH (FORMAT csv, HEADER true) 只会跳过输入文件的表头。HEADER MATCH 可以检查表头是否与目标表的列名、数量和顺序一致，但不会创建列，也不会根据表头名称重新排列数据。详见 PostgreSQL 18 COPY 文档。

生成列名和推断数据类型需要分别处理。表头能提供名称，但 CSV 通常没有完整的类型定义。例如，001234 可能是需要保留前导零的人员编号；03/04/2000 也无法单凭字符串确定日期顺序。抽样可以推断候选类型，却不能保证后续几百万行都符合该类型。

如何自动建表并流式导入？

下面的示例使用 Python 和 Psycopg 3，读取表头生成列定义，再通过 COPY FROM STDIN 分块传输文件，无需一次将整个 CSV 装入内存。Psycopg 支持这种分块 COPY 写入方式。

示例假定文件有表头、采用 UTF-8 编码、以逗号分隔，并使用标准双引号引用字段；目标数据库也使用 UTF-8。如果文件采用其他格式，需要相应调整读取和 COPY 配置。

安装依赖：

python -m pip install "psycopg[binary]"

将以下代码保存为 import_csv.py：

import csv
import json
import sys

import psycopg
from psycopg import sql


def import_csv(dsn, path, table_name):
    with open(path, encoding="utf-8-sig", newline="") as source:
        header = next(csv.reader(source), None)
        if not header:
            raise ValueError("CSV is empty or has no header")
        if len(header) > 1600:
            raise ValueError("Too many columns for one PostgreSQL table")

        names = []
        used = set()
        for i, label in enumerate(header, start=1):
            base = label.strip()
            if "\x00" in base:
                raise ValueError("NUL character in header")
            if not base or len(base.encode("utf-8")) > 63:
                base = f"column_{i}"
            name = base
            suffix = 1
            while name in used:
                name = f"column_{i}_{suffix}"
                suffix += 1
            used.add(name)
            names.append(name)

        table = sql.Identifier("public", table_name)
        columns = sql.SQL(", ").join(
            sql.SQL("{} text").format(sql.Identifier(name))
            for name in names
        )

        source.seek(0)
        with psycopg.connect(dsn) as conn:
            with conn.cursor() as cur:
                cur.execute(
                    sql.SQL("CREATE TABLE {} ({})").format(table, columns)
                )
                command = sql.SQL(
                    "COPY {} FROM STDIN WITH (FORMAT csv, HEADER true)"
                ).format(table)
                with cur.copy(command) as copy:
                    while block := source.read(1024 * 1024):
                        copy.write(block)

        print(json.dumps(
            list(zip(header, names)), ensure_ascii=False
        ))


if __name__ == "__main__":
    import_csv(*sys.argv[1:])

分别导入两个文件。下面的数据库名、用户名和文件名都是示例，需替换为实际值；连接认证可以使用本机的 PostgreSQL 配置：

python import_csv.py "dbname=mydb user=myuser" voters.csv voters_raw
python import_csv.py "dbname=mydb user=myuser" voting_history.csv voting_history_raw

脚本会保留可用的表头名称，为空列名、过长列名和重复名称生成替代名称，并输出原始名称与数据库列名的对应关系。建议保存这份映射。sql.Identifier 负责正确引用标识符，避免表头中的空格、引号或 SQL 关键字破坏建表语句。

每个文件的建表和导入都在同一事务中完成，失败时会回滚。如果目标表已存在，脚本会报错；重新导入时应另选表名或明确处理已有数据。

导入后如何确定类型？

先检查数据，再创建用于查询的正式表，通常比直接修改唯一一份原始数据更方便核对。

人员编号、邮政编码等字段即使只包含数字，也可能需要保留为 text。
出生日期需要确认格式，并检查空值、非法日期和占位值，再转换为 date。
计数或金额等字段应检查全部非空值，不能只依赖前几行样本。

默认 CSV 导入规则会把未加引号的空字段解释为 SQL NULL，而 "" 表示空字符串。如果数据提供方使用其他空值标记，应先确认其含义，再配置 NULL 选项，避免把正常文本误当成缺失值。相关规则见 COPY 的 CSV 格式说明。

两个文件怎样关联？

导入过程不会自动发现业务关系。需要先确认两个文件是否含有同一个稳定的人员标识，并检查选民表中的该标识是否唯一。姓名和地址可能重复或变化，不宜直接假定为唯一关联键。

假设两个文件确实都有 voter_id，可以先检查选民表中的重复值：

SELECT voter_id, count(*)
FROM public.voters_raw
GROUP BY voter_id
HAVING count(*) > 1;

确认字段含义、空值和唯一性后，再创建索引并关联。以下列名仅作示例，应换成实际导入后的名称：

CREATE INDEX voters_raw_voter_id_idx
    ON public.voters_raw (voter_id);

CREATE INDEX voting_history_raw_voter_id_idx
    ON public.voting_history_raw (voter_id);

ANALYZE public.voters_raw;
ANALYZE public.voting_history_raw;

SELECT v.voter_id, h.*
FROM public.voters_raw AS v
JOIN public.voting_history_raw AS h
    ON h.voter_id = v.voter_id
LIMIT 100;

同一人员有多条投票历史时，连接结果出现多行是正常的一对多关系。是否添加主键或外键，应在验证数据后决定。

SQLite 中转方案是否可行？

SQLite 可以完成自动建表：目标表不存在时，.import 会使用第一条记录作为列名；目标表已存在时，则默认把第一条记录也作为数据导入。详见 SQLite CLI 文档。

不过，.dump 导出的是 SQLite 的 SQL，不能保证整个文件可以原样在 PostgreSQL 中执行。可能需要处理 SQLite 专用语句、类型和转义表达式等兼容问题；它也不会替你确认 PostgreSQL 中合适的业务类型或关联键。SQLite 可以用于中间检查，但如果只是为了自动建表而增加一次完整导入、导出，收益有限。

如果确实需要 SQLite 中转，建议启动时指定磁盘数据库：

sqlite3 staging.db

原方案只运行 sqlite3，默认使用临时内存数据库，退出后数据会消失。对于 GiB 级文件，应避免无意中采用这一方式。参见 SQLite 启动说明。

关于 pgfutter，题目中的项目停滞说法不足以判断它在当前环境中的可靠性；上述方案不依赖它。

大文件导入需要注意什么？

接近 1 GiB 和 3.5 GiB 的文件不一定需要拆分，但要为数据库数据、WAL、索引及后续转换预留磁盘空间。不能按 CSV 总大小直接估算最终占用，也无法仅凭文件大小承诺导入时间。

应先确认第二个文件的真实列数。PostgreSQL 单表最多支持 1,600 列，实际还受行宽限制；默认标识符长度上限为 63 字节。详见 PostgreSQL 18 限制说明。如果历史数据确实按每次选举不断增加列，可以考虑改用包含“人员标识、选举标识、投票记录”等字段的多行结构，避免持续扩展宽表。

全列使用 text 能减少类型推断错误，但无法绕过编码错误、CSV 引号不匹配或字段数量不一致的问题。正式导入前，应先用一份完整记录组成的小样本确认格式；导入后再核对记录数、关键字段缺失和关联覆盖情况。