python - Cloud SQL - Postgres - 插入速度非常慢

Question

我可能已经习惯了大数据技术，但是当我尝试插入约 300k 行以达到 30Mb 的总大小(作为 csv)时，我认为 15 分钟是用 Postgres 进行 INSERT 的可接受时间。

我首先了解到增加 GCP 上的总磁盘大小也会增加 IOPS，因此我将磁盘从 20Go 增加到 400Go。

然后环顾互联网，我发现了这篇文章:https://naysan.ca/2020/05/09/pandas-to-postgresql-using-psycopg2-bulk-insert-performance-benchmark/

我使用旧的 df.to_sql() 来插入我的数据(我疯了，对吧!)。我记得对于 MSSQL 驱动程序有一个类似 fast_executemany 的参数，但没有适合 psycopg2 的参数。

所以我尝试了文章中更快的方法，将 df 复制到内存中并写入数据库。通过所有这些升级(磁盘大小 + 代码优化)，我从 30/35 分钟缩短到 ~15 分钟。

改进!但仍然在 ~1000s 左右

这里可能是什么问题？我的互联网连接非常好，所以瓶颈不在这里。

这是我正在做的一个例子:

import pandas as pd
import psycopg2
import time
from io import StringIO


if __name__ == '__main__':
    startTime = time.time()
    df = pd.read_gbq(
        "SELECT * FROM dataset.defaut",
        project_id="my_id_12345",
        location="europe-west1"
    )

    executionTime = (time.time() - startTime)
    print('Execution GBQ Read time in seconds: ' + str(executionTime))

    df_qgis = df[
        ["voie", "direction", "date_mesure",
         "dfo_id", "pk_ref", "type_defaut",
         "niveau_ref", "val_ref", "longueur",
         "pk_debut", "pk_fin", "pk", "pk_original"]]

    param_dic = {
        "host": "1.2.3.4",
        "database": "qgis",
        "user": "user",
        "password": "y0lo123"
    }


    def connect(params_dic):
        """ Connect to the PostgreSQL database server """
        conn = None
        try:
            # connect to the PostgreSQL server
            print('Connecting to the PostgreSQL database...')
            conn = psycopg2.connect(**params_dic)
        except (Exception, psycopg2.DatabaseError) as error:
            print(error)
            exit(1)
        print("Connection successful")
        return conn


    conn = connect(param_dic)


    def copy_from_stringio(conn, df, table):
        """
        Here we are going save the dataframe on disk as
        a csv file, load the csv file
        and use copy_from() to copy it to the table
        """
        # save dataframe to an in memory buffer
        buffer = StringIO()
        df.to_csv(buffer, index_label='id', header=False)
        buffer.seek(0)
        cursor = conn.cursor()
        try:
            cursor.copy_from(buffer, table, sep=",")
            conn.commit()
        except (Exception, psycopg2.DatabaseError) as error:
            print("Error: %s" % error)
            conn.rollback()
            cursor.close()
            return 1
        print("copy_from_file() done")
        cursor.close()


    copy_from_stringio(conn, df_qgis, "develop.defaut")

    executionTime = (time.time() - startTime)
    print('Execution final time in seconds: ' + str(executionTime))

我定位的表本身:

CREATE TABLE IF NOT EXISTS public.defaut
(
    id integer GENERATED ALWAYS AS IDENTITY,
    voie varchar NOT NULL,
    direction varchar NULL,
    date_mesure varchar NOT NULL,
    dfo_id int4 NOT NULL,
    pk_ref varchar NOT NULL,
    type_defaut varchar NOT NULL,
    niveau_ref varchar NOT NULL,
    val_ref float8 NOT NULL,
    longueur float8 NULL,
    pk_debut float8 NOT NULL,
    pk_fin float8 NOT NULL,
    pk float8 NOT NULL,
    pk_original float8 NULL,
    secteur varchar NULL,
    date_sect2 varchar NULL
);

Answer 1

根据云SQL best practices , 以加快导入速度(对于小实例大小):

you can temporarily increase the tier of an instance to improveperformance when importing large datasets.

Here您可以找到 PostgreSQL 实例的示例机器类型。

---

如果这没有帮助，我建议您联系 GCP technical support使用他们的内部工具检查您的实例，看看它的任何资源是否耗尽。