如何将 json 文件复制到 postgres？

Question

我想将 json 数据导入 postgres。我拥有的数据大约为一百万行，大小至少为 700 MB，并且一直延伸到 3 GB。

这是我根据我拥有的数据结构创建的示例数据。我尝试将其导入 postgres，但出现错误。

示例 (1) 数据

{"offers":{"offer":[
{"url": "https://some1-value.com", "nested": {"id":4,"value":"some1 text value"}, "quotes": "5\"  side"},
{"url": "https://some2-value.com", "nested": {"id":5,"value":"some2 text value"}, "quotes": "6\"  side"},
{"url": "https://some3-value.com", "nested": {"id":6,"value":"some3 text value"}, "quotes": "7\"  side"}]}}

我使用的命令和我得到的错误

# copy contrial from '/home/ubuntu/sample-data.json';
ERROR:  invalid input syntax for type json
DETAIL:  The input string ended unexpectedly.
CONTEXT:  JSON data, line 1: {"offers":{"offer":[
COPY contrial, line 1, column info: "{"offers":{"offer":["

我修改了文件以删除前两个键，并且只有一个如下所示的 json 列表，但我仍然收到错误。

示例 (2) 数据

[
{"url": "https://some1-value.com", "nested": {"id":4,"value":"some1 text value"}, "quotes": "5\"  side"},
{"url": "https://some2-value.com", "nested": {"id":5,"value":"some2 text value"}, "quotes": "6\"  side"},
{"url": "https://some3-value.com", "nested": {"id":6,"value":"some3 text value"}, "quotes": "7\"  side"}]

错误

# copy contrial from '/home/ubuntu/sample2-data.json';
ERROR:  invalid input syntax for type json
DETAIL:  The input string ended unexpectedly.
CONTEXT:  JSON data, line 1: [
COPY contrial, line 1, column info: "["

我进一步修改的示例 (3) 数据

[{"url": "https://some1-value.com", "nested": {"id":4,"value":"some1 text value"}, "quotes": "5\"  side"},
{"url": "https://some2-value.com", "nested": {"id":5,"value":"some2 text value"}, "quotes": "6\"  side"},
{"url": "https://some3-value.com", "nested": {"id":6,"value":"some3 text value"}, "quotes": "7\"  side"}]

不同的错误

# copy contrial from '/home/ubuntu/sample2-data.json';
ERROR:  invalid input syntax for type json
DETAIL:  Token "side" is invalid.
CONTEXT:  JSON data, line 1: ...,"value":"some1 text value"}, "quotes": "5"  side...
COPY contrial, line 1, column info: "[{"url": "https://some1-value.com", "nested": {"id":4,"value":"some1 text value"}, "quotes": "5"  si..."

创建表语句

CREATE TABLE public.contrial (
    info json NOT NULL
);

最终目标是创建一个表，其中键作为列，值作为记录。嵌套键需要展平。

+-------------------------+-----------+------------------+----------+
| url                     | nested_id | nested_value     | quotes   |
+-------------------------+-----------+------------------+----------+
| https://some1-value.com | 4         | some1 text value | 5\" side |
+-------------------------+-----------+------------------+----------+
| https://some2-value.com | 5         | some2 text value | 6\" side |
+-------------------------+-----------+------------------+----------+
| https://some3-value.com | 6         | some3 text value | 7\" side |
+-------------------------+-----------+------------------+----------+

Answer 1

我最终使用了 Andre Dunstan's blog 和 this SO answer，它们表示以特定方式格式化 json 以使用复制命令。

由于我的结构是为我正在解析的文件定义的，所以我最终得到了以下脚本。

def file_len(fname):
    # to find the number of lines in the file.
    # Has been pretty efficient even for millions of records
    with open(fname) as f:
        for i, l in enumerate(f):
            pass
    return i + 1

INPUTFILE = '/path/to/input.json'
OUTPUTFILE = '/path/to/output.json.csv'
LEN = file_len(INPUTFILE)
with open(OUTPUTFILE, 'w') as fo:
    with open(INPUTFILE, 'r') as fi:
        for i, l in enumerate(fi):
            # I skip the first line
            if i == 0: continue 
            
            # To remove the ']}}' from the end
            elif i+1 == LEN: _ = fo.write(l[:-3])
            
            # To remove the ',' from the end 
            # and add \n since write does not add newline on its own
            else: _ = fo.write(l[:-2]+'\n') 

# load statement

import sqlalchemy
POSTGRESQL = f'postgresql+psycopg2://{USERNAME}:{PASSWORD}@{HOSTNAME}/{DB}'
engine = sqlalchemy.create_engine(POSTGRESQL, echo=True)
            
con = engine.connect()
trans = con.begin()
LOAD_SQL = f"COPY tablename from '{OUTPUTFILE}' with csv delimiter E'\x01' quote E'\x02' null as '';"
try:
    con.execute(LOAD_SQL)
    trans.commit()
except Exception as e:
    trans.rollback()
finally:
    con.close()