Enable Spark data lineage for all Airflow instances
Open, Needs TriagePublic
Actions

Assigned To

Authored By

	Ahoelzl
	Feb 19 2025, 6:59 PM

Description

After successfully delivering Spark based column-level data lineage for Data Engineering, we want to provide data lineage to all other data platform Airflow instances.

https://wikitech.wikimedia.org/wiki/Data_Platform/Systems/Airflow/Instances

Stakeholders

Implementation Steps

list out all Airflow instances and stakeholders
coordinate with stakeholders (Slack)
for each instance list out DAGs and mark suitable ones
submit data lineage integration patches for suitable DAGs
roll out changes and monitor

Details

Other Assignee: Ahoelzl

Related Changes in Gerrit:

	Subject	Repo	Branch	Lines +/-
	Add a connection for kafka-jumbo-eqiad to be used by systems outside Kubernetes	operations/deployment-charts	master	+9 -0

Customize query in gerrit

Related Changes in GitLab:

Title	Reference	Author	Source Branch	Dest Branch
Draft: Set DataHub Kafka connection depending on execution environment	repos/data-engineering/airflow-dags!1340	mforns	fix-spark-lineage-for-kubernetes	main
Add port to spark.datahub.kafka.bootstrap	repos/data-engineering/airflow-dags!1205	tchin	add-port-spark-lineage	main
Enable spark lineage for search/dags/query_clicks.py	repos/data-engineering/airflow-dags!1190	tchin	enable-spark-lineage-query-clicks	main
Add spark lineage config to search instance	repos/data-engineering/airflow-dags!1125	tchin	add-spark-lineage-search	main

Customize query in GitLab

Related Objects
Search...

Status	Assigned	Task
Open	None	T369756 [EPIC] Datahub Improvements
Open	None	T377789 Data Platform Data Lineage
Open	mforns	T386862 Enable Spark data lineage for all Airflow instances

Event Timeline

Ahoelzl created this task.Feb 19 2025, 6:59 PM

Restricted Application added a subscriber: Aklapper. · View Herald TranscriptFeb 19 2025, 6:59 PM

Ahoelzl updated the task description. (Show Details)Feb 19 2025, 7:10 PM

Ahoelzl added a parent task: T377789: Data Platform Data Lineage.Feb 19 2025, 10:58 PM

Planning document:
https://docs.google.com/spreadsheets/d/1QduhKl4pK0kgeRPNnXMNwk0cCVlhyq5q-l4k3Ka4o84/edit?gid=0#gid=0

Ahoelzl updated the task description. (Show Details)Feb 19 2025, 11:23 PM

Ahoelzl updated Other Assignee, added: Ahoelzl.Feb 20 2025, 2:45 PM

Ahoelzl merged a task: T386724: Integrate Spark with DataHub with lineage (non Data-Engineering Airflow instances).

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

Ahoelzl added a subscriber: tchin.

Ahoelzl updated the task description. (Show Details)Mar 5 2025, 2:14 PM

Ahoelzl added subscribers: AndrewTavis_WMDE, isarantopoulos, • fab and 3 others.

Ahoelzl updated the task description. (Show Details)Mar 5 2025, 2:21 PM

KCVelaga_WMF subscribed.Mar 10 2025, 9:36 AM

Ahoelzl claimed this task.Mar 11 2025, 9:08 PM

Ahoelzl updated Other Assignee, added: tchin; removed: Ahoelzl.

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

Add spark lineage config to search instance

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

Add spark lineage config to search instance

Maintenance_bot removed a project: Patch-For-Review.Mar 12 2025, 4:30 PM

Fixed k8s DatahubEventEmitter issue and re-deploy on all Airflow instances.

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

Enable spark lineage for search/dags/query_clicks.py

dcausse merged https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1190

Enable spark lineage for search/dags/query_clicks.py

Maintenance_bot removed a project: Patch-For-Review.Mar 28 2025, 8:30 AM

Currently encountering this error, with the search instance, think it's the same issue we encountered when migrating the analytics instance to k8s:

25/04/01 15:44:20 ERROR AsyncEventQueue: Listener DatahubSparkListener threw an exception
datahub.shaded.org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
	at datahub.client.kafka.KafkaEmitter.<init>(KafkaEmitter.java:55)
	at datahub.spark.DatahubEventEmitter.getEmitter(DatahubEventEmitter.java:85)
	at datahub.spark.DatahubEventEmitter.emitMcps(DatahubEventEmitter.java:401)
	at datahub.spark.DatahubEventEmitter.emitCoalesced(DatahubEventEmitter.java:190)
	at datahub.spark.DatahubSparkListener.onApplicationEnd(DatahubSparkListener.java:279)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:57)
	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)
Caused by: datahub.shaded.org.apache.kafka.common.config.ConfigException: Invalid url in bootstrap.servers: kafka-jumbo-eqiad.external-services.svc.cluster.local
	at datahub.shaded.org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:54)
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:408)
	... 20 more

Actually nevermind, the error we had last time was

Caused by: datahub.shaded.org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
        at datahub.shaded.org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:84)
        at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:408)
        ... 20 more

@brouberol would you happen to have some insight into this issue?

The previous issue you linked to in https://phabricator.wikimedia.org/T386862#10701370 was fixed by introducing the following patch:

    datahub_kafka_jumbo:
      conn_type: datahub_kafka
-     host: kafka-jumbo-eqiad.external-services.svc.cluster.local:9092
+     host: kafka-jumbo-eqiad.external-services.svc.cluster.local

What I'm reading here is

_extra_default_args = {
    **instance_default_args,
    # Arguments for operators that have artifact dependencies
    "hdfs_tools_shaded_jar_path": artifact("hdfs-tools-0.0.6-shaded.jar"),
    "spark_sql_driver_jar_path": artifact("wmf-sparksqlclidriver-1.0.0.jar"),
    "spark_datahub_lineage_enabled": False,
    "spark_datahub_lineage_jar": artifact("acryl-spark-lineage-0.2.16.jar"),
    "spark_datahub_lineage_conf": {
        "spark.datahub.emitter": "kafka",
        "spark.extraListeners": "datahub.spark.DatahubSparkListener",
        "spark.datahub.kafka.bootstrap": "{{ conn.datahub_kafka_jumbo.host }}", # <--
        "spark.datahub.kafka.schema_registry_url": "{{ conn.datahub_kafka_jumbo.extra_dejson.connection.schema_registry_url }}",  # noqa: E501
        "spark.datahub.flow_name": "airflow_{{ dag.dag_id }}__{{ task.task_id }}",
        "spark.datahub.log.mcps": "false",
    },
}

which is consumed by the spark config builder. IMO bootstrap.server should contain the port (cf my own experience and this answer), so we might be missing :9092 here.

What I don't understand is how things work over in airflow-main in the first place.

Hit me up when you're around and we'll investigate.

What I'm thinking atm is that multiple things use datahub_kafka_jumbo.host: the python Kafka client used by airflow to interact with datahub as a hook, which concatenates host and port, and the spark client, that assumes that both hostname and port are found in conn.datahub_kafka_jumbo.host.

My hypothesis is that things might actually be broken in airflow-main (we'll see if I'm right or not). If so, I think a proper fix would be to have

_extra_default_args = {
    **instance_default_args,
    # Arguments for operators that have artifact dependencies
    "hdfs_tools_shaded_jar_path": artifact("hdfs-tools-0.0.6-shaded.jar"),
    "spark_sql_driver_jar_path": artifact("wmf-sparksqlclidriver-1.0.0.jar"),
    "spark_datahub_lineage_enabled": False,
    "spark_datahub_lineage_jar": artifact("acryl-spark-lineage-0.2.16.jar"),
    "spark_datahub_lineage_conf": {
        "spark.datahub.emitter": "kafka",
        "spark.extraListeners": "datahub.spark.DatahubSparkListener",
-       "spark.datahub.kafka.bootstrap": "{{ conn.datahub_kafka_jumbo.host }}", 
+       "spark.datahub.kafka.bootstrap": "{{ conn.datahub_kafka_jumbo.host }}:{{ conn.datahub_kafka_jumbo.port }}", 
        "spark.datahub.kafka.schema_registry_url": "{{ conn.datahub_kafka_jumbo.extra_dejson.connection.schema_registry_url }}",  # noqa: E501
        "spark.datahub.flow_name": "airflow_{{ dag.dag_id }}__{{ task.task_id }}",
        "spark.datahub.log.mcps": "false",
    },
}

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

Add port to spark.datahub.kafka.bootstrap

I think it actually is broken on main as well and it's just been silently failing. I opened a patch to add the port back in

Ahoelzl edited projects, added Data-Engineering (Q4 2025 April 1st - June 30th); removed Data-Engineering (Q3 2025 January 1st - March 31th).Apr 3 2025, 2:44 PM

Ahoelzl moved this task from Next Up to In progress on the Data-Engineering (Q4 2025 April 1st - June 30th) board.

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

Add port to spark.datahub.kafka.bootstrap

Maintenance_bot removed a project: Patch-For-Review.Apr 8 2025, 12:30 AM

How did it go?

Ok so that wasn't it, now we get this error:

25/04/08 19:42:35 ERROR AsyncEventQueue: Listener DatahubSparkListener threw an exception
datahub.shaded.org.apache.kafka.common.KafkaException: Failed to construct kafka producer
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:430)
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:298)
	at datahub.client.kafka.KafkaEmitter.<init>(KafkaEmitter.java:55)
	at datahub.spark.DatahubEventEmitter.getEmitter(DatahubEventEmitter.java:85)
	at datahub.spark.DatahubEventEmitter.emitMcps(DatahubEventEmitter.java:401)
	at datahub.spark.DatahubEventEmitter.emitCoalesced(DatahubEventEmitter.java:190)
	at datahub.spark.DatahubSparkListener.onApplicationEnd(DatahubSparkListener.java:279)
	at org.apache.spark.scheduler.SparkListenerBus.doPostEvent(SparkListenerBus.scala:57)
	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)
Caused by: datahub.shaded.org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers
	at datahub.shaded.org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses(ClientUtils.java:84)
	at datahub.shaded.org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:408)
	... 20 more

Ahoelzl reassigned this task from Ahoelzl to mforns.Apr 9 2025, 8:51 PM

Ahoelzl updated Other Assignee, added: Ahoelzl; removed: tchin.

Hmm, we might need to iterate on a test environment so we can figure this out.

mforns moved this task from In progress to Next Up on the Data-Engineering (Q4 2025 April 1st - June 30th) board.Apr 14 2025, 3:12 PM

Throwing out a guess, it seems like because where the lineage runs depends on the driver, it needs to somehow be aware of whether or not it's running on k8s and choose the correct kafka bootstrap url. I wonder if there's an easy way to figure this out? This will probably require some refactoring.

Ahoelzl updated the task description. (Show Details)Apr 28 2025, 9:40 PM

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

Use DataHub rest endpoint instead of kafka for lineage

I think we can make use of the existing methods: k8s_proof_url() or is_running_in_kubernetes().

@brouberol, usually when we use them, we transform an external URL like https://meta.wikimedia.org to something like http://envoy:6501.
I could not find any envoy configuration for datahub, though. Does this mean that datahub is not accessible from inside Kubernetes? 🙏

Ahoelzl moved this task from Next Up to In progress on the Data-Engineering (Q4 2025 April 1st - June 30th) board.May 16 2025, 5:22 PM

@mforns we shoud have an entry in the service mesh for datahub. We did cut corners at the time, and circumvented the mesh when it comes to how we deployed Datahub (we were not using the mesh at the time). We defined a Kubernetes Service resource for Datahub though, meaning it is reachable via vanilla K8s service discovery, at http://datahub-gms-production.datahub.svc:8080

brouberol@deploy1003:~$ k exec -it airflow-scheduler-695fc84dbb-25hhn -- python3
Defaulted container "airflow-production" out of: airflow-production, statsd-exporter
Python 3.10.15 (main, Mar 31 2025, 16:03:46) [GCC 10.2.1 20210110] on linux
Type "help", "copyright", "credits" or "license" for more information.
>>> import socket
>>> socket.gethostbyname('datahub-gms-production.datahub.svc')
'10.67.45.254'

The issue (as I understand it) is that Spark applications running outside of Kubernetes need to talk to either Kafka or Datahub, using a different DNS than when they are running in Kubernetes.

This isn't so much of a configuration issue than a code issue. When you launch spark in cluster ( I think?) mode, then the driver will be running in YARN, meaning that any service discovery that is Kubernetes specific will not work. So your code needs to figure out that when the driver is running in YARN, it should be calling kafka-jumbo100x.eqiad.wmnet and datahub.wikimedia.org instead of kafka-jumbo-eqiad.external-services:9092 and datahub-gms-production.datahub.svc.

I think the solution is to make the code aware of both endpoints, and then pick the correct one inside the SparkSubmitOperator based off of the launcher param before it sets the rest of the config. Right now the endpoint is set by a jinja template, but by the time airflow templates the string it's probably too late?

Thanks @brouberol and @tchin!

OK, I think we can indeed switch the desired URL depending on whether we are in Kubernetes or not.
Working on that.

@brouberol I've written some code that I think might work.
But I think that we would need to define another connection in values-analytics-production.yaml, say, datahub_kafka_jumbo_external, that points to kafka-jumbo1007.eqiad.wmnet:9092.
Would that make sense? 🙏

mforns opened https://gitlab.wikimedia.org/repos/data-engineering/airflow-dags/-/merge_requests/1340

Set DataHub Kafka connection depending on execution environment

Change #1148408 had a related patch set uploaded (by Brouberol; author: Brouberol):

[operations/deployment-charts@master] Add a connection for kafka-jumbo-eqiad to be used by systems outside Kubernetes

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

Change #1148408 merged by Brouberol:

[operations/deployment-charts@master] Add a connection for kafka-jumbo-eqiad to be used by systems outside Kubernetes

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

@mforns I didn't report back, but you should be gtg by the way!

mforns moved this task from In progress to Blocked/Paused on the Data-Engineering (Q4 2025 April 1st - June 30th) board.Jun 23 2025, 3:28 PM

dr0ptp4kt mentioned this in T400207: ERROR AsyncEventQueue: Listener DatahubSparkListener threw an exception.Jul 22 2025, 7:04 PM

Ahoelzl edited projects, added Data-Engineering (Q1 FY25/26 July 1st - September 30th); removed Data-Engineering (Q4 2025 April 1st - June 30th).Jul 25 2025, 10:56 PM

Ahoelzl moved this task from Next Up to Blocked/Paused on the Data-Engineering (Q1 FY25/26 July 1st - September 30th) board.

Ahoelzl edited projects, added Data-Engineering (Q2 FY25/26 October 1st - December 31th); removed Data-Engineering (Q1 FY25/26 July 1st - September 30th).Oct 9 2025, 12:21 AM

Ahoelzl moved this task from Next Up to Blocked/Paused on the Data-Engineering (Q2 FY25/26 October 1st - December 31th) board.

Ahoelzl edited projects, added Data-Engineering; removed Data-Engineering (Q2 FY25/26 October 1st - December 31th).Oct 29 2025, 2:56 AM

Ahoelzl moved this task from Q2 FY25/26 October 1st - December 31th to Next Up on the Data-Engineering board.

Enable Spark data lineage for all Airflow instancesOpen, Needs TriagePublicActions

Description

Details

Related ObjectsSearch...

Event Timeline

Enable Spark data lineage for all Airflow instances
Open, Needs TriagePublic
Actions

Related Objects
Search...