yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-08-31 12:39:01 +00:00
Code Issues Packages Projects Releases Wiki Activity

FIX #18309: fixed task deserialization in Airflow metadata ingestion (#18310)

Browse Source 
* fixed task deserialization in Airflow metadata ingestion

* fixed formatting

---------

Co-authored-by: Katarzyna Kałek <kkalek@olx.pl>
This commit is contained in:
Katarzyna Kałek 2024-10-17 23:51:55 +02:00 committed by GitHub
parent 47c993be92
commit c9995eecb6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 14 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
ingestion/src/metadata/ingestion/source/pipeline/airflow/metadata.py
View File

@ -15,7 +15,7 @@ import traceback
from collections import Counter from collections import Counter
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from typing import Iterable, List, Optional, cast from typing import Dict, Iterable, List, Optional, cast
from airflow.models import BaseOperator, DagRun, TaskInstance from airflow.models import BaseOperator, DagRun, TaskInstance
from airflow.models.dag import DagModel from airflow.models.dag import DagModel
@ -142,6 +142,16 @@ class AirflowSource(PipelineServiceSource):
return self._session return self._session
@staticmethod
def _extract_serialized_task(task: Dict) -> Dict:
"""
Given the serialization changes introduced in Airflow 2.10,
ensure compatibility with all versions.
"""
if task.keys() == {"__var", "__type"}:
return task["__var"]
return task
def get_pipeline_status(self, dag_id: str) -> List[DagRun]: def get_pipeline_status(self, dag_id: str) -> List[DagRun]:
""" """
Return the DagRuns of given dag Return the DagRuns of given dag
@ -328,7 +338,9 @@ class AirflowSource(PipelineServiceSource):
max_active_runs=data.get("max_active_runs", None), max_active_runs=data.get("max_active_runs", None),
description=data.get("_description", None), description=data.get("_description", None),
start_date=data.get("start_date", None), start_date=data.get("start_date", None),
tasks=data.get("tasks", []), tasks=list(
map(self._extract_serialized_task, data.get("tasks", []))
),
schedule_interval=get_schedule_interval(data), schedule_interval=get_schedule_interval(data),
owner=self.fetch_dag_owners(data), owner=self.fetch_dag_owners(data),
) )