pyspark从本地加载bigquery中的数据框

Question

我想将 pyspark 数据框加载到 Google BigQuery 表中。

我通过运行来运行作业

spark-submit --jars batch/jars/gcs-connector-hadoop2-latest.jar,batch/jars/spark-bigquery-latest.jar main_batch.py

我想我已经设置了所有必需的配置：

sc = SparkSession\
        .builder\
        .config("temporaryGcsBucket", "test_project_bucket_pyspark") \
        .config("fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem")\
        .config("fs.gs.auth.service.account.enable", "true") \
        .config("google.cloud.auth.service.account.json.keyfile", "config/google/service_account_project.json") \
        .master("local[1]")\
        .appName("spark_etl")\
        .getOrCreate()

我正在运行它以保存到 BigQuery：

df.write \
        .format("bigquery") \
        .option('table', 'dataset_project_test.tickets') \
        .option("encoding", "UTF-8") \
        .option("nullValue", "\u0000") \
        .option("emptyValue", "\u0000") \
        .mode("append") \
        .save()

但我收到以下错误：

py4j.protocol.Py4JJavaError: An error occurred while calling o1176.save.
: java.io.IOException: No FileSystem for scheme: gs

我做错了什么？谢谢

Answer 1

我认为处理或将 DataFrame 写入 Bigquery 肯定需要按照 here 的解释定义临时 GCS 存储桶。

因此，我没有在您的代码示例中看到在 df.write 函数中为特定 Spark Bigquery connector 定义的参数 temporaryGcsBucket 可能导致 gs 文件系统错误。

您可以尝试在 Spark 配置中全局传播 temporaryGcsBucket 参数（如果您还没有这样做的话）：

spark.conf.set('temporaryGcsBucket', bucket)

或者在df.write中完全调整：

df.write
  .....
  .option("temporaryGcsBucket","some-bucket")
  .save()

您还可以查看reference 列表，了解采用的 I/O Bigquery API 参数。

更新：

假设配置参数.config("temporaryGcsBucket", "test_project_bucket_pyspark") 已根据代码示例在SparkSession settings 中声明，我可以确认它也会在SparkConf 运行时属性中传播，这是这里的常用方法。

但是，为了进一步调试您面临的问题，我建议您检查 Hadoop connector 的配置：

1. 确保在 Spark 会话执行期间正确加载所有与 Hadoop 相关的配置，检查任何java.lang.ClassNotFoundException 记录的错误输出；

2. 如果您还有任何与授权相关的错误，请确保您可以使用 gsutil (gsutil ls -b gs://<some-bucket>) 访问 temporaryGcsBucket，并验证您的配置中的凭据是否正确。

3. 尝试在 Spark 运行时属性中添加 fs.gs.project.id 参数：

   .config("fs.gs.project.id", "<MY_PROJECT>")