In this lab module, we will repeat what we did with lineage of BigQuery based Airflow DAG, except, we will use Apache Spark on Dataproc on GCE instead. Dataproc on GCE reports lineage to Dataplex if lineage is enabled at cluster or Spark job level.
Successful completion of prior lab modules
~20 minutes
Dataproc on GCE supports that can be enabled at a cluster lebel or at a job level. Once you enable the lineage feature in your Dataproc cluster or Dataproc Spark job, the jobs capture lineage events and publish them to the Dataplex Data Lineage API. Dataproc integrates with the Data Lineage API through OpenLineage, using the OpenLineage Spark plugin.
Data Lineage API and Data Catalog API
https://cloud.google.com/dataproc/docs/guides/lineage#required-roles
To report lineage, the Dataproc cluster must have GCP OAuth scope.
- Cluster level at creation time
- All jobs automatically report lineage unless we explicitly disable at job level.
- To disable specific jobs from reporting lineage, include the Spark property
spark.extraListenerwith value of null. - Once enabled, lineage cannot be disabled; Delete and recreate cluster should you need to
- At job level at submission time
Set the property
spark.extraListenerwith valueio.openlineage.spark.agent.OpenLineageSparkListenerat job submission time.
Supported:
- Data lineage is available for all Dataproc Spark jobs
- Dataproc Compute Engine 2.0.74+ and 2.1.22+ images
- Lineage is available for BigQuery and Cloud Storage data sources.
Not supported:
- SparkR
- Spark Structured Streaming
At the moment, Cloud Composer reports lineage for Spark SQL and Hive QL jobs.
- Create a Dataproc on GCE cluster with lineage enabled
- Run pre-created PySpark scripts that curate Chicago crimes, and then generate Crime trend reports
- Review the lineage graph
- Run an Airflow DAG on Cloud Composer to orchestrate the Spark jobs on the existing Dataproc Spark GCE cluster
- Review the lineage graph with Cloud Composer in the mix
- Understand the nuances
- Learn how to clean up/delete lineage using the lineage API
Paste the below in Cloud Shell-
PROJECT_ID=`gcloud config list --format "value(core.project)" 2>/dev/null`
PROJECT_NBR=`gcloud projects describe $PROJECT_ID | grep projectNumber | cut -d':' -f2 | tr -d "'" | xargs`
LOCATION="us-central1"
SUBNET=lab-snet
SUBNET_URI="projects/$PROJECT_ID/regions/$LOCATION/subnetworks/$SUBNET"
UMSA_FQN="lab-sa@$PROJECT_ID.iam.gserviceaccount.com"
DPGCE_CLUSTER_NM=lineage-enabled-spark-cluster-$PROJECT_NBR
SPARK_BUCKET=dataproc-lineage-spark-bucket-$PROJECT_NBR
SPARK_BUCKET_FQN=gs://$SPARK_BUCKET
DPMS_NM=lab-dpms-$PROJECT_NBR
Paste the below in Cloud Shell-
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
--member=serviceAccount:${UMSA_FQN} \
--role=roles/datalineage.admin
Paste the below in Cloud Shell-
gsutil mb -p $PROJECT_ID -c STANDARD -l $LOCATION -b on $SPARK_BUCKET_FQN
Paste the below in Cloud Shell-
gcloud dataproc clusters create $DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--subnet $SUBNET \
--region $LOCATION \
--enable-component-gateway \
--bucket $SPARK_BUCKET \
--scopes=cloud-platform \
--master-machine-type n1-standard-4 \
--master-boot-disk-size 500 \
--num-workers 2 \
--worker-machine-type n1-standard-4 \
--worker-boot-disk-size 500 \
--image-version 2.1.75-debian11 \
--optional-components JUPYTER \
--dataproc-metastore projects/$PROJECT_ID/locations/$LOCATION/services/$DPMS_NM \
--properties 'dataproc:dataproc.lineage.enabled=true' \
--impersonate-service-account $UMSA_FQN \
--scopes https://www.googleapis.com/auth/cloud-platform
# 2.2 does not support open lineage
This should take 1-2 minutes (or less) to complete.
- Review the code at the location below-
cd ~/dataplex-quickstart-labs/00-resources/scripts/pyspark/chicago-crimes-analytics/
Review the two PySpark scripts
- Copy the PySpark scripts from local to the code bucket (in case you modified anything) -
cd ~/dataplex-quickstart-labs/00-resources/scripts/pyspark/
gsutil cp chicago-crimes-analytics/* gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/
In this section we will curate Chicago crimes with PySpark on Dataproc Serverless - we will dedupe, and augment the crimes data with some temporal attributes for trending.
Run the command below to curate crimes with PySpark-
JOB_NM=curate-crimes-spark-dataproc
JOB_ID=$JOB_NM-$RANDOM
PROJECT_ID=`gcloud config list --format "value(core.project)" 2>/dev/null`
PROJECT_NBR=`gcloud projects describe $PROJECT_ID | grep projectNumber | cut -d':' -f2 | tr -d "'" | xargs`
LOCATION="us-central1"
SUBNET=lab-snet
SUBNET_URI="projects/$PROJECT_ID/regions/$LOCATION/subnetworks/$SUBNET"
UMSA_FQN="lab-sa@$PROJECT_ID.iam.gserviceaccount.com"
DPGCE_CLUSTER_NM=lineage-enabled-spark-cluster-$PROJECT_NBR
SPARK_BUCKET=dataproc-lineage-spark-bucket-$PROJECT_NBR
SPARK_BUCKET_FQN=gs://$SPARK_BUCKET
DPMS_NM=lab-dpms-$PROJECT_NBR
gcloud dataproc jobs submit pyspark gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/curate_crimes.py \
--cluster=$DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--region $LOCATION \
--id $JOB_ID \
--impersonate-service-account $UMSA_FQN \
--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.29.0.jar \
--properties=spark.openlineage.namespace=$PROJECT_ID,spark.openlineage.appName=$JOB_NM \
-- --projectID=$PROJECT_ID --tableFQN="oda_curated_zone.crimes_curated_spark_dataproc" --peristencePath="gs://curated-data-$PROJECT_NBR/crimes-curated-spark-dataproc/"
Visualize the execution in the Dataproc->Jobs UI-
Navigate to the Cloud Storage to check for output files-
Navigate to the Dataplex UI and you should see discovery job capture and create an entity-
Navigate to BigQuery and you should see a BQ external table-
Run the crimes_report.py script to generate the "Crimes by Year" report-
baseName="crimes-by-year-spark-dataproc"
JOB_ID="$baseName-$RANDOM"
reportName='Chicago Crime Trend by Year'
reportDirGcsURI="gs://product-data-${PROJECT_NBR}/$baseName"
reportSQL='SELECT cast(case_year as int) case_year,count(*) AS crime_count FROM oda_curated_zone.crimes_curated_spark_dataproc GROUP BY case_year;'
reportPartitionCount=1
reportTableFQN="oda_product_zone.crimes_by_year_spark_dataproc"
reportTableDDL="CREATE TABLE IF NOT EXISTS ${reportTableFQN}(case_year int, crime_count long) STORED AS PARQUET LOCATION \"$reportDirGcsURI\""
gcloud dataproc jobs submit pyspark gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/crimes_report.py \
--cluster=$DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--region $LOCATION \
--id $JOB_ID \
--impersonate-service-account $UMSA_FQN \
--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.29.0.jar \
--properties=spark.openlineage.namespace=$PROJECT_ID,spark.openlineage.appName=$baseName \
-- --projectNbr=$PROJECT_NBR --projectID=$PROJECT_ID --reportDirGcsURI="$reportDirGcsURI" --reportName="$reportName" --reportSQL="$reportSQL" --reportPartitionCount=$reportPartitionCount --reportTableFQN="$reportTableFQN" --reportTableDDL="$reportTableDDL"
Should take ~2-3 minutes to complete.
- Visualize the execution in the Dataproc->Batches UI
- Navigate to GCS and look for the dataset in the bucket - gs://product-data-YOUR_PROJECT_NUMBER/crimes-by-year-spark-dataproc/
- Then check for entity creation in Dataplex->Product Zone
- Finally check for the external table in BigQuery dataset oda_product_zone.crimes_by_year_spark_dataproc and run a simple query - this is import for BQ external table lineage
Run the crimes_report.py script to generate the "Crimes by Month" report-
baseName="crimes-by-month-spark-dataproc"
JOB_ID="$baseName-$RANDOM"
reportName='Chicago Crime Trend by Month'
reportDirGcsURI="gs://product-data-${PROJECT_NBR}/$baseName"
reportSQL='SELECT case_month,count(*) AS crime_count FROM oda_curated_zone.crimes_curated_spark_dataproc GROUP BY case_month;'
reportPartitionCount=1
reportTableFQN="oda_product_zone.crimes_by_month_spark_dataproc"
reportTableDDL="CREATE TABLE IF NOT EXISTS ${reportTableFQN}(case_month string, crime_count long) STORED AS PARQUET LOCATION \"$reportDirGcsURI\""
gcloud dataproc jobs submit pyspark gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/crimes_report.py \
--cluster=$DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--region $LOCATION \
--id $JOB_ID \
--impersonate-service-account $UMSA_FQN \
--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.29.0.jar \
--properties=spark.openlineage.namespace=$PROJECT_ID,spark.openlineage.appName=$baseName \
-- --projectNbr=$PROJECT_NBR --projectID=$PROJECT_ID --reportDirGcsURI="$reportDirGcsURI" --reportName="$reportName" --reportSQL="$reportSQL" --reportPartitionCount=$reportPartitionCount --reportTableFQN="$reportTableFQN" --reportTableDDL="$reportTableDDL"
Should take ~2-3 minutes to complete.
- Visualize the execution in the Dataproc->Batches UI
- Navigate to GCS and look for the dataset in the bucket - gs://product-data-YOUR_PROJECT_NUMBER/crimes-by-month-spark-dataproc/
- Then check for entity creation in Dataplex->Product Zone
- Finally check for the external table in BigQuery dataset oda_product_zone.crimes_by_month_spark_dataproc and run a simple query - this is import for BQ external table lineage
Run the crimes_report.py script to generate the "Crimes by Day" report-
baseName="crimes-by-day-spark-dataproc"
JOB_ID="$baseName-$RANDOM"
reportName='Chicago Crime Trend by Day'
reportDirGcsURI="gs://product-data-${PROJECT_NBR}/$baseName"
reportSQL='SELECT case_day_of_week,count(*) AS crime_count FROM oda_curated_zone.crimes_curated_spark_dataproc GROUP BY case_day_of_week;'
reportPartitionCount=1
reportTableFQN="oda_product_zone.crimes_by_day_spark_dataproc"
reportTableDDL="CREATE TABLE IF NOT EXISTS ${reportTableFQN}(case_day_of_week string, crime_count long) STORED AS PARQUET LOCATION \"$reportDirGcsURI\""
gcloud dataproc jobs submit pyspark gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/crimes_report.py \
--cluster=$DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--region $LOCATION \
--id $JOB_ID \
--impersonate-service-account $UMSA_FQN \
--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.29.0.jar \
--properties=spark.openlineage.namespace=$PROJECT_ID,spark.openlineage.appName=$baseName \
-- --projectNbr=$PROJECT_NBR --projectID=$PROJECT_ID --reportDirGcsURI="$reportDirGcsURI" --reportName="$reportName" --reportSQL="$reportSQL" --reportPartitionCount=$reportPartitionCount --reportTableFQN="$reportTableFQN" --reportTableDDL="$reportTableDDL"
Should take ~2-3 minutes to complete.
- Visualize the execution in the Dataproc->Batches UI
- Navigate to GCS and look for the dataset in the bucket - gs://product-data-YOUR_PROJECT_NUMBER/crimes-by-day-spark-dataproc/
- Then check for entity creation in Dataplex->Product Zone
- Finally check for the external table in BigQuery dataset oda_product_zone.crimes_by_day_spark_dataproc and run a simple query - this is import for BQ external table lineage
baseName="crimes-by-hour-spark-dataproc"
JOB_ID="$baseName-$RANDOM"
reportName='Chicago Crime Trend by Hour'
reportDirGcsURI="gs://product-data-${PROJECT_NBR}/$baseName"
reportSQL='SELECT CAST(case_hour_of_day AS int) case_hour_of_day,count(*) AS crime_count FROM oda_curated_zone.crimes_curated_spark_dataproc GROUP BY case_hour_of_day;'
reportPartitionCount=1
reportTableFQN="oda_product_zone.crimes_by_hour_spark_dataproc"
reportTableDDL="CREATE TABLE IF NOT EXISTS ${reportTableFQN}(case_hour_of_day int, crime_count long) STORED AS PARQUET LOCATION \"$reportDirGcsURI\""
gcloud dataproc jobs submit pyspark gs://raw-code-${PROJECT_NBR}/pyspark/chicago-crimes-analytics/crimes_report.py \
--cluster=$DPGCE_CLUSTER_NM \
--project $PROJECT_ID \
--region $LOCATION \
--id $JOB_ID \
--impersonate-service-account $UMSA_FQN \
--jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.29.0.jar \
--properties=spark.openlineage.namespace=$PROJECT_ID,spark.openlineage.appName=$baseName \
-- --projectNbr=$PROJECT_NBR --projectID=$PROJECT_ID --reportDirGcsURI="$reportDirGcsURI" --reportName="$reportName" --reportSQL="$reportSQL" --reportPartitionCount=$reportPartitionCount --reportTableFQN="$reportTableFQN" --reportTableDDL="$reportTableDDL"
Should take ~2-3 minutes to complete.
- Visualize the execution in the Dataproc->Batches UI
- Navigate to GCS and look for the dataset in the bucket - gs://product-data-YOUR_PROJECT_NUMBER/crimes-by-hour-spark-dataproc/
- Then check for entity creation in Dataplex->Product Zone
- Finally check for the external table in BigQuery dataset oda_product_zone.crimes_by_hour_spark_dataproc and run a simple query - this is import for BQ external table lineage
- Lineage is captured when there is data processing. And the supported sources are BigQuery external tables and BigQuery native tables.
- In our case, we used a BigQuery native table (oda_raw_zone.crimes_raw) as our source and created 4 external Hive tables (in Dataproc Metastore/Hive Metatsore) on it.
- We persisted data in Cloud Storage.
Therefore, to find our lineage graph in BigQuery UI, we need to navigate to oda_raw_zone.crimes_raw to see the lineage.
Once you run a query on each BigQuery external table created, if you click on the BQ external table in the BigQuery UI and then on the lineage tab, you should see lineage as follows-
4. Orchestrating the above jobs with Apache Airflow on Cloud Composer - and associated lineage graph
Review the DAG specification at dataplex-quickstart-labs/00-resources/scripts/airflow/chicago-crimes-analytics/spark_dataproc_lineage_pipeline.py.
Run the below in Cloud Shell-
PROJECT_ID=`gcloud config list --format "value(core.project)" 2>/dev/null`
PROJECT_NBR=`gcloud projects describe $PROJECT_ID | grep projectNumber | cut -d':' -f2 | tr -d "'" | xargs`
LOCATION="us-central1"
COMPOSER_ENVIRONMENT_NAME=`gcloud composer environments list --locations $LOCATION | grep NAME | cut -d':' -f2`
DAG_DIR=`gcloud composer environments describe $COMPOSER_ENVIRONMENT_NAME --location $LOCATION | grep dagGcsPrefix | cut -d' ' -f4`/chicago-crimes-analytics
cd ~/dataplex-quickstart-labs/00-resources/scripts/airflow/chicago-crimes-analytics
gsutil cp spark_dataproc_lineage_pipeline.py $DAG_DIR/
- Is denoted by a Dataproc icon
- Click on the icon to see the lineage process ID
For each execution of a Dataproc Spark job reporting lineage, you should see a "RUN"
Click on a "RUN" (where it says "MORE") to see specifc details of the Dataproc Spark job
View the job execution details in the Dataproc GUI
In the Spark job command below, we added properties for lineage-
Here is where it manifests-
Its important to have a unique spark.openlineage.appName value for a Dataproc Spark process, but not for Dataproc Spark process runs. If you have unique values per run, there will be an explosion of Dataproc process icons, one for each run.
-
Running the Spark jobs via Airflow on Cloud Composer did not enrich the lineage graph with Airflow task metadata.
This is because Cloud Composer currently, only reports lineage for Dataproc GCE - SparkSQL and HiveQL jobs at the moment. -
Its important to have a unique
spark.openlineage.appNamevalue for a Dataproc Spark process, but not for Dataproc Spark process runs. If you have unique values per run, there will be an explosion of Dataproc process icons, one for each run.
Cleaning up lineage can be done by removing process runs, and then lineage process using the Lineage API.
E.g.
# Delete process run ac9682e0-3934-409e-866b-45c3799d766c
curl -X DELETE -H "Authorization: Bearer $(gcloud auth print-access-token)" -H "Content-Type: application.json" https://datalineage.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/YOUR_PROJECT_LOCATION/processes/cf61a4a1-89f8-0da2-2ba9-846ad44f2bfe/runs/ac9682e0-3934-409e-866b-45c3799d766c
# Delete process cf61a4a1-89f8-0da2-2ba9-846ad44f2bfe
curl -X DELETE -H "Authorization: Bearer $(gcloud auth print-access-token)" -H "Content-Type: application.json" https://datalineage.googleapis.com/v1/projects/YOUR_PROJECT_ID/locations/YOUR_PROJECT_LOCATION/processes/cf61a4a1-89f8-0da2-2ba9-846ad44f2bfe
This concludes the lab module. Proceed to the next module.