pyspark在bigquery中从本地加载数据框架

0 人关注

我想把一个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
我做错了什么？谢谢你


           
            
             
              
               你是否尝试在你的原始火花配置中加入这个条目
               
                .config("fs.AbstractFileSystem.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS")
               
               ？


           
            
             
              
               你好 @mk_sta, 谢谢你的回答。通过添加这个配置，我有同样的错误

csv


         
          
           
            google-bigquery


         
          
           
            google-cloud-storage


          
           
            
             
              我想，将DataFrame写入Bigquery的过程肯定需要定义临时的GCS桶，正如
              
               这里
              
              所解释的。
             
             
              因此，我在你的代码例子中没有看到
              
               temporaryGcsBucket
              
              ，这个参数是为特定的Spark Bigquery
              
               连接器
              
              在
              
               df.write
              
              函数中定义的，可能会导致Gs文件系统错误。
             
             
              你可以尝试在Spark配置中全局传播
              
               temporaryGcsBucket
              
              参数（如果你还没有做）。
             
             spark.conf.set('temporaryGcsBucket', bucket)
或者完全在df.write 中进行调整。
df.write
  .....
  .option("temporaryGcsBucket","some-bucket")
  .save()
你也可以查看采用的I/O Bigquery API参数的参考列表。
更新一下。
假设配置参数.config("temporaryGcsBucket", "test_project_bucket_pyspark") 已经按照代码示例在SparkSession 设置中声明，我可以确认它也会在SparkConf运行时属性中传播，这是这里常见的做法。
然而，为了进一步调试你所面临的问题，我建议检查Hadoop连接器的配置。
确保所有与Hadoop相关的配置在Spark会话执行期间被正确加载，检查错误输出是否有任何java.lang.ClassNotFoundException 。