使用 Spark 数据沿袭

本文档介绍了如何在项目或集群级层为 Managed Service for Apache Spark 作业启用数据沿袭。

数据沿袭是Knowledge Catalog的一项功能，可让您跟踪数据在系统中的移动方式：数据来自何处、传递到何处以及对其应用了哪些转换。

数据沿袭适用于所有 Managed Service for Apache Spark 作业（SparkR 和 Spark 流式作业除外），并支持 BigQuery 和 Cloud Storage 数据源。它包含在 Managed Service for Apache Spark 2.0.74+、2.1.22+、2.2.50+、2.3.1+ 和 3.0 映像版本中。

在 Managed Service for Apache Spark 集群中启用此功能后，Managed Service for Apache Spark Spark 作业会捕获数据沿袭事件并将其发布到 Knowledge Catalog Data Lineage API。Managed Service for Apache Spark 使用 OpenLineage Spark 插件通过 OpenLineage 与 Data Lineage API 集成。

您可以通过 Knowledge Catalog 使用以下方式访问数据沿袭信息：

准备工作

  1. 在 Google Cloud 控制台中的项目选择器页面上，选择包含您要跟踪其沿袭的 Managed Service for Apache Spark 集群的项目。

    转到“项目选择器”

  2. 启用 Data Lineage API。

    启用 API

    即将发生的 Spark 数据沿袭变更：请参阅 Managed Service for Apache Spark 版本说明，了解即将发生的一项变更。该变更生效后，当您启用 Data Lineage API 时，系统会自动为您的项目和集群提供 Spark 数据沿袭（请参阅控制服务的沿袭提取），而无需额外的项目或集群级设置。

所需的角色

如果您使用默认虚拟机服务账号创建 Managed Service for Apache Spark 集群，该账号将具有 Dataproc Worker 角色。此角色包含 datalineage.locations.processOpenLineageMessage 权限，OpenLineage Spark 插件使用该权限来捕获沿袭事件并将其发布到 Data Lineage API。除此之外没有其他要求。

如果您创建的 Managed Service for Apache Spark 集群使用自定义服务账号，则该服务账号必须具有发布沿袭事件的权限。为遵循最小权限原则，请授予您的用例所需的最低角色：

  • Dataproc Worker (roles/dataproc.worker)：Managed Service for Apache Spark 数据平面操作所需的标准角色。由于此角色已包含 datalineage.locations.processOpenLineageMessage 权限，因此如果您的自定义服务账号已具有此角色，则无需授予其他 Data Lineage 角色。
  • Data Lineage Events Producer (roles/datalineage.producer)：自定义服务账号发布沿袭事件所需的最低独立 Data Lineage 角色。如果服务账号没有 Dataproc Worker 角色，请授予此角色。
  • Data Lineage Editor (roles/datalineage.editor)：除了发布沿袭流程和事件之外，还授予修改和删除沿袭流程和事件的权限。此角色提供的权限超出了标准流水线执行所需的权限，只有在工作负载必须修改现有沿袭元数据时才需要。
  • Data Lineage Administrator (roles/datalineage.admin)：授予对 Data Lineage 资源的完整管理员访问权限。集群服务账号不需要此角色。

启用 Spark 数据沿袭

您可以在项目或集群级层启用 Spark 数据沿袭。

在项目级层启用 Spark 数据沿袭

在项目级层启用 Spark 数据沿袭后，在项目中的 Managed Service for Apache Spark 集群上运行的后续 Spark 作业将启用 Spark 数据沿袭。

如需在项目级层启用 Spark 数据沿袭，请设置以下自定义项目元数据：

键 值
DATAPROC_LINEAGE_ENABLED true
DATAPROC_CLUSTER_SCOPES https://www.googleapis.com/auth/cloud-platform
只有 2.0 映像版本集群才需要设置此虚拟机访问权限范围。它会在 2.1 及更高版本的映像版本集群上自动设置。

您可以通过将 DATAPROC_LINEAGE_ENABLED 元数据设置为 false 在项目级层停用 Spark 数据沿袭。

在集群级层启用 Spark 数据沿袭

如果您在创建集群时启用 Spark 数据沿袭，则在 Managed Service for Apache Spark 集群上运行的受支持 Spark 作业将启用 Spark 数据沿袭。此设置会覆盖项目级层的所有 Spark 数据沿袭设置：如果 Spark 数据沿袭在项目级层处于停用状态，但在集群级层处于启用状态，则集群级层设置优先，并且在集群上运行的受支持 Spark 作业将启用数据沿袭。

如需在集群上启用 Spark 数据沿袭，请创建 Managed Service for Apache Spark 集群，并将 dataproc:dataproc.lineage.enabled 集群属性设置为 true。

gcloud CLI 示例

gcloud dataproc clusters create CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --properties 'dataproc:dataproc.lineage.enabled=true'

您可以在创建集群时将 dataproc:dataproc.lineage.enabled 属性设置为 false，从而在集群上停用 Spark 数据沿袭。

  • 在集群上停用数据沿袭：如需创建停用沿袭的集群，请设置 dataproc:dataproc.lineage.enabled=false。创建集群后，您无法在集群上停用 Spark 数据沿袭。如需在现有集群上停用 Spark 数据沿袭，您可以重新创建集群，并将 dataproc:dataproc.lineage.enabled 属性设置为 false。

  • 在 2.0 映像版本集群上设置范围：Spark 数据沿袭需要 Managed Service for Apache Spark 集群虚拟机 cloud-platform 访问范围。使用映像版本 2.1 及更高版本创建的 Managed Service for Apache Spark 映像版本集群已启用 cloud-platform。如果您在创建集群时指定 Managed Service for Apache Spark 映像版本 2.0，请将范围设置为 cloud-platform。

在作业上停用 Spark 数据沿袭

如果集群上启用了 Spark 数据沿袭，您可以在提交作业时传递值为空 ("") 的 spark.extraListeners 属性，从而在作业上停用 Spark 数据沿袭。

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --class CLASS \
    --jars=gs://APPLICATION_BUCKET/spark-application.jar \
    --properties=spark.extraListeners=''

提交 Spark 作业

当您在启用了 Spark 数据沿袭的 Managed Service for Apache Spark 集群上提交受支持的 Spark 作业时，Managed Service for Apache Spark 会捕获数据沿袭信息并将其报告给 Data Lineage API。

gcloud dataproc jobs submit spark \
    --cluster=CLUSTER_NAME \
    --project PROJECT_ID \
    --region REGION \
    --class CLASS \
    --jars=gs://APPLICATION_BUCKET/spark-application.jar \
    --properties=spark.openlineage.namespace=CUSTOM_NAMESPACE,spark.openlineage.appName=CUSTOM_APPNAME

注意：

  • 添加用于唯一标识作业的 spark.openlineage.namespace 和 spark.openlineage.appName 属性是可选的。如果您不添加这些属性，Managed Service for Apache Spark 将使用以下默认值：
    • spark.openlineage.namespace 的默认值：PROJECT_ID
    • spark.openlineage.appName 的默认值：spark.app.name

在 Knowledge Catalog 中查看沿袭

沿袭图显示了项目资源与创建这些资源的进程之间的关系。您可以在 Google Cloud 控制台中查看数据沿袭信息，也可以从 Data Lineage API 中以 JSON 数据的形式检索该信息。

PySpark 示例代码：

以下 PySpark 作业从公共 BigQuery 表中读取数据，然后将输出写入现有 BigQuery 数据集中的新表。它使用 Cloud Storage 存储桶进行临时存储。

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

spark = SparkSession \
  .builder \
  .appName('LINEAGE_BQ_TO_BQ') \
  .getOrCreate()

bucket = 'gs://BUCKET`
spark.conf.set('temporaryCloudStorageBucket', bucket)

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

word_count = spark.sql('SELECT word, SUM(word_count) AS word_count FROM words GROUP BY word')

destination_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .save()

进行以下替换：

  • BUCKET：现有 Cloud Storage 存储桶的名称

  • PROJECT_ID、DATASET 和 TABLE：项目 ID、现有 BigQuery 数据集的名称，以及要在该数据集中创建的新表的名称（该表不得存在）

您可以在 Knowledge Catalog 界面中查看沿袭图。

沿袭图示例

后续步骤