Integrate Spark with DataHub with lineage (Data-Engineering)
Closed, ResolvedPublic13 Estimated Story Points
Actions

Assigned To

Authored By

	BTullis
	Apr 26 2022, 12:59 PM

Description

Spike Goal

Determine what the user experience is when integrating DataHub with Spark

Key Questions:

What do we get when we integrate Spark with DataHub? Is this something we want to support?
Evaluate the creation of
- Pipelines
- Tasks
- Lineage between source and destination datasets
Can this play a part in the broader Data-Platform strategy.
Can we just do this for one Airflow/Spark job and see what we can visualize in DataHub

Deprecated: https://datahubproject.io/docs/metadata-integration/java/spark-lineage/
New version: https://datahubproject.io/docs/metadata-integration/java/spark-lineage-beta
New-new version: https://datahubproject.io/docs/metadata-integration/java/acryl-spark-lineage

Spike results:
The Spike has been successfully completed in Q1, demonstrating the usage of the DatahubSparkListener in practise, surfacing column level lineage information in DataHub.

Completion Requirements

Get Spark Lineage working in DataHub
Make Spark based lineage configurable
Enable Spark based lineage for a suitable test Spark job (hive)
Enable Spark based lineage for all suitable Spark jobs in the analytics airflow instance(hive)

Follow ups

Version upgrade Spark and Iceberg and the connector to support Iceberg
Enable Spark based lineage for all remaining Spark jobs using Iceberg tables
Enable for other airflow instances.

Details

Related Changes in Gerrit:

	Subject	Repo	Branch	Lines +/-
	Add airflow connection conf for datahub	operations/puppet	production	+29 -0

Customize query in gerrit

Related Changes in GitLab:

Title	Reference	Author	Source Branch	Dest Branch
Enable Spark Lineage on even more dags	repos/data-engineering/airflow-dags!1059	tchin	enable-even-more-spark-lineage	main
Disable Spark Lineage on questionable Spark operations	repos/data-engineering/airflow-dags!991	tchin	disable-spark-lineage	main
Enable Spark Lineage on a DAG that generates temp files	repos/data-engineering/airflow-dags!976	tchin	temp-file-lineage	main
Enable spark lineage on more dags	repos/data-engineering/airflow-dags!935	tchin	enable-more-lineage	main
Add Datahub Spark lineage support (no iceberg)	repos/data-engineering/airflow-dags!820	tchin	support-spark-lineage	main

Customize query in GitLab

Related Objects
Search...

Status	Assigned	Task
Open	None	T369756 [EPIC] Datahub Improvements
Open	None	T377789 Data Platform Data Lineage
Resolved	tchin	T306896 Integrate Spark with DataHub with lineage (Data-Engineering)
Resolved	tchin	T372899 Ingest a test hive database into datahub
Open	None	T378899 Upgrade to Spark 3.2 to support Spark lineage for Iceberg tables

Event Timeline

There are a very large number of changes, so older changes are hidden. Show Older Changes

• lbowmaker moved this task from Incoming (new tickets) to Data Lineage on the Data-Engineering board.Feb 9 2024, 3:09 PM

• lbowmaker moved this task from Data Lineage to Backlog on the Data-Engineering board.Mar 25 2024, 2:35 PM

• lbowmaker removed the point value 2 for this task.Mar 26 2024, 12:15 AM

• lbowmaker edited projects, added Data-Engineering (Q4 2024 April 1st - June 30th); removed Data-Engineering.Apr 1 2024, 12:21 PM

• lbowmaker set the point value for this task to 8.

• lbowmaker moved this task from Q4 2024 April 1st - June 30th to Data Lineage on the Data-Engineering board.Jul 15 2024, 2:34 PM

• lbowmaker edited projects, added Data-Engineering; removed Data-Engineering (Q4 2024 April 1st - June 30th).

• lbowmaker added a parent task: T369756: [EPIC] Datahub Improvements.Jul 17 2024, 6:16 PM

• lbowmaker edited projects, added Data-Engineering (Q1 2024 July 1st - September 30th); removed Data-Engineering.Jul 17 2024, 6:19 PM

• lbowmaker updated the task description. (Show Details)

• lbowmaker edited parent tasks, added: T369758: [SPIKE] Define process to build out lineage in DataHub; removed: T369756: [EPIC] Datahub Improvements.

• lbowmaker mentioned this in T369758: [SPIKE] Define process to build out lineage in DataHub.Jul 18 2024, 1:19 PM

Ottomata added a subscriber: amastilovic.Jul 18 2024, 4:10 PM

Ottomata added a subscriber: tchin.Jul 23 2024, 8:18 PM

• lbowmaker assigned this task to tchin.Jul 25 2024, 5:21 PM

tchin moved this task from Next Up to In progress on the Data-Engineering (Q1 2024 July 1st - September 30th) board.Aug 1 2024, 4:18 PM

Ottomata updated the task description. (Show Details)Aug 8 2024, 4:21 PM

tchin updated the task description. (Show Details)Aug 12 2024, 6:45 PM

Ahoelzl subscribed.Aug 14 2024, 11:35 PM

I ran a simple spark sql job on a statbox with:

sudo -u analytics-privatedata spark3-sql --jars ./acryl-spark-lineage-0.2.16.jar --conf "spark.extraListeners=datahub.spark.DatahubSparkListener" --master local[12] --driver-memory 8G --conf "spark.datahub.emitter=file" --conf "spark.datahub.file.filename=./il_lineage" -f il_test.hql

il_test.hql

INSERT OVERWRITE TABLE tchin.interlanguage_navigation  
    SELECT /*+ COALESCE(1) */  
        *
    FROM wmf_traffic.interlanguage_navigation
    WHERE day="2024-08-14"
;

Note that these are both iceberg tables.
It outputted these urns:

[
{
    "aspectName" : "dataFlowInfo",
    "entityUrn" : "urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)",
    "entityType" : "dataFlow",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"customProperties\":{\"spark.master\":\"local[12]\",\"jobId\":\"0\",\"processingEngineVersion\":\"3.1.2\",\"appId\":\"local-1723723383169\",\"startedAt\":\"2024-08-15T12:03:02.435Z\",\"spark.app.name\":\"SparkSQL::10.64.21.17\",\"sparkUser\":\"analytics-privatedata\",\"jobDescription\":\"INSERT OVERWRITE TABLE tchin.interlanguage_navigation  \\n    SELECT /*+ COALESCE(1) */  \\n        *\\n    FROM wmf_traffic.interlanguage_navigation\\n    WHERE day=\\\"2024-08-14\\\"\\n\",\"processingEngine\":\"spark\",\"finishedAt\":\"2024-08-15T12:03:12.220Z\"},\"name\":\"SparkSQL::10.64.21.17\"}"
    }
},
{
    "aspectName" : "status",
    "entityUrn" : "urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)",
    "entityType" : "dataFlow",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"removed\":false}"
    }
},
{
    "aspectName" : "dataJobInfo",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"customProperties\":{\"spark.master\":\"local[12]\",\"jobId\":\"0\",\"processingEngineVersion\":\"3.1.2\",\"spark.app.name\":\"SparkSQL::10.64.21.17\",\"jobDescription\":\"INSERT OVERWRITE TABLE tchin.interlanguage_navigation  \\n    SELECT /*+ COALESCE(1) */  \\n        *\\n    FROM wmf_traffic.interlanguage_navigation\\n    WHERE day=\\\"2024-08-14\\\"\\n\",\"processingEngine\":\"spark\"},\"created\":{\"time\":1723723382435},\"name\":\"SparkSQL::10.64.21.17\",\"flowUrn\":\"urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)\",\"type\":{\"string\":\"spark\"}}"
    }
},
{
    "aspectName" : "status",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"removed\":false}"
    }
},
{
    "aspectName" : "dataJobInputOutput",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"inputDatajobs\":[],\"inputDatasetEdges\":[],\"outputDatasetEdges\":[],\"inputDatasets\":[],\"outputDatasets\":[]}"
    }
},
{
    "aspectName" : "dataProcessInstanceInput",
    "entityUrn" : "urn:li:dataProcessInstance:019155eb-8b42-79f7-a071-0d10165278b2",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"inputs\":[]}"
    }
},
{
    "aspectName" : "dataProcessInstanceOutput",
    "entityUrn" : "urn:li:dataProcessInstance:019155eb-8b42-79f7-a071-0d10165278b2",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"outputs\":[]}"
    }
},
{
    "aspectName" : "dataProcessInstanceProperties",
    "entityUrn" : "urn:li:dataProcessInstance:019155eb-8b42-79f7-a071-0d10165278b2",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"name\":\"019155eb-8b42-79f7-a071-0d10165278b2\",\"created\":{\"actor\":\"urn:li:corpuser:datahub\",\"time\":1723723392200}}"
    }
},
{
    "aspectName" : "dataProcessInstanceRunEvent",
    "entityUrn" : "urn:li:dataProcessInstance:019155eb-8b42-79f7-a071-0d10165278b2",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"result\":{\"type\":\"SUCCESS\",\"nativeResultType\":\"COMPLETE\"},\"timestampMillis\":1723723392146,\"status\":\"COMPLETE\"}"
    }
},
{
    "aspectName" : "dataProcessInstanceRelationships",
    "entityUrn" : "urn:li:dataProcessInstance:019155eb-8b42-79f7-a071-0d10165278b2",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"parentTemplate\":\"urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)\",\"upstreamInstances\":[]}"
    }
}
]

I also ran it inserting into datahub so you can see what it looks like:
https://datahub.wikimedia.org/tasks/urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)/Documentation?is_lineage_mode=false

For some reason, it seems like it's not linking the lineage of wmf_traffic.interlanguage_navigation, even though the documentation says it defaults to looking up the tables in the hive data platform for lineage.

When I look at the logs some things jump out at me:

12:03:12.183 [spark-listener-group-shared] DEBUG io.openlineage.spark.agent.lifecycle.plan.StreamingDataSourceV2RelationVisitor - The result of checking whether org.apache.spark.sql.execution.datasources.v2.DataSourceV2ScanRelation is an instance of org.apache.spark.sql.execution.datasources.v2.StreamingDataSourceV2Relation is false
12:03:12.184 [spark-listener-group-shared] DEBUG io.openlineage.spark.agent.lifecycle.OpenLineageRunEventBuilder - Visiting query plan Optional[== Parsed Logical Plan ==
'InsertIntoStatement 'UnresolvedRelation [tchin, interlanguage_navigation], [], false, true, false
+- 'UnresolvedHint COALESCE, [1]
   +- 'Project [*]
      +- 'Filter ('day = 2024-08-14)
         +- 'UnresolvedRelation [wmf_traffic, interlanguage_navigation], [], false

== Analyzed Logical Plan ==

OverwriteByExpression RelationV2[project_family#5, current_project#6, previous_project#7, navigation_count#8L, day#9] spark_catalog.tchin.interlanguage_navigation, true, false
+- Repartition 1, false
   +- Project [project_family#0, current_project#1, previous_project#2, navigation_count#3L, day#4]
      +- Filter (day#4 = cast(2024-08-14 as date))
         +- SubqueryAlias spark_catalog.wmf_traffic.interlanguage_navigation
            +- RelationV2[project_family#0, current_project#1, previous_project#2, navigation_count#3L, day#4] spark_catalog.wmf_traffic.interlanguage_navigation

== Optimized Logical Plan ==
OverwriteByExpression RelationV2[project_family#5, current_project#6, previous_project#7, navigation_count#8L, day#9] spark_catalog.tchin.interlanguage_navigation, true, false
+- Repartition 1, false
   +- Filter (isnotnull(day#4) AND (day#4 = 19949))
      +- RelationV2[project_family#0, current_project#1, previous_project#2, navigation_count#3L, day#4] spark_catalog.wmf_traffic.interlanguage_navigation

== Physical Plan ==
OverwriteByExpression spark_catalog.tchin.interlanguage_navigation, [AlwaysTrue()], [], org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$1983/1139659498@7202009e
+- Coalesce 1
   +- *(1) Filter (isnotnull(day#4) AND (day#4 = 19949))
      +- *(1) ColumnarToRow
         +- BatchScan[project_family#0, current_project#1, previous_project#2, navigation_count#3L, day#4] spark_catalog.wmf_traffic.interlanguage_navigation [filters=day IS NOT NULL, day = 19949]
] with output dataset builders [<function1>, <function1>, <function1>, <function1>, <function1>, <function1>, <function1>, <function1>, <function1>]
12:03:12.187 [spark-listener-group-shared] DEBUG io.openlineage.spark.agent.lifecycle.plan.WriteToDataSourceV2Visitor - The supplied logical plan IS NOT an instance of org.apache.spark.sql.execution.datasources.v2.WriteToDataSourceV2
12:03:12.188 [spark-listener-group-shared] INFO io.openlineage.spark.agent.util.PlanUtils - apply method failed with
java.lang.NoSuchMethodError: org.apache.iceberg.spark.SparkSessionCatalog.icebergCatalog()Lorg/apache/iceberg/catalog/Catalog;
	at io.openlineage.spark3.agent.lifecycle.plan.catalog.IcebergHandler.getIcebergTable(IcebergHandler.java:171)
	at io.openlineage.spark3.agent.lifecycle.plan.catalog.IcebergHandler.getDatasetVersion(IcebergHandler.java:156)
	at io.openlineage.spark3.agent.lifecycle.plan.catalog.CatalogUtils3.getDatasetVersion(CatalogUtils3.java:100)
	at io.openlineage.spark3.agent.utils.DatasetVersionDatasetFacetUtils.extractVersionFromDataSourceV2Relation(DatasetVersionDatasetFacetUtils.java:46)
	at io.openlineage.spark3.agent.utils.DatasetVersionDatasetFacetUtils.includeDatasetVersion(DatasetVersionDatasetFacetUtils.java:86)
	at io.openlineage.spark3.agent.lifecycle.plan.TableContentChangeDatasetBuilder.apply(TableContentChangeDatasetBuilder.java:83)
	at io.openlineage.spark3.agent.lifecycle.plan.TableContentChangeDatasetBuilder.apply(TableContentChangeDatasetBuilder.java:30)
	at io.openlineage.spark.api.AbstractQueryPlanDatasetBuilder$1.apply(AbstractQueryPlanDatasetBuilder.java:97)
	at io.openlineage.spark.api.AbstractQueryPlanDatasetBuilder$1.apply(AbstractQueryPlanDatasetBuilder.java:86)
	at io.openlineage.spark.agent.util.PlanUtils.safeApply(PlanUtils.java:295)
	at io.openlineage.spark.api.AbstractQueryPlanDatasetBuilder.lambda$apply$0(AbstractQueryPlanDatasetBuilder.java:76)
	at java.util.Optional.map(Optional.java:215)
	at io.openlineage.spark.api.AbstractQueryPlanDatasetBuilder.apply(AbstractQueryPlanDatasetBuilder.java:68)
	at io.openlineage.spark.api.AbstractQueryPlanDatasetBuilder.apply(AbstractQueryPlanDatasetBuilder.java:40)
	at io.openlineage.spark.agent.util.PlanUtils.safeApply(PlanUtils.java:295)
	at io.openlineage.spark.agent.lifecycle.OpenLineageRunEventBuilder.lambda$null$29(OpenLineageRunEventBuilder.java:381)
	at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
	at java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:175)
	at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
	at java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150)
	at java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:485)
	at java.util.stream.ReferencePipeline$7$1.accept(ReferencePipeline.java:272)
	at java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1384)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
	at java.util.stream.StreamSpliterators$WrappingSpliterator.forEachRemaining(StreamSpliterators.java:313)
	at java.util.stream.Streams$ConcatSpliterator.forEachRemaining(Streams.java:742)
	at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:482)
	at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:472)
	at java.util.stream.ReduceOps$ReduceOp.evaluateSequential(ReduceOps.java:708)
	at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.util.stream.ReferencePipeline.collect(ReferencePipeline.java:566)
	at io.openlineage.spark.agent.lifecycle.OpenLineageRunEventBuilder.buildOutputDatasets(OpenLineageRunEventBuilder.java:335)
	at io.openlineage.spark.agent.lifecycle.OpenLineageRunEventBuilder.buildRun(OpenLineageRunEventBuilder.java:196)
	at io.openlineage.spark.agent.lifecycle.SparkSQLExecutionContext.end(SparkSQLExecutionContext.java:139)
	at io.openlineage.spark.agent.OpenLineageSparkListener.lambda$sparkSQLExecEnd$4(OpenLineageSparkListener.java:131)
	at io.openlineage.client.circuitBreaker.NoOpCircuitBreaker.run(NoOpCircuitBreaker.java:27)
	at io.openlineage.spark.agent.OpenLineageSparkListener.sparkSQLExecEnd(OpenLineageSparkListener.java:128)
	at io.openlineage.spark.agent.OpenLineageSparkListener.onOtherEvent(OpenLineageSparkListener.java:103)
	at datahub.spark.DatahubSparkListener.onOtherEvent(DatahubSparkListener.java:339)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:100)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
	at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
	at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105)
	at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105)
	at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
	at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96)
	at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1381)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)
12:03:12.191 [spark-listener-group-shared] DEBUG io.openlineage.spark.agent.lifecycle.SparkSQLExecutionContext - Posting event for end 0: {"eventTime":"2024-08-15T12:03:12.146Z","producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","schemaURL":"https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunEvent","eventType":"COMPLETE","run":{"runId":"019155eb-9955-7a97-8414-885778bc84e6","facets":{"parent":{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","_schemaURL":"https://openlineage.io/spec/facets/1-0-1/ParentRunFacet.json#/$defs/ParentRunFacet","run":{"runId":"019155eb-8b42-79f7-a071-0d10165278b2"},"job":{"namespace":"default","name":"spark_sql::10.64.21.17"}},"spark_properties":{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","_schemaURL":"https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunFacet","properties":{"spark.master":"local[12]","spark.app.name":"SparkSQL::10.64.21.17"}},"processing_engine":{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","_schemaURL":"https://openlineage.io/spec/facets/1-1-1/ProcessingEngineRunFacet.json#/$defs/ProcessingEngineRunFacet","version":"3.1.2","name":"spark"},"environment-properties":{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","_schemaURL":"https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/RunFacet","environment-properties":{}}}},"job":{"namespace":"default","name":"spark_sql::10.64.21.17.overwrite_by_expression.spark_catalog_tchin_interlanguage_navigation","facets":{"jobType":{"_producer":"https://github.com/OpenLineage/OpenLineage/tree/1.19.0/integration/spark","_schemaURL":"https://openlineage.io/spec/facets/2-0-3/JobTypeJobFacet.json#/$defs/JobTypeJobFacet","processingType":"BATCH","integration":"SPARK","jobType":"SQL_JOB"}}},"inputs":[],"outputs":[]}

Unsure if the spark datasource class mismatches matter, but it seems like it can't find some method in the iceberg catalog. It might be that we need to use a different version of iceberg to get a working lineage? I'll test with non-iceberg tables and see if we get anything.

I can see that Iceberg for Spark 3.1 does not in fact have an icebergCatalog method but for > Spark 3.3 it does. Going to see if I can use the Spark 3.3 configs from the airflow dags repo

sudo -u analytics-privatedata spark3-sql --jars ./acryl-spark-lineage-0.2.16.jar...

I might be confiused, but are you working with the correct jar here?

Maybe try this jar instead? https://central.sonatype.com/artifact/io.acryl/datahub-spark-lineage

I was reading here: https://datahubproject.io/docs/0.13.1/metadata-integration/java/spark-lineage/#configuration-instructions-standalone-java-applications

We're currently on version 0.13.3 of datahub, so I would perhaps try these: https://repo1.maven.org/maven2/io/acryl/datahub-spark-lineage/0.13.3-6/

I'm using the newer acryl-spark-lineage which works for datahub 0.13.3 https://datahubproject.io/docs/metadata-integration/java/acryl-spark-lineage

Oh yes, I see. Sorry, I got that the wrong way around :-)

Update: Tried using spark 3.3.2 with this:

sudo -u analytics-privatedata spark3-sql \
--master local[12] --driver-memory 8G \
--jars ./iceberg-spark-runtime-3.3_2.12-1.2.1.jar,./acryl-spark-lineage-0.2.16.jar \
--conf "spark.extraListeners=datahub.spark.DatahubSparkListener" \
--conf "spark.datahub.emitter=file" \
--conf "spark.datahub.file.filename=./il_lineage" \
--conf "spark.yarn.archive=hdfs:///user/spark/share/lib/spark-3.3.2-assembly.zip" \
--conf "spark.jars.ivySettings=/etc/maven/ivysettings.xml" \
-f il_test.hql

Got an error:

24/08/16 21:56:56 ERROR Utils: uncaught error in thread spark-listener-group-shared, stopping SparkContext
java.lang.IncompatibleClassChangeError: Implementing class
	at java.lang.ClassLoader.defineClass1(Native Method)
	at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
	at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
	at java.net.URLClassLoader.defineClass(URLClassLoader.java:473)
	at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
	at java.security.AccessController.doPrivileged(Native Method)
	at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	at java.lang.ClassLoader.defineClass1(Native Method)
	at java.lang.ClassLoader.defineClass(ClassLoader.java:756)
	at java.security.SecureClassLoader.defineClass(SecureClassLoader.java:142)
	at java.net.URLClassLoader.defineClass(URLClassLoader.java:473)
	at java.net.URLClassLoader.access$100(URLClassLoader.java:74)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:369)
	at java.net.URLClassLoader$1.run(URLClassLoader.java:363)
	at java.security.AccessController.doPrivileged(Native Method)
	at java.net.URLClassLoader.findClass(URLClassLoader.java:362)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:418)
	at java.lang.ClassLoader.loadClass(ClassLoader.java:351)
	at io.openlineage.spark.agent.util.ReflectionUtils.hasClass(ReflectionUtils.java:47)
	at io.openlineage.spark34.agent.lifecycle.plan.column.MergeIntoIceberg13ColumnLineageVisitor.hasClasses(MergeIntoIceberg13ColumnLineageVisitor.java:32)
	at io.openlineage.spark.agent.lifecycle.Spark3DatasetBuilderFactory.getColumnLevelLineageVisitors(Spark3DatasetBuilderFactory.java:104)
	at io.openlineage.spark.agent.lifecycle.InternalEventHandlerFactory.createColumnLevelLineageVisitors(InternalEventHandlerFactory.java:243)
	at io.openlineage.spark.agent.lifecycle.InternalEventHandlerFactory.createColumnLevelLineageVisitors(InternalEventHandlerFactory.java:55)
	at io.openlineage.spark.agent.lifecycle.OpenLineageRunEventBuilder.<init>(OpenLineageRunEventBuilder.java:175)
	at io.openlineage.spark.agent.lifecycle.ContextFactory.createSparkApplicationExecutionContext(ContextFactory.java:60)
	at io.openlineage.spark.agent.OpenLineageSparkListener.getSparkApplicationExecutionContext(OpenLineageSparkListener.java:231)
	at io.openlineage.spark.agent.OpenLineageSparkListener.lambda$onApplicationStart$20(OpenLineageSparkListener.java:300)
	at io.openlineage.client.circuitBreaker.NoOpCircuitBreaker.run(NoOpCircuitBreaker.java:27)
	at io.openlineage.spark.agent.OpenLineageSparkListener.onApplicationStart(OpenLineageSparkListener.java:298)
	at datahub.spark.DatahubSparkListener.onApplicationStart(DatahubSparkListener.java:94)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:55)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent$(SparkListenerBus.scala:28)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.scheduler.AsyncEventQueue.doPostEvent(AsyncEventQueue.scala:37)
	at org.apache.spark.util.ListenerBus.postToAll(ListenerBus.scala:117)
	at org.apache.spark.util.ListenerBus.postToAll$(ListenerBus.scala:101)
	at org.apache.spark.scheduler.AsyncEventQueue.super$postToAll(AsyncEventQueue.scala:105)
	at org.apache.spark.scheduler.AsyncEventQueue.$anonfun$dispatch$1(AsyncEventQueue.scala:105)
	at scala.runtime.java8.JFunction0$mcJ$sp.apply(JFunction0$mcJ$sp.java:23)
	at scala.util.DynamicVariable.withValue(DynamicVariable.scala:62)
	at org.apache.spark.scheduler.AsyncEventQueue.org$apache$spark$scheduler$AsyncEventQueue$$dispatch(AsyncEventQueue.scala:100)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.$anonfun$run$1(AsyncEventQueue.scala:96)
	at org.apache.spark.util.Utils$.tryOrStopSparkContext(Utils.scala:1381)
	at org.apache.spark.scheduler.AsyncEventQueue$$anon$2.run(AsyncEventQueue.scala:96)

Seems like >0.13.1, they introduced support for Openlineage, and this line jumps out at me: io.openlineage.spark34.agent.lifecycle.plan.column.MergeIntoIceberg13ColumnLineageVisitor.hasClasses(MergeIntoIceberg13ColumnLineageVisitor.java:32) maybe it thinks it's on spark 3.4? I don't know where to get the spark 3.4 assembly zip to test this theory

Just for fun I also tried using the older connector:

sudo -u analytics-privatedata spark3-sql \
--master local[12] --driver-memory 8G \
--jars ./iceberg-spark-runtime-3.3_2.12-1.2.1.jar,./datahub-spark-lineage-0.13.3-6.jar \
--conf "spark.extraListeners=datahub.spark.DatahubSparkListener" \
--conf "spark.datahub.emitter=file" \
--conf "spark.datahub.file.filename=./il_lineage" \
--conf "spark.yarn.archive=hdfs:///user/spark/share/lib/spark-3.3.2-assembly.zip" \
--conf "spark.jars.ivySettings=/etc/maven/ivysettings.xml" \
-f il_test.hql

Exception in thread "main" java.lang.UnsupportedClassVersionError: datahub/spark/DatahubSparkListener has been compiled by a more recent version of the Java Runtime (class file version 55.0), this version of the Java Runtime only recognizes class file versions up to 52.0

I don't know if it is useful, but @xcollazo did some work on using custom spark assemblies and there are some notes in T340861#9136895

In T306896#10072820, @BTullis wrote:

I don't know if it is useful, but @xcollazo did some work on using custom spark assemblies and there are some notes in T340861#9136895

Unfortunately, using Spark 3.3+ only works from a PySpark context, as we need a conda artifact that bundles the whole of Spark. See here for a production example, and take note of L149:

"SPARK_HOME": "venv/lib/python3.10/site-packages/pyspark",  # point to the packaged Spark

So we are telling Skein to run spark3-sql out of the conda environment (3.3) rather than the production one (3.1).

If we want to use Spark 3.3 generally in production, we have to upgrade the production Spark.

(Shameless plug T338057: Upgrade Spark to a version with long term Iceberg support, and with fixes to support Dumps 2.0)

I ran a job using our regular prod configs just without iceberg tables. It ran successfully and outputted this:

[
{
    "aspectName" : "dataFlowInfo",
    "entityUrn" : "urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)",
    "entityType" : "dataFlow",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"customProperties\":{\"spark.master\":\"local[12]\",\"jobId\":\"0\",\"processingEngineVersion\":\"3.1.2\",\"appId\":\"local-1724123844455\",\"startedAt\":\"2024-08-20T03:17:23.782Z\",\"spark.app.name\":\"SparkSQL::10.64.21.17\",\"sparkUser\":\"analytics-privatedata\",\"jobDescription\":\"INSERT INTO tchin.lineage_test  \\n    SELECT\\n        *\\n    FROM wmf.interlanguage_navigation\\n    WHERE date=\\\"2024-08-14\\\"\\n\",\"processingEngine\":\"spark\",\"finishedAt\":\"2024-08-20T03:17:31.513Z\"},\"name\":\"SparkSQL::10.64.21.17\"}"
    }
},
{
    "aspectName" : "status",
    "entityUrn" : "urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)",
    "entityType" : "dataFlow",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"removed\":false}"
    }
},
{
    "aspectName" : "dataJobInfo",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"customProperties\":{\"spark.master\":\"local[12]\",\"jobId\":\"0\",\"processingEngineVersion\":\"3.1.2\",\"spark.app.name\":\"SparkSQL::10.64.21.17\",\"jobDescription\":\"INSERT INTO tchin.lineage_test  \\n    SELECT\\n        *\\n    FROM wmf.interlanguage_navigation\\n    WHERE date=\\\"2024-08-14\\\"\\n\",\"processingEngine\":\"spark\"},\"created\":{\"time\":1724123843782},\"name\":\"SparkSQL::10.64.21.17\",\"flowUrn\":\"urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default)\",\"type\":{\"string\":\"spark\"}}"
    }
},
{
    "aspectName" : "status",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"removed\":false}"
    }
},
{
    "aspectName" : "upstreamLineage",
    "entityUrn" : "urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD)",
    "entityType" : "dataset",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"fineGrainedLineages\":[{\"confidenceScore\":0.5,\"downstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD),project_family)\"],\"upstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),date)\",\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),project_family)\"],\"downstreamType\":\"FIELD_SET\",\"upstreamType\":\"FIELD_SET\"},{\"confidenceScore\":0.5,\"downstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD),current_project)\"],\"upstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),current_project)\",\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),date)\"],\"downstreamType\":\"FIELD_SET\",\"upstreamType\":\"FIELD_SET\"},{\"confidenceScore\":0.5,\"downstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD),previous_project)\"],\"upstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),date)\",\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),previous_project)\"],\"downstreamType\":\"FIELD_SET\",\"upstreamType\":\"FIELD_SET\"},{\"confidenceScore\":0.5,\"downstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD),navigation_count)\"],\"upstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),date)\",\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),navigation_count)\"],\"downstreamType\":\"FIELD_SET\",\"upstreamType\":\"FIELD_SET\"},{\"confidenceScore\":0.5,\"downstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD),date)\"],\"upstreams\":[\"urn:li:schemaField:(urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD),date)\"],\"downstreamType\":\"FIELD_SET\",\"upstreamType\":\"FIELD_SET\"}],\"upstreams\":[{\"type\":\"TRANSFORMED\",\"dataset\":\"urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD)\"}]}"
    }
},
{
    "aspectName" : "dataJobInputOutput",
    "entityUrn" : "urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)",
    "entityType" : "dataJob",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"inputDatajobs\":[],\"inputDatasetEdges\":[{\"destinationUrn\":\"urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD)\",\"lastModified\":{\"actor\":\"urn:li:corpuser:datahub\",\"time\":1724123851500}}],\"outputDatasetEdges\":[{\"destinationUrn\":\"urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD)\",\"lastModified\":{\"actor\":\"urn:li:corpuser:datahub\",\"time\":1724123851500}}],\"inputDatasets\":[],\"outputDatasets\":[]}"
    }
},
{
    "aspectName" : "dataProcessInstanceInput",
    "entityUrn" : "urn:li:dataProcessInstance:01916dca-1750-7480-b365-a311ff5ad22d",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"inputs\":[\"urn:li:dataset:(urn:li:dataPlatform:hive,wmf.interlanguage_navigation,PROD)\"]}"
    }
},
{
    "aspectName" : "dataProcessInstanceOutput",
    "entityUrn" : "urn:li:dataProcessInstance:01916dca-1750-7480-b365-a311ff5ad22d",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"outputs\":[\"urn:li:dataset:(urn:li:dataPlatform:hive,tchin.lineage_test,PROD)\"]}"
    }
},
{
    "aspectName" : "dataProcessInstanceProperties",
    "entityUrn" : "urn:li:dataProcessInstance:01916dca-1750-7480-b365-a311ff5ad22d",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"name\":\"01916dca-1750-7480-b365-a311ff5ad22d\",\"created\":{\"actor\":\"urn:li:corpuser:datahub\",\"time\":1724123851500}}"
    }
},
{
    "aspectName" : "dataProcessInstanceRunEvent",
    "entityUrn" : "urn:li:dataProcessInstance:01916dca-1750-7480-b365-a311ff5ad22d",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"result\":{\"type\":\"SUCCESS\",\"nativeResultType\":\"COMPLETE\"},\"timestampMillis\":1724123851448,\"status\":\"COMPLETE\"}"
    }
},
{
    "aspectName" : "dataProcessInstanceRelationships",
    "entityUrn" : "urn:li:dataProcessInstance:01916dca-1750-7480-b365-a311ff5ad22d",
    "entityType" : "dataProcessInstance",
    "changeType" : "UPSERT",
    "aspect" : {
        "contentType" : "application/json",
        "value" : "{\"parentTemplate\":\"urn:li:dataJob:(urn:li:dataFlow:(spark,SparkSQL::10.64.21.17,default),SparkSQL::10.64.21.17)\",\"upstreamInstances\":[]}"
    }
}
]

Note the addition of the upstreamLineage object. This means that lineage is being emitting. So the issue before is purely because of Iceberg.

You can see the working pipeline in Datahub now

Screenshot 2024-08-19 at 11.50.47 PM.png (598×2 px, 95 KB)

It seems like right now, unless we upgrade to at least Spark 3.4 and Iceberg 1.4, we will not be able to use Datahub's spark lineage connector on iceberg tables

Ottomata renamed this task from Spike: Integrate Spark with DataHub to Spike: Integrate Spark with DataHub with lineage.Aug 20 2024, 2:25 PM

Looks like we might need to prioritise: T338057: Upgrade Spark to a version with long term Iceberg support, and with fixes to support Dumps 2.0 then?

Ottomata mentioned this in T372899: Ingest a test hive database into datahub.Aug 20 2024, 2:30 PM

Yeah I think we should prioritize that.

I also tested joins and they work:

In T306896#10077861, @tchin wrote:

Yeah I think we should prioritize that.

I also tested joins and they work:

That is pretty cool!

bking subscribed.Aug 20 2024, 2:49 PM

tchin opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/820

Add Datahub Spark lineage support (no iceberg)

I don't know how/where spark's appName is autogenerated, but for dags to use spark lineage we should make it required for them to also define a static appName or else there will be a new pipeline + task(s) for every dag run

Screenshot 2024-09-12 at 8.16.44 AM.png (714×1 px, 87 KB)

This looks like it is happening because of the way the default SparkSubmitOperator name is set.

https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/blob/main/wmf_airflow_common/operators/spark.py#L51

We should change that to be named after the dag+task name, but not necessarily the task instance.

We had set the name with date information on purpose, to facilitate identifying tasks in Yarn (there is a bug, hourly tasks are named with day data, and this should be corrected).
I understand it would make things better for lineage to name them without the date-moving part. But looking in the Datahub spark doc, I found that we can set the flow name explicitely using spark.datahub.flow_name.

Wathever direction we prefer to take, we should make and log an official decision on this, probably involving Data product.

Nice! that sounds good! That is going to be a job specific setting then, so hm. That will make the discussion around how to parameterize more annoying.

Ahoelzl edited projects, added Data-Engineering (Q2 2024 October 1st - December 31th); removed Data-Engineering (Q1 2024 July 1st - September 30th).Oct 9 2024, 1:51 PM

Ahoelzl moved this task from Next Up to In progress on the Data-Engineering (Q2 2024 October 1st - December 31th) board.

Ahoelzl closed subtask T372899: Ingest a test hive database into datahub as Resolved.Oct 9 2024, 2:01 PM

Ottomata moved this task from In progress to In Review on the Data-Engineering (Q2 2024 October 1st - December 31th) board.Oct 21 2024, 2:13 PM

Snwachukwu subscribed.Oct 21 2024, 2:15 PM

tchin renamed this task from Spike: Integrate Spark with DataHub with lineage to Integrate Spark with DataHub with lineage.Oct 21 2024, 2:26 PM

tchin updated the task description. (Show Details)

Ahoelzl edited parent tasks, added: T377789: Data Platform Data Lineage; removed: T369758: [SPIKE] Define process to build out lineage in DataHub, T299910: Data Catalog MVP.Oct 21 2024, 10:54 PM

Ahoelzl updated the task description. (Show Details)Oct 21 2024, 10:58 PM

Change #1085449 had a related patch set uploaded (by Ottomata; author: Ottomata):

[operations/puppet@production] Add airflow connection conf for datahub

https://gerrit.wikimedia.org/r/1085449

Ahoelzl updated the task description. (Show Details)Nov 1 2024, 3:57 PM

Ahoelzl changed the point value for this task from 8 to 13.

Change #1085449 abandoned by Ottomata:

[operations/puppet@production] Add airflow connection conf for datahub

https://gerrit.wikimedia.org/r/1085449

tchin merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/820

Add Datahub Spark lineage support (no iceberg)

Maintenance_bot removed a project: Patch-For-Review.Nov 10 2024, 4:30 PM

Ahoelzl updated the task description. (Show Details)Nov 13 2024, 3:11 PM

First integration results:
https://datahub.wikimedia.org/tasks/urn:li:dataJob:(urn:li:dataFlow:(spark,airflow_interlanguage_daily__filter_pageview_actor,default),airflow_interlanguage_daily__filter_pageview_actor)/Lineage?filter_degree___false___EQUAL___0=1&is_lineage_mode=false&page=1&unionType=0

I went through all the dags in the analytics instance for a surface-level evaluation on whether we can apply spark lineage to it

DAG	Uses easy_dag?	Uses Spark?	Uses Iceberg?	Other Notes	Can apply Spark Lineage?
AnomalyDetectionDAG	No	Yes	No	Special Custom DAG class	No
apis_metrics_to_graphite_hourly	No	Yes	No		No
aqs_hourly	Yes	Yes	Yes, for one operation	Has manual lineage from DatahubLineageEmitterOperator	Done (manually)
browser_general_daily	Yes	Yes	Yes		No
browser_metrics_weekly	Yes	Yes	Yes	Operator generates a temp file. Unsure how that appears in DataHub	No
canary_events	Yes	No	-		No
load_commons_impact_metrics	Yes	Yes	Yes	Uses create_easy_cassandra_loading_dag	No
cassandra_load_editors_by_country_monthly	Yes	Yes	No	Uses create_easy_cassandra_loading_dag	Cassandra tables aren't in datahub, should probably not enable until they are
cassandra_load_mediarequest_per_file_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_mediarequest_per_referer_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_mediarequest_top_files_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_pageview_per_article_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_pageview_per_project_hourly	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_pageview_top_articles_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_pageview_top_by_country_monthly	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_pageview_top_per_country_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
cassandra_load_unique_devices_daily	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
create_easy_cassandra_loading_dag	Yes	Yes	No	Uses create_easy_cassandra_loading_dag
clickstream_monthly	Yes	Yes	No	SQL dynamically generated using a java class. It also creates files.	Maybe?
commons_impact_metrics_dumps_monthly	Yes	Yes	Yes	Operator generates a temp file.	No
commons_impact_metrics_monthly	Yes	Yes	Yes		No
metadata_ingest_daily	No	No	-	Datahub ingestion dag	No
druid_load_banner_activity_minutely_aggregated_daily	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
druid_load_edit_hourly	No	No	-		No
druid_load_editattemptstep	No	No	-		No
druid_load_geoeditors_monthly	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
druid_load_navigationtiming	No	No	-		No
druid_load_netflow	No	No	-		No
druid_load_network_flows_internal	No	No	-		No
druid_load_pageviews_daily_aggregated_monthly	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
druid_load_pageviews_hourly	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
druid_load_prefupdate	No	No	-		No
druid_load_unique_devices_per_domain_daily	Yes	Yes	Yes		No
druid_load_unique_devices_per_domain_monthly	Yes	Yes	Yes		No
druid_load_unique_devices_per_project_family_daily	Yes	Yes	Yes		No
druid_load_unique_devices_per_project_family_monthly	Yes	Yes	Yes		No
druid_load_virtualpageview_daily	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
druid_load_webrequest_sampled_128_hourly	Yes	Yes	No	Uses an intermediate temp table and creates files in a temp directory	Needs mechanism to handle temp files
commons_structured_data_dump_to_hive_weekly	No	Yes	No(?)	Uses a java class to read from a file and put into a table. Unsure how it would be reflected in lineage	No
wikidata_dump_to_hive_weekly	No	Yes	No(?)	See above	No
dumps_merge_backfill_to_wikitext_raw	Yes	Yes	Yes	Dag still in development. Uses python script, creates and deletes intermediate tables and files	No
dumps_merge_events_to_wikitext_raw_daily	Yes	Yes	Yes	Dag still in development	Not yet
dumps_publish_wikitext_raw_to_xml	Yes	Yes	Yes	Dag still in development	Not yet
dumps_reconcile_wikitext_raw_daily	Yes	Yes	Yes	Dag still in development	Not yet
edit_hourly	No	Yes	No		Potentially in the future
editors_daily_monthly	No	Yes	No		Potentially in the future
geoeditors_edits_monthly	No	Yes	No		Potentially in the future
geoeditors_monthly	No	Yes	No		Potentially in the future
geoeditors_public_monthly	No	Yes	No	Creates/deletes files	Needs mechanism to handle temp files
geoeditors_yearly	No	Yes	No	Creates/deletes files	Needs mechanism to handle temp files
unique_editors_by_country_monthly	No	Yes	No		Potentially in the future
hdfs_usage_weekly	Yes	Yes	No	Uses a java class	Potentially in the future
interlanguage_daily	Yes	Yes	Yes, for one operation		Done
mediacounts_archive_daily	No	Yes	No	Generates temp files	Needs mechanism to handle temp files
mediacounts_load_hourly	No	Yes	No		Potentially in the future
mediarequest_hourly	No	Yes	No		Potentially in the future
mediawiki_history_check_denormalize	No	Yes	No
mediawiki_history_denormalize	No	Yes	No	Generates temp files	Needs mechanism to handle temp files
mediawiki_history_dumps	No	Yes	No	Generates files	Needs mechanism to handle temp files
mediawiki_history_load	No	Yes	No	Creating partitions over files
mediawiki_history_metrics_monthly	Yes	Yes	Yes (metrics)	Outputs metrics into tables
mediawiki_history_reduced	Yes	Yes	No		Yes, partially
mediawiki_history_shapshot_config	Yes	Yes	No	Uses create_easy_cassandra_loading_dag	Potentially in the future
dump_day_of_hourly_pageviews	No	Yes	No	Generates temp files	Needs mechanism to handle temp files
dump_month_of_daily_pageviews	No	Yes	No	Generates temp files	Needs mechanism to handle temp files
pageview_actor_hourly	No	Yes	No		Potentially in the future
pageview_allowlist_check	No	Yes	No		Potentially in the future
pageview_hourly	No	Yes	No	Generates temp files	Potentially in the future
pingback_report_weekly_v2	Yes	Yes	No	Generates temp files
projectview_geo	No	Yes	No	Generates temp files
projectview_hourly	No	Yes	No	Generates temp files
referrer_daily	Yes	Yes	Yes		No
refine_hourly	Yes	Yes	Unsure
refine_to_hive_hourly	Yes	Yes	Unsure
session_length_daily	Yes	Yes	Yes, for one operation		Yes, partially
table_maintenance_iceberg_	No	Yes	Yes		No
unique_devices_	Yes	Yes	Yes, for one operation	Generates temp files	Yes for one operation
virtualpageview_hourly	No	Yes	No		Potentially in the future
webrequest_actor_label_hourly	No	Yes	No		Potentially in the future
webrequest_actor_metrics_analyzer_hourly	Yes	Yes	No		Yes
webrequest_actor_metrics_hourly	No	Yes	No		Potentially in the future
webrequest_actor_metrics_rollup_hourly	No	Yes	No		Potentially in the future
webrequest_analyzer	Yes	Yes	No	Think this dag is still in development
wikidata_coeditors_metrics_to_graphite_monthly	No	Yes	No	Graphite is deprecated
wikidata_metrics_to_graphite_daily	No	Yes	No	Graphite is deprecated
wmcs_report_monthly	Yes	Yes	No	Generates temp files	Needs mechanism to handle temp files

tchin opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/935

Enable spark lineage on more dags

tchin merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/935

Enable spark lineage on more dags

Maintenance_bot removed a project: Patch-For-Review.Nov 26 2024, 4:31 PM

Thomas and I just checked a few things to see how we should move forward. tl;dr

spark jobs that write to files (instead of tables) might end up with some incomplete lineage info, but nothing will break.

Screenshot 2024-12-16 at 12.17.01.png (262×886 px, 19 KB)

iceberg jobs shouldn't break, but there will be error logs in the job, and maybe incomplete lineage info.

Next steps:

Enable lineage for a spark DAG that uses both Hive tables and writes to output files and verify that everything is okay
Enable lineage by default for all analytics instance dags in dag_config, but manually disable lineage for any DAG (or operator?) that uses Iceberg.

How we verified file lineage doesn't break:

$ cat ./spark_datahub_otto_test2.hql



INSERT INTO otto.p1
select 
    meta.domain as name,
    1 as val,
    null as s
from event.navigationtiming    
limit 10
;    


INSERT OVERWRITE DIRECTORY "./spark_datahub_otto_test1"
    USING csv
    OPTIONS ('compression' 'uncompressed', 'sep' ' ')
SELECT * FROM otto.p1 
limit 10
;

$ spark3-sql \
--jars hdfs://analytics-hadoop/wmf/cache/artifacts/airflow/analytics/acryl-spark-lineage-0.2.16.jar \
--conf "spark.extraListeners=datahub.spark.DatahubSparkListener" \
--conf "spark.datahub.emitter=rest" \
--conf "spark.datahub.rest.server=https://datahub-gms.discovery.wmnet:30443" \
-f ./spark_datahub_otto_test2.hql

tchin opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/976

Enable Spark Lineage on a DAG that generates temp files

tchin merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/976

Enable Spark Lineage on a DAG that generates temp files

Maintenance_bot removed a project: Patch-For-Review.Dec 18 2024, 9:30 AM

tchin opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/991

Disable Spark Lineage on questionable Spark operations

Ottomata mentioned this in T383743: Identify Internal Users of MediaWiki Wikitext Tables .Jan 15 2025, 3:04 PM

Ahoelzl edited projects, added Data-Engineering (Q3 2025 January 1st - March 31th); removed Data-Engineering (Q2 2024 October 1st - December 31th).Jan 18 2025, 12:32 AM

Ahoelzl moved this task from Next Up to In Review on the Data-Engineering (Q3 2025 January 1st - March 31th) board.

tchin merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/991

Disable Spark Lineage on questionable Spark operations

Maintenance_bot removed a project: Patch-For-Review.Jan 23 2025, 5:31 PM

tchin opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1059

Enable Spark Lineage on even more dags

tchin merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1059

Enable Spark Lineage on even more dags

Maintenance_bot removed a project: Patch-For-Review.Feb 12 2025, 7:30 PM

Mentioned in SAL (#wikimedia-operations) [2025-02-13T03:39:09Z] <tchin@deploy2002> Started deploy [airflow-dags/analytics@aaba3ff]: Deploying airflow for T306896

Mentioned in SAL (#wikimedia-operations) [2025-02-13T03:40:04Z] <tchin@deploy2002> Finished deploy [airflow-dags/analytics@aaba3ff]: Deploying airflow for T306896 (duration: 01m 07s)

Column-level data lineage of the newly deployed Spark jobs can be queried and navigated in DataHub:
https://datahub.wikimedia.org/dataset/urn:li:dataset:(urn:li:dataPlatform:hive,wmf.webrequest,PROD)/Lineage?end_time_millis&is_lineage_mode=true&separate_siblings=false&show_columns=false&start_time_millis

More enabled jobs will become visible as they execute and submit data in their upcoming run.

Ahoelzl renamed this task from Integrate Spark with DataHub with lineage to Integrate Spark with DataHub with lineage (Data-Engineering).Feb 14 2025, 6:26 PM

Ahoelzl updated the task description. (Show Details)

Ahoelzl moved this task from In Review to Done on the Data-Engineering (Q3 2025 January 1st - March 31th) board.Feb 14 2025, 9:59 PM

Ottomata updated the task description. (Show Details)Feb 18 2025, 4:16 PM

Ahoelzl closed this task as Resolved.Mar 5 2025, 12:13 AM

Ahoelzl triaged this task as High priority.

	F58024183: Screenshot 2024-12-16 at 12.17.01.png
	Dec 16 2024, 5:20 PM

	F57502804: Screenshot 2024-09-12 at 8.16.44 AM.png
	Sep 12 2024, 12:20 PM

	F57282161: Screenshot 2024-08-19 at 11.50.47 PM.png
	Aug 20 2024, 3:52 AM

	F57282734: image.png
	Aug 20 2024, 2:42 PM

	F35070143: image.png
	Apr 27 2022, 11:41 AM

Integrate Spark with DataHub with lineage (Data-Engineering)Closed, ResolvedPublic13 Estimated Story PointsActions