如何在BigQuery PySpark存储过程中运行标准SQL查询并解决表不存在问题

问题根源 #

PySpark的spark.sql默认仅识别Spark元数据内的表/视图，而BigQuery公共数据集的表不在Spark默认元数据中，直接写SQL会触发找不到表的错误。

解决方案 #

方案1：加载BigQuery表为临时视图后查询 #

先通过BigQuery连接器将目标表加载为Spark DataFrame，再注册为临时视图，之后就能在spark.sql中直接引用视图名。

修改后的代码示例：

from pyspark.sql import SparkSession

# 构建SparkSession时配置BigQuery连接器
spark = SparkSession.builder \
    .appName("work_with_sql") \
    # 配置BigQuery Catalog（适用于Spark 3.0+）
    .config("spark.sql.catalog.google", "com.google.cloud.spark.bigquery.BigQueryCatalog") \
    .config("spark.hadoop.google.cloud.project.id", "你的GCP项目ID") \
    .getOrCreate()

# 将BigQuery公共表加载为临时视图
spark.read.format("bigquery") \
    .option("table", "bigquery-public-data.stackoverflow.posts_questions") \
    .load() \
    .createOrReplaceTempView("posts_questions")

# 基于临时视图执行SQL查询
df = spark.sql(
'''
SELECT tag, COUNT(*) c
FROM (
     SELECT SPLIT(tags, '|') tags
     FROM posts_questions a
     WHERE EXTRACT(YEAR FROM creation_date)>=2024
     ), UNNEST(tags) tag
GROUP BY 1
ORDER BY 2 DESC
LIMIT 10
'''
)

df.show()
spark.stop()

方案2：直接使用BigQuery Catalog语法查询 #

如果已配置BigQuery Catalog，可在spark.sql中直接通过Catalog路径引用BigQuery表，无需创建临时视图。

修改后的代码示例：

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("work_with_sql") \
    .config("spark.sql.catalog.google", "com.google.cloud.spark.bigquery.BigQueryCatalog") \
    .config("spark.hadoop.google.cloud.project.id", "你的GCP项目ID") \
    .getOrCreate()

# 直接通过Catalog路径引用BigQuery表
df = spark.sql(
'''
SELECT tag, COUNT(*) c
FROM (
     SELECT SPLIT(tags, '|') tags
     FROM google.`bigquery-public-data`.stackoverflow.posts_questions a
     WHERE EXTRACT(YEAR FROM creation_date)>=2024
     ), UNNEST(tags) tag
GROUP BY 1
ORDER BY 2 DESC
LIMIT 10
'''
)

df.show()
spark.stop()

注意事项 #

• 在BigQuery Serverless Spark环境中，BigQuery连接器依赖已默认内置，无需额外添加spark.jars.packages配置；如果是本地或其他Spark环境，需添加对应版本的连接器依赖包。
• 替换代码中的你的GCP项目ID为实际运行Spark作业的GCP项目ID，用于授权访问BigQuery公共数据集。

内容的提问来源于stack exchange，提问作者Ugur Selim Ozen