Spark写入postgres慢

Posted 2023-04-17

【中文标题】Spark写入postgres慢

【英文标题】：Spark write to postgres slow

【发布时间】：2016-09-08 17:22:46

【问题描述】：

我正在将数据帧中的数据（大约 83M 条记录）写入 postgresql，这有点慢。需要 2.7 小时才能完成对 db 的写入。

查看执行程序，只有一个执行程序在运行一个活动任务。有什么方法可以使用 Spark 中的所有执行程序并行写入数据库？

...
val prop = new Properties()
prop.setProperty("user", DB_USER)
prop.setProperty("password", DB_PASSWORD)
prop.setProperty("driver", "org.postgresql.Driver")



salesReportsDf.write
              .mode(SaveMode.Append)
              .jdbc(s"jdbc:postgresql://$DB_HOST:$DB_PORT/$DATABASE", REPORTS_TABLE, prop)

谢谢

【问题讨论】：

能否添加部分代码正在写入 PostGres？

@ThiagoBaldim 刚刚为此发布了代码 sn-p，谢谢

【参考方案1】：

所以我发现了问题所在。基本上，重新分区我的数据帧会将数据库写入吞吐量提高 100%

def srcTable(config: Config): Map[String, String] = 

  val SERVER             = config.getString("db_host")
  val PORT               = config.getInt("db_port")
  val DATABASE           = config.getString("database")
  val USER               = config.getString("db_user")
  val PASSWORD           = config.getString("db_password")
  val TABLE              = config.getString("table")
  val PARTITION_COL      = config.getString("partition_column")
  val LOWER_BOUND        = config.getString("lowerBound")
  val UPPER_BOUND        = config.getString("upperBound")
  val NUM_PARTITION      = config.getString("numPartitions")

  Map(
    "url"     -> s"jdbc:postgresql://$SERVER:$PORT/$DATABASE",
    "driver"  -> "org.postgresql.Driver",
    "dbtable" -> TABLE,
    "user"    -> USER,
    "password"-> PASSWORD,
    "partitionColumn" -> PARTITION_COL,
    "lowerBound" -> LOWER_BOUND,
    "upperBound" -> UPPER_BOUND,
    "numPartitions" -> NUM_PARTITION
  )

【讨论】：

能否请您提供更多详细信息（例如重新分区之前和之后的分区）我也面临类似的问题，非常感谢您在这方面的帮助 - 谢谢

@user2359997 更新了我的答案，具体取决于表的大小 - 您可以指定分区数，以便每个执行程序可以并行化数据的摄取。

@AdetiloyePhilipKehinde lowerBound、upperBound 和 partitionColumn 对写入到您的数据库没有任何影响。根据文档，这些选项“......仅适用于阅读”：spark.apache.org/docs/latest/…

@AdetiloyePhilipKehinde 您能否指定整个写入查询，以便更好地了解您用来加快流程的速度？

【参考方案2】：

Spark 在使用 jdbc 编写时还有一个名为“batchsize”的选项。默认值非常低。(1000)

connectionProperties.put("batchsize", "100000")

将其设置为更高的值应该会加快写入外部数据库的速度。