"Managed Service for Apache Spark" is the new name for the product formerly known as "Dataproc on Compute Engine" (cluster deployment) and "Google Cloud Serverless for Apache Spark" (serverless deployment).

Use data lineage with Managed Service for Apache Spark

This document describes how to enable data lineage on Managed Service for Apache Spark batch workloads and interactive sessions at the project, batch workload, or interactive session or runtime profile (session template) level.

Overview

Data lineage is a Knowledge Catalog feature that lets you track how data moves through your systems: where it comes from, where it is passed to, and what transformations are applied to it.

Managed Service for Apache Spark workloads and sessions capture lineage events and publish them to the Knowledge Catalog Data Lineage API. Managed Service for Apache Spark integrates with the Data Lineage API through OpenLineage, using the OpenLineage Spark plugin.

You can access lineage information through Knowledge Catalog, using lineage graphs and the Data Lineage API. For more information, see View lineage graphs in Knowledge Catalog.

Availability

Data lineage, which supports BigQuery and Cloud Storage data sources, is available for workloads and sessions that run with supported Managed Service for Apache Spark runtime versions with the following exceptions and limitations:

Data lineage is not available for SparkR or Spark streaming workloads or sessions.

Before you begin

On the project selector page in the Google Cloud console, select the project to use for your Managed Service for Apache Spark workloads or sessions.

Go to project selector
Enable the Data Lineage API.

Enable the APIs

Upcoming Spark data lineage changes See the Managed Service for Apache Spark release notes for the announcement of a change that will automatically make Spark data lineage available to your projects, batch workloads, and interactive sessions when you enable the Data Lineage API (see Control lineage ingestion for a service) without requiring additional project, batch workload, or interactive session settings.

Required roles

If your batch workload uses the default Managed Service for Apache Spark service account, it has the Dataproc Worker role. This role includes the datalineage.locations.processOpenLineageMessage permission, which the OpenLineage Spark plugin uses to capture and publish lineage events to the Data Lineage API. No additional action is necessary.

If your batch workload or interactive session uses a custom service account, the service account must have permission to publish lineage events. To follow the principle of least privilege, grant the minimum role required for your use case:

Dataproc Worker (roles/dataproc.worker): the standard role required for Managed Service for Apache Spark workload execution. Because this role already includes the datalineage.locations.processOpenLineageMessage permission, if your custom service account already has this role, you don't need to grant an additional Data Lineage role.
Data Lineage Events Producer (roles/datalineage.producer): the minimum standalone Data Lineage role required for a custom service account to publish lineage events. Grant this role if the service account doesn't have the Dataproc Worker role.
Data Lineage Editor (roles/datalineage.editor): grants permissions to edit and delete lineage processes and events, in addition to publishing them. This role provides more permissions than needed for standard pipeline execution, and is only required if workloads must modify existing lineage metadata.
Data Lineage Administrator (roles/datalineage.admin): grants full administrative access to Data Lineage resources. This role is not required for workload service accounts.

Enable Spark data lineage

You can enable Spark data lineage for your project, batch workload, or interactive session.

Enable data lineage at the project level

After you enable Spark data lineage at the project level, subsequent Spark jobs that run in a batch workload or interactive session will have Spark data lineage enabled.

To enable Spark data lineage on your project, set the following custom project metadata:

Key	Value
`DATAPROC_LINEAGE_ENABLED`	`true`

You can disable Spark data lineage for a project by setting the DATAPROC_LINEAGE_ENABLED metadata to false.

Enable data lineage on a Spark batch workload

To enable data lineage on a batch workload, set the spark.dataproc.lineage.enabled property to true when you submit the workload. This setting overrides any Spark data lineage setting at the project level: if Spark data lineage is disabled at the project level but enabled for the batch workload, the batch workload setting takes precedence.

You can disable Spark data lineage on a Spark batch workload by setting the spark.dataproc.lineage.enabled property to false when you submit the workload.

This example uses the gcloud CLI to submit a batch lineage-example.py workload with Spark lineage enabled.

gcloud dataproc batches submit pyspark lineage-example.py \
    --region=REGION \
    --deps-bucket=gs://BUCKET \
    --properties=spark.dataproc.lineage.enabled=true

The following lineage-example.py code reads data from a public BigQuery table, and then writes the output a new table in an existing BigQuery dataset. It uses a Cloud Storage bucket for temporary storage.

#!/usr/bin/env python

from pyspark.sql import SparkSession
import sys

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

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) \
  .option('writeMethod', 'direct') \
  .save()

Replace the following:

REGION: The region to run the workload
BUCKET: The name of an existing Cloud Storage bucket to store dependencies
PROJECT_ID, DATASET, and TABLE: The project ID, the name of an existing BigQuery dataset, and the name of a new table to create in the dataset (the table must not exist)

You can view the lineage graph in the Knowledge Catalog UI.

Spark lineage graph

Enable data lineage on a Spark interactive session or runtime profile (session template)

To enable data lineage on a Spark interactive session or runtime profile (session template), set the spark.dataproc.lineage.enabled property to true when you create the session or runtime profile (session template). For example, on the Create a runtime profile page in the Google Cloud console, under Additional configuration, edit Other customizations and add the spark.dataproc.lineage.enabled property with the value true under Spark properties. This setting overrides any Spark data lineage setting at the project level: if Spark data lineage is disabled at the project level but enabled for the interactive session, the interactive session setting takes precedence.

You can disable Spark data lineage on a Spark interactive session or runtime profile (session template) by setting the spark.dataproc.lineage.enabled property to false when you create the interactive session or runtime profile (session template).

The following PySpark notebook code configures a Managed Service for Apache Spark interactive session with Spark data lineage enabled. It then creates a Spark Connect session that runs a word count query on a public BigQuery Shakespeare dataset, and then writes the output to a new table in an existing BigQuery dataset (see Use the Spark Connect client).

# Configure the Dataproc Serverless interactive session
# to enable Spark data lineage.
from google.cloud.dataproc_v1 import Session

session = Session()
session.runtime_config.properties["spark.dataproc.lineage.enabled"] = "true"

# Create the Spark Connect session.
from google.cloud.managed_spark_connect import ManagedSparkSession

spark = (
    ManagedSparkSession.builder
    .sessionId("lineage-example")
    .sessionConfig(session)
    .getOrCreate()
)

# Run a wordcount query on the public BigQuery Shakespeare dataset.
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')

# Output the results to a BigQuery destination table.
destination_table = 'PROJECT_ID:DATASET.TABLE'
word_count.write.format('bigquery') \
  .option('table', destination_table) \
  .save()