如何将JSON文件数据导入本地Docker Compose部署的PostgreSQL？

前提准备 #

首先确保你的Docker Compose PostgreSQL容器处于运行状态，并且能通过docker exec或客户端工具连接到数据库。先创建匹配数据结构的用户表：

CREATE TABLE users (
    user_id INT PRIMARY KEY,
    name VARCHAR(50) NOT NULL,
    last_name VARCHAR(50) NOT NULL,
    gender VARCHAR(10),
    age INT,
    birth_day VARCHAR(5),
    city VARCHAR(20),
    ip INET,
    time_created TIMESTAMP NOT NULL,
    premium BOOLEAN,
    balance NUMERIC(15,10)
);

方法一：纯SQL + Docker命令导入 #

情况1：JSON文件为JSON Lines格式（每行一个用户对象） #

• 将本地JSON文件复制到PostgreSQL容器内（假设容器名为postgres-db，目标路径/tmp/users.json）：

docker cp /本地路径/users.json postgres-db:/tmp/users.json

• 进入容器的psql终端（替换your_db、your_user为实际数据库名和用户名）：

docker exec -it postgres-db psql -U your_user -d your_db

• 执行COPY命令导入数据，同时转换时间戳字段：

COPY users (user_id, name, last_name, gender, age, birth_day, city, ip, time_created, premium, balance)
FROM '/tmp/users.json'
FORMAT JSON;

-- 转换Unix时间戳为标准时间格式
UPDATE users SET time_created = TO_TIMESTAMP(time_created);

情况2：JSON文件为数组格式（整个文件是包含所有用户对象的数组） #

在psql终端中使用json_populate_recordset函数批量导入：

INSERT INTO users
SELECT * FROM json_populate_recordset(NULL::users, (SELECT pg_read_file('/tmp/users.json')::json));

-- 转换Unix时间戳为标准时间格式
UPDATE users SET time_created = TO_TIMESTAMP(time_created);

方法二：Python脚本导入 #

步骤1：安装依赖 #

pip install psycopg2-binary

步骤2：编写导入脚本 #

import json
import psycopg2
from psycopg2.extras import execute_values

# 数据库连接参数（根据你的Docker Compose配置修改）
DB_PARAMS = {
    'dbname': 'your_db',
    'user': 'your_user',
    'password': 'your_password',
    'host': 'localhost',  # 容器端口映射到本地用localhost，否则用容器名
    'port': '5432'
}

# 读取JSON数据
def load_json_data(file_path):
    with open(file_path, 'r') as f:
        # 若为JSON Lines格式，用下面的代码
        # data = [json.loads(line) for line in f if line.strip()]
        # 若为数组格式，用下面的代码
        data = json.load(f)
    return data

# 批量插入数据
def insert_users(data):
    conn = None
    try:
        conn = psycopg2.connect(**DB_PARAMS)
        cur = conn.cursor()

        # 插入语句，自动处理主键冲突
        insert_query = """
            INSERT INTO users (user_id, name, last_name, gender, age, birth_day, city, ip, time_created, premium, balance)
            VALUES %s
            ON CONFLICT (user_id) DO NOTHING
        """

        # 转换数据格式，将Unix时间戳转为PostgreSQL时间类型
        records = [
            (
                user['user_id'],
                user['name'],
                user['last_name'],
                user['gender'],
                user['age'],
                user['birth_day'],
                user['city'],
                user['ip'],
                psycopg2.Timestamp.fromtimestamp(user['time_created']),
                user['premium'],
                user['balance']
            )
            for user in data
        ]

        # 批量插入提升效率
        execute_values(cur, insert_query, records)
        conn.commit()
        print(f"成功插入{len(records)}条用户数据")

    except Exception as e:
        print(f"导入失败: {str(e)}")
        if conn:
            conn.rollback()
    finally:
        if conn:
            cur.close()
            conn.close()

if __name__ == "__main__":
    user_data = load_json_data('/本地路径/users.json')
    insert_users(user_data)

运行脚本 #

python import_users.py

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