saveAsBigQueryTable 不起作用

Question

我尝试使用spotify的spark-bigquerySpark包将数据帧保存到Google BigQuery表中，但失败了。我在谷歌的dataproc平台上使用了它。

df.saveAsBigQueryTable("my-project:my_dataset.my_table") 

以下是错误日志：

org.apache.spark.SparkException：作业中止。 at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala：215) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala：173) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala：173) at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala：65) at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala：173) at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala：145) at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala：58) at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(command.scala：56) at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(command.scala：74) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala：117) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala：117) at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala：138) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala：151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala：135) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala：116) at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala：92) at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala：92) at org.apache.spark.sql.execution.datasources.DataSource.writeInFileFormat(DataSource.scala：438) at org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala：474) at org.apache.spark.sql.execution.datasources.SaveIntoDataSourceCommand.run(SaveIntoDataSourceCommand.scala：48) at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala：58) at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(command.scala：56) at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(command.scala：74) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala：117) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala：117) at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala：138) at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala：151) at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala：135) at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala：116) at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala：92) at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala：92) at org.apache.spark.sql.DataFrameWriter.runCommand(DataFrameWriter.scala：610) at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala：233) at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala：217) at com.databricks.spark.avro.package$AvroDataFrameWriter$$anonfun$avro$1.apply(package.scala：26) at com.databricks.spark.avro.package$AvroDataFrameWriter$$anonfun$avro$1.apply(package.scala：26) at com.spotify.spark.bigquery.BigQueryDataFrame.saveAsBigQueryTable(BigQueryDataFrame.scala：54) at com.spotify.spark.bigquery.BigQueryDataFrame.saveAsBigQueryTable(BigQueryDataFrame.scala：67) ...84 省略

由以下原因引起：org.apache.spark.Spark异常：作业由于阶段失败而中止：阶段 107.0 中的任务 0 失败 4 次 at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala：272)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala：191)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala：190)" at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala：87)" at org.apache.spark.scheduler.Task.run(Task.scala：108)" at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala：335)" 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：748)"原因：java.lang.AbstractMethodError： org.apache.spark.sql.execution.datasources.OutputWriterFactory.getFileExtension(Lorg/apache/hadoop/mapreduce/TaskTryContext;)Ljava/lang/String; at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.newOutputWriter(FileFormatWriter.scala：299)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.execute(FileFormatWriter.scala：314)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala：258)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala：256)" at org.apache.spark.util.Utils$.tryWithSafeLast AndFailureCallbacks(Utils.scala：1375)" at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala：261)" ...8 更多">

驱动程序堆栈跟踪： at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala：1499) at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala：1487) at org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala：1486) at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala：59) at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala：48) at org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala：1486) at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala：814) at org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala：814) 在斯卡拉。Option.foreach(Option.scala：257) at org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala：814) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala：1714) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala：1669) at org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala：1658) at org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala：48) at org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala：630) at org.apache.spark.SparkContext.runJob(SparkContext.scala：2022) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply$mcV$sp(FileFormatWriter.scala：188) ...121 更多 由以下原因引起：org.apache.spark.SparkException：写入行时任务失败 at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala：272) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala：191) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1$$anonfun$apply$mcV$sp$1.apply(FileFormatWriter.scala：190) at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala：87) at org.apache.spark.scheduler.Task.run(Task.scala：108) at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scal 答：335) ...3 更多 原因：java.lang.AbstractMethodError： org.apache.spark.sql.execution.datasources.OutputWriterFactory.getFileExtension(Lorg/apache/hadoop/mapreduce/TaskTryContext;)Ljava/lang/String; at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.newOutputWriter(FileFormatWriter.scala：299) at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.execute(FileFormatWriter.scala：314) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala：258) at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala：256) at org.apache.spark.util.Utils$.tryWithSafeLast AndFailureCallbacks(Utils.scala：1375) at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala：261) ...8 更多

Answer 1

我觉得可疑的部分是以下部分： 由： java.lang.AbstractMethodError： org.apache.spark.sql.execution.datasources.OutputWriterFactory.getFileExtension(Lorg/apache/hadoop/mapreduce/TaskTryContext;)Ljava/lang/String;at org.apache.spark.sql.execution.datasources.FileFormatWriter$SingleDirectoryWriteTask.newOutputWriter(FileFormatWriter.scala：299)"

搜索此特定错误会导致：https://github.com/databricks/spark-avro/issues/208。

由于 spark-avro 在 Spark 2.2 中存在问题，您可能会发现使用 Dataproc 映像版本 1.1 (Spark 2.0) 可以使 spark-bigquery 库正常运行。