使用Python Psycopg2将JSON数据插入到远程Postgres表中需要花费太长的time.Is时间，这是正确的方法吗？

Question

我有来自Mongodb导出的2 2GB数据。我正在尝试用这些数据做一些分析，并将这些分析数据发布到远程Postgres table.This是我的代码：

import json

from psycopg2 import connect, Error
import dateutil.parser as parser
from datetime import datetime

record_list = []

for line in open('december.json', 'r'):
    record_list.append(json.loads(line))

    #print(record_list[0])
    


    devices = []

    for key, values in record_list[0].items():


       if key == "deviceSerial":
          devices.append(values)
    test_list = list(set(devices)) 
    device_data = []

    for x in test_list:
        device_1 = [i for i in record_list if (i['deviceSerial'] == x)] 
        datetime_object_first = parser.parse(device_1[0]['createdAt'],fuzzy=True) 
        datetime_object_last = parser.parse(device_1[len(device_1)-1]['createdAt'],fuzzy=True) 

        devices = device_1[0]['deviceSerial']
        device_model = device_1[0]['device']
        device_Usage=round((datetime.timestamp(datetime_object_last)-datetime.timestamp(datetime_object_first))/60/1000,3)


        #calculating device usage
        device_data_elements ={

            'deviceModel': device_model,
            'deviceSerial':devices,
            'deviceUsageInMin':device_Usage
     
        }

        device_data.append(device_data_elements)

        if type(device_data) == list:



            first_record = device_data[0]


        columns = list(first_record.keys())

        #print ("\ncolumn names:", columns)

        #print(device_data)

        table_name = "test_data"
sql_string = 'INSERT INTO {} '.format( table_name )
sql_string += "(" + ', '.join(columns) + ")\nVALUES "



# enumerate over the record
for i, record_dict in enumerate(device_data):

    # iterate over the values of each record dict object
    values = []
    for col_names, val in record_dict.items():

        if type(val) == str:
          
            val = val.replace("'", "''")
            val = "'" + val + "'"

        values += [ str(val) ]

        # join the list of values and enclose record in parenthesis
    sql_string += "(" + ', '.join(values) + "),\n"

# remove the last comma and end statement with a semicolon
sql_string = sql_string[:-2] + ";"

print ("\nSQL statement:")
print (sql_string)

try:
    # declare a new PostgreSQL connection object
    conn = connect(
        dbname = "events_data",
        user = "chao",
        host = "localhost",
        
        # attempt to connect for 10 seconds then raise exception
        connect_timeout = 10
    )

    cur = conn.cursor()
    print ("\ncreated cursor object:", cur)
    
#Post data into postgres table
except (Exception, Error) as err:
    print ("\npsycopg2 connect error:", err)
    conn = None
    cur = None

if cur != None:

    try:
        
        cur.execute( sql_string )
        conn.commit()

        print ('\nfinished INSERT INTO execution')

    except (Exception, Error) as error:
        print("\nexecute_sql() error:", error)
        conn.rollback()

    # close the cursor and connection
    cur.close()
    conn.close()

我正在做一些计算(分析)与这个script.So有一些script.When循环中的数据长度较小，脚本插入这些分析到Postgres successfully.If数据长度很大，它需要太长的时间，我等了12个小时，但没有成功.Right现在我的脚本是在本地运行和读取数据也locally.What将是读取和操纵大量数据和Postgres分析的最佳方法table.This是我的样本json数据。

[
    {
   "createdAt": "Fri Nov 27 2020 08:07:39 GMT+0000 ",
   "sessionId": null,
   "text": null,
   "device": null,
   "deviceSerial": null
 },
 {
   "createdAt": "Tue Sep 01 2020 06:59:18 GMT+0000",
   "sessionId": null,
   "text": "Approve",
   "device": "Android",
   "deviceSerial": null
 },
 {
   "createdAt": "Wed Sep 02 2020 08:40:10 GMT+0000",
   "pageTitle": "submit option",
   "sessionId": null,
   "text": "launchComponent",
   "device": "Android",
   "deviceSerial": "636363636890"
 },
 {
   "createdAt": "Wed Sep 02 2020 08:40:11",
   "pageTitle": "quick check",
   "sessionId": "88958d89c65f4fcea56e148a5a2838cfhdhdhd",
   "text": "",
   "device": "Android",
   "deviceSerial": "6625839827"
 }
]

Answer 1

我建议将JSON转储到Postgres中，并在Postgres中进行分析。这就是Postgres擅长的。不需要处理JSON，您可以直接将JSON数组转换为Postgres行。

一种选择是创建一个只有一个jsonb列的表，并使用jsonb_array_elements将每个项目作为一行插入。

create table devices_json (
    data jsonb
)

insert into devices_json (data)
select * from jsonb_array_elements('
[
 {
   "createdAt": "Fri Nov 27 2020 08:07:39 GMT+0000 ",
   "sessionId": null,
   "text": null,
   "device": null,
   "deviceSerial": null
 },
 ...and so on...
]
')

然后使用its JSON functions在Postgres中进行分析。

如果字段是众所周知的，您可以使用json_populate_recordset将字段插入到传统SQL表的单个列中。然后你就有了一个传统的SQL表，它可能更容易使用。

-- NOTE Postgres columns are case-sensitive, so they must be quoted to
-- ensure they exactly match the JSON keys
create table devices (
   "createdAt" timestamp,
   "sessionId" text,
   "text" text,
   device text,
   "deviceSerial" text,
   "pageTitle" text
);

insert into devices
select * from json_populate_recordset(NULL::devices, '
[
 {
   "createdAt": "Fri Nov 27 2020 08:07:39 GMT+0000 ",
   "sessionId": null,
   "text": null,
   "device": null,
   "deviceSerial": null
 },
 ...and so on...
]
')

Try it。

您也可以混合使用这两种方法:将JSON数组转储到一个jsonb列中，然后在Postgres中将其拆分到单独的列中。

Answer 2

Posgres中的批量插入最好由cursor.copy_from完成。

在Python语言中将json转换为所需的结构，然后写入io.TextIO内存文件。只需调用一次copy_from()，即可将其推送到数据库。在copy_from之前对TextIO对象执行seek(0)。

import io
import psycopg2

buf = io.StringIO()
buf.write('1|{"x":0}\n')
buf.write('2|{"y":0}\n')
buf.seek(0)

with psycopg2.connect(service="my_service") as con:
    cur = con.cursor()
    cur.execute("create table if not exists t (a int, b jsonb)")
    cur.copy_from(buf, "t", sep="|")
    print(cur.rowcount)

插入到postgres的速度不能比这更快。

cursor.copy_expert可以更灵活地做同样的事情。

关于Niels

Answer 3

谢谢你@Schwern.It。真的很好的帮助。我想做我的分析dynamically.For示例我想排序deviceSerial的列，并计算它的使用从最后和第一个index.This是我的analytics.There将是数千行数据的差异。我想将这些分析发布到另一个表。我正在考虑再次在Python中获取这些Postgres行，并进行一些计算，并将结果推送到Postgres.Would。这是一个好方法吗?或者我可以这样做，而不是从Postgres.Instead中获取数据。