yujunjun/datahub
2
0
Fork 0
mirror of https://github.com/datahub-project/datahub.git synced 2025-09-26 09:35:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ingestion): airflow - import emitters indirectly to avoid unneeded dependency (#4668)

Browse Source
This commit is contained in:
Tamas Nemeth 2022-04-14 19:22:16 +02:00 committed by GitHub
parent 8f6718104b
commit 61dc6e8723
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
metadata-ingestion/src/datahub_provider/client/airflow_generator.py
View File

@ -7,8 +7,6 @@ from datahub.api.entities.dataprocess.dataprocess_instance import (
DataProcessInstance, DataProcessInstance,
InstanceRunResult, InstanceRunResult,
) )
from datahub.emitter.kafka_emitter import DatahubKafkaEmitter
from datahub.emitter.rest_emitter import DatahubRestEmitter
from datahub.metadata.schema_classes import DataProcessTypeClass from datahub.metadata.schema_classes import DataProcessTypeClass
from datahub.utilities.urns.data_flow_urn import DataFlowUrn from datahub.utilities.urns.data_flow_urn import DataFlowUrn
from datahub.utilities.urns.data_job_urn import DataJobUrn from datahub.utilities.urns.data_job_urn import DataJobUrn
@ -18,6 +16,9 @@ if TYPE_CHECKING:
from airflow import DAG from airflow import DAG
from airflow.models import BaseOperator, DagRun, TaskInstance from airflow.models import BaseOperator, DagRun, TaskInstance
from datahub.emitter.kafka_emitter import DatahubKafkaEmitter
from datahub.emitter.rest_emitter import DatahubRestEmitter
class AirflowGenerator: class AirflowGenerator:
@staticmethod @staticmethod
@ -253,7 +254,7 @@ class AirflowGenerator:
@staticmethod @staticmethod
def run_dataflow( def run_dataflow(
emitter: DatahubRestEmitter, emitter: Union["DatahubRestEmitter", "DatahubKafkaEmitter"],
cluster: str, cluster: str,
dag_run: "DagRun", dag_run: "DagRun",
start_timestamp_millis: Optional[int] = None, start_timestamp_millis: Optional[int] = None,
@ -300,7 +301,7 @@ class AirflowGenerator:
@staticmethod @staticmethod
def complete_dataflow( def complete_dataflow(
emitter: DatahubRestEmitter, emitter: Union["DatahubRestEmitter", "DatahubKafkaEmitter"],
cluster: str, cluster: str,
dag_run: "DagRun", dag_run: "DagRun",
end_timestamp_millis: Optional[int] = None, end_timestamp_millis: Optional[int] = None,
@ -345,7 +346,7 @@ class AirflowGenerator:
@staticmethod @staticmethod
def run_datajob( def run_datajob(
emitter: Union[DatahubRestEmitter, DatahubKafkaEmitter], emitter: Union["DatahubRestEmitter", "DatahubKafkaEmitter"],
cluster: str, cluster: str,
ti: "TaskInstance", ti: "TaskInstance",
dag: "DAG", dag: "DAG",
@ -416,7 +417,7 @@ class AirflowGenerator:
@staticmethod @staticmethod
def complete_datajob( def complete_datajob(
emitter: Union[DatahubRestEmitter, DatahubKafkaEmitter], emitter: Union["DatahubRestEmitter", "DatahubKafkaEmitter"],
cluster: str, cluster: str,
ti: "TaskInstance", ti: "TaskInstance",
dag: "DAG", dag: "DAG",