postgresql - Spark 写入 Postgresql。批量更新异常？

Question

我有一个简单的 Spark 作业，它读取大型日志文件、过滤它们并将结果写入新表。简化的 Scala 驱动程序应用程序代码是:

val sourceRdd = sc.textFile(sourcePath)

val parsedRdd = sourceRdd.flatMap(parseRow)

val filteredRdd = parsedRdd.filter(l => filterLogEntry(l, beginDateTime, endDateTime))

val dataFrame = sqlContext.createDataFrame(filteredRdd)

val writer = dataFrame.write

val properties = new Properties()
properties.setProperty("user", "my_user")
properties.setProperty("password", "my_password")
writer.jdbc("jdbc:postgresql://ip_address/database_name", "my_table", properties)

这对较小的批处理非常有效。在大批量上，执行两个小时后，我在目标表中看到大约 800 万条记录，Spark 作业失败并出现以下错误:

Caused by: java.sql.BatchUpdateException: Batch entry 524 INSERT INTO my_table <snip>  was aborted.  Call getNextException to see the cause.
    at org.postgresql.jdbc.BatchResultHandler.handleError(BatchResultHandler.java:136)
    at org.postgresql.core.v3.QueryExecutorImpl$ErrorTrackingResultHandler.handleError(QueryExecutorImpl.java:308)
    at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2004)
    at org.postgresql.core.v3.QueryExecutorImpl.flushIfDeadlockRisk(QueryExecutorImpl.java:1187)
    at org.postgresql.core.v3.QueryExecutorImpl.sendQuery(QueryExecutorImpl.java:1212)
    at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:351)
    at org.postgresql.jdbc.PgStatement.executeBatch(PgStatement.java:1019)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$.savePartition(JdbcUtils.scala:210)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:277)
    at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$saveTable$1.apply(JdbcUtils.scala:276)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920)
    at org.apache.spark.rdd.RDD$$anonfun$foreachPartition$1$$anonfun$apply$33.apply(RDD.scala:920)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.SparkContext$$anonfun$runJob$5.apply(SparkContext.scala:1858)
    at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
    at org.apache.spark.scheduler.Task.run(Task.scala:89)
    at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)

如果我将给定的 SQL INSERT 语句复制粘贴到 SQL 控制台中，它可以正常工作。在 Postgresql 服务器日志中，我看到:

(这是未修改/未匿名的日志)

2012016-04-26 22:38:09 GMT [3769-12] nginxlogs_admin@nginxlogs ERROR:  syntax error at or near "was" at character 544
2016-04-26 22:38:09 GMT [3769-13] nginxlogs_admin@nginxlogs STATEMENT:  INSERT INTO log_entries2 (client,host,req_t,request,seg,server,timestamp_value) VALUES ('68.67.161.5','"204.13.197.104"','0.000s','"GET /bid?apnx_id=&ip=67.221.130.195&idfa=&dmd5=&daid=&lt=32.90630&lg=-95.57920&u=branovate.com&ua=Mozilla%2F5.0+%28Linux%3B+Android+5.1%3B+XT1254+Build%2FSU3TL-39%3B+wv%29+AppleWebKit%2F537.36+%28KHTML%2C+like+Gecko%29+Version%2F4.0+Chrome%2F44.0.2403.90+Mobile+Safari%2F537.36+%5BFB_IAB%2FFB4A%3BFBAV%2F39.0.0.36.238%3B%5D&ap=&c=1&dmdl=&dmk= HTTP/1.1"','samba_info_has_geo','','2015-08-02T20:24:30.482000112') was aborted.  Call getNextException to see the cause.

似乎 Spark 向触发此特定错误的 Postgresql 发送了文本“已中止。调用 getNextException ...”。这似乎是一个合法的 Spark 错误。第二个问题是为什么 Spark 首先中止了它？

所以，afaik，我不能调用 getNextException，因为我没有直接使用 JDBC，而是通过 Spark。

仅供引用，这是针对 Spark 1.6.1 和 Scala 2.11 的。

Answer 1

如果其他人正在搜索并点击这个，我的数据库服务器(在 VM 中运行)达到磁盘空间限制，Spark 似乎被这个错误弄糊涂了，没有记录真正的错误，导致不同的内部错误，并记录结果。从技术上讲，这可能是响应不常见的数据库磁盘已满错误的内部 Spark 错误。