1
大きなログファイルを読み込んでフィルタリングし、結果を新しいテーブルに書き込む単純なSparkジョブがあります。簡略化されたScalaドライバのアプリコードは次のとおりです。Postgresqlに書き込みを書き込みます。 BatchUpdateException？
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)
これは、小規模なバッチでは完全に機能します。大規模なバッチでは、実行の2時間後、私は、ターゲット表には約8万レコードを参照し、スパーク・ジョブは次のエラーで失敗しました：私はに与えられたSQLのINSERT文をコピー＆ペーストした場合
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コンソール、それは正常に動作します。 PostgreSQLサーバログでは、私は以下を参照してください。
（これは修正されていない/ unanonymizedログ）
2012016-04-26 22:38:09 GMT [3769-12] [email protected] ERROR: syntax error at or near "was" at character 544
2016-04-26 22:38:09 GMT [3769-13] [email protected] 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=<=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.
テキストを送った「中止されたgetNextExceptionを呼び出し...。」PostgreSQLとのこの特定のトリガーエラー。それは正当なスパークバグのようだ。 2番目の質問は、なぜSparkがこれを最初に中止したかということです。
私はJDBCを直接使用するのではなく、Sparkを経由しているので、私はgetNextExceptionを呼び出すことができません。
参考までに、これはSpark 1.6.1とScala 2.11です。
値節内の項目は、 'VALUES（ '" 2016-04-27 "'、...）のように奇妙な方法で引用されています。日付。 – wildplasser
はい、「要求」と「ホスト」のテキストフィールドに無関係な引用問題があり、それはクリーンアップのために修正する必要がありますが、エラーは発生していません。 1つの日付フィールドは現在文字列として扱われており、無関係な引用問題はありません。文字列ではなく日付フィールドを適切なpostgresqlの日付型にする方が良いでしょうが、それは問題の原因ではありません。 – clay
スティングについては、引用符を含めることができます。日時とタイムスタンプの場合は、引用符で解読できません。 – wildplasser