yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-09-26 09:22:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

Airflow exception management (#8410)

Browse Source 
* Airflow exception management

* Airflow exception management
This commit is contained in:
Pere Miquel Brull 2022-10-28 16:11:23 +02:00 committed by GitHub
parent f027c26cc1
commit c8f4d496a5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 29 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -18,7 +18,7 @@ from typing import Any, Iterable, List, Optional, cast
from airflow.models import BaseOperator, DagRun, TaskInstance from airflow.models import BaseOperator, DagRun, TaskInstance
from airflow.models.serialized_dag import SerializedDagModel from airflow.models.serialized_dag import SerializedDagModel
from airflow.serialization.serialized_objects import SerializedDAG from airflow.serialization.serialized_objects import SerializedDAG
from pydantic import BaseModel from pydantic import BaseModel, ValidationError
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from metadata.generated.schema.api.data.createPipeline import CreatePipelineRequest from metadata.generated.schema.api.data.createPipeline import CreatePipelineRequest
@ -320,19 +320,34 @@ class AirflowSource(PipelineServiceSource):
:return: Create Pipeline request with tasks :return: Create Pipeline request with tasks
""" """
dag: SerializedDAG = self._build_dag(pipeline_details.data) try:
yield CreatePipelineRequest( dag: SerializedDAG = self._build_dag(pipeline_details.data)
name=pipeline_details.dag_id, yield CreatePipelineRequest(
description=dag.description, name=pipeline_details.dag_id,
pipelineUrl=f"/tree?dag_id={dag.dag_id}", # Just the suffix description=dag.description,
concurrency=dag.concurrency, pipelineUrl=f"/tree?dag_id={dag.dag_id}", # Just the suffix
pipelineLocation=pipeline_details.fileloc, concurrency=dag.concurrency,
startDate=dag.start_date.isoformat() if dag.start_date else None, pipelineLocation=pipeline_details.fileloc,
tasks=self.get_tasks_from_dag(dag), startDate=dag.start_date.isoformat() if dag.start_date else None,
service=EntityReference( tasks=self.get_tasks_from_dag(dag),
id=self.context.pipeline_service.id.__root__, type="pipelineService" service=EntityReference(
), id=self.context.pipeline_service.id.__root__, type="pipelineService"
) ),
)
except TypeError as err:
logger.debug(traceback.format_exc())
logger.warning(
f"Error building DAG information from {pipeline_details}. There might be Airflow version"
f" incompatibilities - {err}"
)
except ValidationError as err:
logger.debug(traceback.format_exc())
logger.warning(
f"Error building pydantic model for {pipeline_details} - {err}"
)
except Exception as err:
logger.debug(traceback.format_exc())
logger.warning(f"Wild error ingesting pipeline {pipeline_details} - {err}")
@staticmethod @staticmethod
def parse_xlets(xlet: List[Any]) -> Optional[List[str]]: def parse_xlets(xlet: List[Any]) -> Optional[List[str]]: