此页面由 Cloud Translation API 翻译。

将 BigQuery 连接器与 Google Cloud Serverless for Apache Spark 搭配使用

将 spark-bigquery-connector 与 Apache Spark 搭配使用，以从 BigQuery 中读取数据以及将数据写入其中。本教程演示了使用 spark-bigquery-connector 的 PySpark 应用。

将 BigQuery 连接器与工作负载搭配使用

请参阅 Serverless for Apache Spark 运行时版本，确定您的批处理工作负载运行时版本中安装的 BigQuery 连接器版本。如果连接器未列出，请参阅下一部分，了解如何使连接器可供应用使用。

如何将连接器与 Spark 运行时版本 2.0 搭配使用

BigQuery 连接器未安装在 Spark 运行时版本 2.0 中。使用 Spark 运行时版本 2.0 时，您可以通过以下任一方式将连接器提供给您的应用：

提交 Google Cloud Serverless for Apache Spark 批处理工作负载时，使用 jars 参数指向连接器 jar 文件。以下示例指定了一个连接器 jar 文件（如需查看可用的连接器 jar 文件列表，请参阅 GitHub 上的 GoogleCloudDataproc/spark-bigquery-connector 代码库）。
- Google Cloud CLI 示例：
```
gcloud dataproc batches submit pyspark \
    --region=region \
    --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.13-version.jar \
    ... other args
```
将连接器 jar 文件作为依赖项添加到 Spark 应用中（请参阅针对连接器进行编译）

计算费用

本教程使用 Google Cloud的收费组件，包括：

Serverless for Apache Spark
BigQuery
Cloud Storage

请使用价格计算器根据您的预计使用情况来估算费用。

Cloud Platform 新用户可能有资格申请免费试用。

BigQuery I/O

此示例展示如何将 BigQuery 中的数据读取到 Spark DataFrame 中，以使用标准数据源 API执行字数统计操作。

连接器通过以下方式将字数统计输出写入 BigQuery：

将数据缓冲到 Cloud Storage 存储桶中的临时文件
将一项操作中的数据从 Cloud Storage 存储桶复制到 BigQuery 中
系统在 BigQuery 加载操作完成后删除 Cloud Storage 中的临时文件（临时文件在 Spark 应用终止后也会被删除）。如果删除失败，您需要删除所有不需要的临时 Cloud Storage 文件，这些文件通常放在 gs://YOUR_BUCKET/.spark-bigquery-JOB_ID-UUID 中。

配置结算功能

默认情况下，与凭证或服务账号关联的项目将为 API 使用支付费用。要对其他项目计费，请设置以下配置：spark.conf.set("parentProject", "<BILLED-GCP-PROJECT>")。

还可以将其添加到读取或写入操作，如下所示：.option("parentProject", "<BILLED-GCP-PROJECT>")。

提交 PySpark 字数统计批处理工作负载

运行一个 Spark 批处理工作负载，用于统计公共数据集中的字词数量。

打开本地终端或 Cloud Shell
在本地终端或 Cloud Shell 中使用 bq 命令行工具创建 wordcount_dataset。
```
bq mk wordcount_dataset
```
使用 Google Cloud CLI 创建 Cloud Storage 存储桶。
```
gcloud storage buckets create gs://YOUR_BUCKET
```
将 YOUR_BUCKET 替换为您创建的 Cloud Storage 存储桶的名称。

通过复制以下 PySpark 代码，在文本编辑器中本地创建文件 wordcount.py。

#!/usr/bin/python
"""BigQuery I/O PySpark example."""
from pyspark.sql import SparkSession

spark = SparkSession \
  .builder \
  .appName('spark-bigquery-demo') \
  .getOrCreate()

# Use the Cloud Storage bucket for temporary BigQuery export data used
# by the connector.
bucket = "YOUR_BUCKET"
spark.conf.set('temporaryGcsBucket', bucket)

# Load data from BigQuery.
words = spark.read.format('bigquery') \
  .option('table', 'bigquery-public-data:samples.shakespeare') \
  .load()
words.createOrReplaceTempView('words')

# Perform word count.
word_count = spark.sql(
    'SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')
word_count.show()
word_count.printSchema()

# Saving the data to BigQuery
word_count.write.format('bigquery') \
  .option('table', 'wordcount_dataset.wordcount_output') \
  .save()

提交 PySpark 批处理工作负载：

gcloud dataproc batches submit pyspark wordcount.py \
    --region=REGION \
    --deps-bucket=YOUR_BUCKET

示例终端输出：

...
+---------+----------+
|     word|word_count|
+---------+----------+
|     XVII|         2|
|    spoil|        28|
|    Drink|         7|
|forgetful|         5|
|   Cannot|        46|
|    cures|        10|
|   harder|        13|
|  tresses|         3|
|      few|        62|
|  steel'd|         5|
| tripping|         7|
|   travel|        35|
|   ransom|        55|
|     hope|       366|
|       By|       816|
|     some|      1169|
|    those|       508|
|    still|       567|
|      art|       893|
|    feign|        10|
+---------+----------+
only showing top 20 rows

root
 |-- word: string (nullable = false)
 |-- word_count: long (nullable = true)

如需在 Google Cloud 控制台中预览输出表，请打开项目的 BigQuery 页面，选择 wordcount_output 表，然后点击预览。