本地模式PySpark连接BigQuery的配置步骤咨询

• 依赖包配置
  下载与你的Spark版本兼容的BigQuery连接器JAR包（Spark 3.x建议使用v0.20及以上版本），启动PySpark时通过--jars参数指定JAR文件路径，示例命令：

  pyspark --jars /path/to/spark-bigquery-with-dependencies.jar
  

  也可在SparkSession初始化时通过config参数添加依赖。

• GCP身份认证

  1. 在GCP控制台创建并下载服务账号的JSON密钥文件，确保该账号拥有BigQuery数据查询权限（至少包含BigQuery Data Viewer和BigQuery Job User角色）
  2. 设置环境变量指向密钥文件：
     • Linux/macOS：export GOOGLE_APPLICATION_CREDENTIALS="/absolute/path/to/your-key.json"
     • Windows：set GOOGLE_APPLICATION_CREDENTIALS="C:\path\to\your-key.json"

• SparkSession核心参数设置
  初始化SparkSession时配置BigQuery连接器相关参数，示例代码：

  from pyspark.sql import SparkSession
  
  spark = SparkSession.builder \
      .appName("LocalSparkToBigQuery") \
      .config("spark.sql.catalog.gcp", "com.google.cloud.spark.bigquery.BigQueryCatalog") \
      .config("spark.sql.catalog.gcp.projectId", "your-gcp-project-id") \
      .getOrCreate()
  

• 数据查询与DataFrame转换
  有两种常用方式生成DataFrame：

  1. 通过Spark SQL直接查询BigQuery表，格式为gcp.数据集名.表名：

     df = spark.sql("SELECT * FROM gcp.your_dataset.your_table LIMIT 100")
     df.show()
     

  2. 使用read.format("bigquery")加载表：

     df = spark.read.format("bigquery") \
         .option("table", "your-gcp-project-id.your_dataset.your_table") \
         .load()
     

• 可选优化配置

  • 若查询大表，需配置临时GCS存储桶用于数据中转：spark.conf.set("temporaryGcsBucket", "your-temp-gcs-bucket")
  • 根据本地机器资源调整Spark内存配置，比如启动时添加--driver-memory 4g --executor-memory 4g避免OOM

内容的提问来源于stack exchange，提问作者PipelineSurfer