编写超过5000万从Pyspark df到PostgresSQL，最有效的方法

Question

从Spark数据框到Postgres表格插入数百万条记录的最有效方法是什么？我在过去通过使用批量复制和批量大小选项也成功地从火花到MSSQL做到了这一点。

有没有类似的东西可以在这里为博士后？

添加我尝试过的代码以及运行流程所需的时间：

def inserter():
    start = timer()
    sql_res.write.format("jdbc").option("numPartitions","5").option("batchsize","200000")\
    .option("url", "jdbc:postgresql://xyz.com:5435/abc_db") \
    .option("dbtable", "public.full_load").option("user", "root").option("password", "password").save()
    end = timer()
    print(timedelta(seconds=end-start))
inserter()

所以我做了上面的方法，1000万记录，并有5个并行连接，如NumPARtions中指定的，还尝试了200k的批量大小。

整个过程的总时间为0:14:05.760926（14分5秒）。

有没有其他有效的方法可以缩短时间？

我能使用的有效或最佳批量是多少？增加批量是否能更快地完成这项工作？或者打开多个连接，即

对于1000万条记录来说，平均14分钟是不错的，但是寻找那些以前会这样做的人来帮助回答这个问题。

Answer 1

我之前也做过类似的工作，但使用的是ApacheSqoop。

我想说，为了回答这个问题，我们必须尝试优化Spark和PostgresSQL之间的通信，特别是从Spark到PostgresSQL的数据流。

但要小心，不要忘记火花的一面。如果分区的数量与PostgreSQL支持的最大连接数量相比太高，那么执行mapPartitions是没有意义的。如果您有太多分区，并且要为每个分区打开一个连接，则可能会出现以下错误org。postgresql。util。PSQLException:致命：抱歉，已经有太多客户端。

为了调整插入过程，我将按照以下步骤处理这个问题：

• 记住分区的数量很重要。检查分区的数量，然后根据您想要的并行连接数量进行调整。您可能希望每个分区都有一个连接，因此我建议检查合并，如下所述

最后，做这项工作没有灵丹妙药。你可以使用我上面提到的所有技巧，但这实际上取决于你的数据和用例。