如何加速 spark df.write jdbc 到 postgres 数据库？答案

【问题标题】：How to speed up spark df.write jdbc to postgres database?如何加速 spark df.write jdbc 到 postgres 数据库？
【发布时间】：2020-02-28 19:04:04
【问题描述】：

我是 spark 新手，正在尝试使用 df.write 加速将数据帧的内容（可以有 200k 到 2M 行）附加到 postgres 数据库：

df.write.format('jdbc').options(
      url=psql_url_spark,
      driver=spark_env['PSQL_DRIVER'],
      dbtable="{schema}.{table}".format(schema=schema, table=table),
      user=spark_env['PSQL_USER'],
      password=spark_env['PSQL_PASS'],
      batchsize=2000000,
      queryTimeout=690
      ).mode(mode).save()

我尝试增加批量大小，但这并没有帮助，因为完成这项任务仍然需要大约 4 小时。我还在下面包含了一些来自 aws emr 的快照，显示了有关作业如何运行的更多详细信息。将数据帧保存到 postgres 表的任务仅分配给一个执行程序（我觉得这很奇怪），是否需要在执行程序之间划分此任务？

另外，我读过spark's performance tuning docs，但增加batchsize 和queryTimeout 似乎并没有提高性能。（我尝试在 df.write 之前在我的脚本中调用 df.cache()，但脚本的运行时间仍然是 4 小时）

另外，我的 aws emr 硬件设置和spark-submit 是：

主节点（1）：m4.xlarge

核心节点（2）：m5.xlarge

spark-submit --deploy-mode client --executor-cores 4 --num-executors 4 ...

【问题讨论】：

你能解决这个问题吗？我也面临同样的问题。

标签： postgresql apache-spark pyspark apache-spark-sql pyspark-sql

【解决方案1】：

Spark 是一个分布式数据处理引擎，因此当您处理数据或将其保存在文件系统上时，它会使用其所有执行程序来执行任务。 Spark JDBC 很慢，因为当您建立 JDBC 连接时，其中一个执行程序会建立到目标数据库的链接，从而导致速度缓慢和失败。

要克服这个问题并加快数据写入数据库的速度，您需要使用以下方法之一：

方法一：

在这种方法中，您需要使用 postgres COPY 命令实用程序 以加快写入操作。这要求您的 EMR 集群上有 psycopg2 库。

COPY 实用程序的文档是here

如果您想了解基准差异以及为什么复制速度更快，请访问here！

Postgres 还建议使用 COPY 命令进行批量插入。现在如何批量插入火花数据帧。现在要实现更快的写入，首先将您的 spark 数据帧以 csv 格式保存到 EMR 文件系统，然后重新分区您的输出，以便没有文件包含超过 100k 行。

#Repartition your dataframe dynamically based on number of rows in df
df.repartition(10).write.option("maxRecordsPerFile", 100000).mode("overwrite").csv("path/to/save/data)

现在使用 python 读取文件并为每个文件执行复制命令。

import psycopg2    
#iterate over your files here and generate file object you can also get files list using os module
file = open('path/to/save/data/part-00000_0.csv')
file1 = open('path/to/save/data/part-00000_1.csv')

#define a function
def execute_copy(fileName):
    con = psycopg2.connect(database=dbname,user=user,password=password,host=host,port=port)
    cursor = con.cursor()
    cursor.copy_from(fileName, 'table_name', sep=",")
    con.commit()
    con.close()

为了获得额外的速度提升，由于您使用的是 EMR 集群，您可以利用 python 多处理一次复制多个文件。

from multiprocessing import Pool, cpu_count
with Pool(cpu_count()) as p:
        print(p.map(execute_copy, [file,file1]))

这是推荐的方法，因为由于连接限制，无法调整 spark JDBC 以获得更高的写入速度。

方法二： 由于您已经在使用 AWS EMR 集群，因此您始终可以利用 hadoop 功能更快地执行表写入。所以在这里我们将使用 sqoop export 将我们的数据从 emrfs 导出到 postgres db。

#If you are using s3 as your source path
sqoop export --connect jdbc:postgresql:hostname:port/postgresDB --table target_table --export-dir s3://mybucket/myinputfiles/ --driver org.postgresql.Driver --username master --password password --input-null-string '\\N' --input-null-non-string '\\N' --direct -m 16

#If you are using EMRFS as your source path
sqoop export --connect jdbc:postgresql:hostname:port/postgresDB --table target_table --export-dir /path/to/save/data/ --driver org.postgresql.Driver --username master --password password --input-null-string '\\N' --input-null-non-string '\\N' --direct -m 16

为什么是sqoop？因为 sqoop 会根据指定的 mapper 数量打开与数据库的多个连接。因此，如果您将 -m 指定为 8，那么将有 8 个并发连接流，它们会将数据写入 postgres。

此外，有关使用 sqoop 的更多信息，请参阅AWS Blog、SQOOP Considerations 和 SQOOP Documentation。

如果您可以用自己的方式编写代码，那么方法 1 肯定会为您提供所需的性能提升，如果您对 SQOOP 等 hadoop 组件感到满意，那么请使用第二种方法。

希望对你有帮助！

【讨论】：

@Shubham_Jain 如何使用 scala 在 spark 中使用复制操作？
' 其中一个执行者建立了到目标数据库的链接，因此导致速度慢和失败。这不是我的经验。相反，写入 db 时的 Spark 并行性与写入文件时相同（每个内核每个分区的线程）。也许 OP 将他的数据放在一个分区中？不过，不要减损一个很好的答案。
如果我运行这段代码，它不会分配给工作人员，对吧？它将全部在 spark 驱动程序上运行，我可以使用的唯一并行化是 Python 并行化，具体取决于驱动程序机器的内核。还是我在这里错过了什么？

【解决方案2】：

Spark 端调优 => 在 Datafarme 上执行 repartition，以便多个执行程序并行写入数据库

df
.repartition(10)        // No. of concurrent connection Spark to PostgreSQL
.write.format('jdbc').options(
      url=psql_url_spark,
      driver=spark_env['PSQL_DRIVER'],
      dbtable="{schema}.{table}".format(schema=schema, table=table),
      user=spark_env['PSQL_USER'],
      password=spark_env['PSQL_PASS'],
      batchsize=2000000,
      queryTimeout=690
      ).mode(mode).save()

Postgresql 端调优 => PostgreSQL下面的参数需要分别调高。

max_connections确定最大并发数与数据库服务器的连接。默认值通常为 100 连接。
shared_buffers配置参数决定多少内存专用于 PostgreSQL，用于缓存数据。

【讨论】：

【解决方案3】：

要解决性能问题，一般需要解决以下2个瓶颈：

确保 spark 作业正在将数据并行写入 DB - 要解决此问题，请确保您有一个分区数据框。使用“df.repartition(n)”对数据帧进行分区，以便每个分区并行写入数据库。注意 - 大量的执行器也会导致插入速度变慢。所以从 5 个分区开始，然后将分区数增加 5 个，直到获得最佳性能。
确保数据库具有摄取批量数据所需的足够计算、内存和存储空间。

【讨论】：

【解决方案4】：

通过重新分区数据帧，您可以获得更好的写入性能，这是一个已知的答案。但是有一种重新分区数据框的最佳方法。由于您在 EMR 集群上运行此过程，因此首先要了解实例类型和每个从属实例上运行的核心数量。根据该指定数据帧上的分区数。在您的情况下，您使用的是 m5.xlarge(2 slaves)，每个都有 4 个 vCPU，这意味着每个实例有 4 个线程。因此，当您处理大量数据时，8 个分区将为您提供最佳结果。

注意：分区数应根据您的数据大小增加或减少。

注意：批量大小也是您在写入时应考虑的因素。批量越大，性能越好

【讨论】：