yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-12-12 18:47:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(airflow): Add comma parsing of owners to DataJobs (#10903)

Browse Source
This commit is contained in:
Ellie O'Neil 2024-07-17 14:58:25 -07:00 committed by GitHub
parent 998d7d1227
commit 0b64de8f2b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 11 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
metadata-ingestion-modules/airflow-plugin/src/datahub_airflow_plugin/client/airflow_generator.py
View File

@ -127,6 +127,10 @@ class AirflowGenerator:
) )
return upstream_tasks return upstream_tasks
@staticmethod
def _extract_owners(dag: "DAG") -> List[str]:
return [owner.strip() for owner in dag.owner.split(",")]
@staticmethod @staticmethod
def generate_dataflow( def generate_dataflow(
config: DatahubLineageConfig, config: DatahubLineageConfig,
@ -175,7 +179,7 @@ class AirflowGenerator:
data_flow.url = f"{base_url}/tree?dag_id={dag.dag_id}" data_flow.url = f"{base_url}/tree?dag_id={dag.dag_id}"
if config.capture_ownership_info and dag.owner: if config.capture_ownership_info and dag.owner:
owners = [owner.strip() for owner in dag.owner.split(",")] owners = AirflowGenerator._extract_owners(dag)
if config.capture_ownership_as_group: if config.capture_ownership_as_group:
data_flow.group_owners.update(owners) data_flow.group_owners.update(owners)
else: else:
@ -282,10 +286,12 @@ class AirflowGenerator:
datajob.url = f"{base_url}/taskinstance/list/?flt1_dag_id_equals={datajob.flow_urn.flow_id}&_flt_3_task_id={task.task_id}" datajob.url = f"{base_url}/taskinstance/list/?flt1_dag_id_equals={datajob.flow_urn.flow_id}&_flt_3_task_id={task.task_id}"
if capture_owner and dag.owner: if capture_owner and dag.owner:
if config and config.capture_ownership_as_group: if config and config.capture_ownership_info:
datajob.group_owners.add(dag.owner) owners = AirflowGenerator._extract_owners(dag)
if config.capture_ownership_as_group:
datajob.group_owners.update(owners)
else: else:
datajob.owners.add(dag.owner) datajob.owners.update(owners)
if capture_tags and dag.tags: if capture_tags and dag.tags:
datajob.tags.update(dag.tags) datajob.tags.update(dag.tags)