Pyspark如何将行附加到BigQuery表

Question

我有一个 Pyspark 程序，最后我需要将行附加到 Bigquery 表。我第一次能够创建表并将行加载到其中，但不知道如何继续追加更多行。

这是我第一次加载到 BQ 的代码 sn-p -

mydf.write.format('bigquery') \
        .option('table', 'my-project.my-dataset.my-table') \
        .option("temporaryGcsBucket","my-bucket") \
        .save()

谁能帮帮我？

完全的例外，我在“附加”行时得到 -

原因：org.apache.spark.SparkException：写入时任务失败 行。 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:257) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:170) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:169) 在 org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90) 在 org.apache.spark.scheduler.Task.run(Task.scala:123) 在 org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408) 在 org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360) 在 org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414) 在 java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) 在 java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ... 1 更多原因：net.razorvine.pickle.PickleException：ClassDict 构造的预期零参数（对于 rdkit.Chem.rdchem.Mol) 在 net.razorvine.pickle.objects.ClassDictConstructor.construct(ClassDictConstructor.java:23) 在 net.razorvine.pickle.Unpickler.load_reduce(Unpickler.java:707) 在 net.razorvine.pickle.Unpickler.dispatch(Unpickler.java:175) 在 net.razorvine.pickle.Unpickler.load(Unpickler.java:99) 在 net.razorvine.pickle.Unpickler.loads(Unpickler.java:112) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:90) 在 org.apache.spark.sql.execution.python.BatchEvalPythonExec$$anonfun$evaluate$1.apply(BatchEvalPythonExec.scala:89) 在 scala.collection.Iterator$$anon$12.nextCur(Iterator.scala:435) 在 scala.collection.Iterator$$anon$12.hasNext(Iterator.scala:441) 在 scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409) 在 scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:409) 在 org.apache.spark.sql.catalyst.expressions.GeneratedClass$GeneratedIteratorForCodegenStage2.processNext（未知 资源） 在 org.apache.spark.sql.execution.BufferedRowIterator.hasNext(BufferedRowIterator.java:43) 在 org.apache.spark.sql.execution.WholeStageCodegenExec$$anonfun$13$$anon$1.hasNext(WholeStageCodegenExec.scala:636) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:244) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:242) 在 org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1394) 在 org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:248)

Answer 1

只需将保存模式设置为append：

df.write \
  .format('bigquery') \
  .option('table', 'my-project.my-dataset.my-table') \
  .option("temporaryGcsBucket","my-bucket") \
  .mode("append") \
  .save()

如果您查看spark-bigquery-connector 源代码，连接器仅支持保存模式overwrite 和append。