如何使用气流操作符触发 CDAP 管道？答案

【问题标题】：How to trigger a CDAP pipeline using airflow operators?如何使用气流操作符触发 CDAP 管道？
【发布时间】：2022-01-05 22:17:09
【问题描述】：

我有一个具有多个命名空间的本地 CDAP 数据融合实例。如何使用气流操作符触发管道？我已尝试探索可用的气流操作符和此页面，但不是很有帮助https://cloud.google.com/data-fusion/docs/reference/cdap-reference#start_a_batch_pipeline

【问题讨论】：

标签： google-cloud-platform airflow google-cloud-data-fusion cdap

【解决方案1】：

假设您已经部署了管道并且您拥有要运行的管道的位置、实例名称和管道名称。有关它接受的参数，请参阅CloudDataFusionStartPipelineOperator()。

使用quickstart pipeline，我使用CloudDataFusionStartPipelineOperator() 触发了管道。请参阅下面的运算符用法：

import airflow
from airflow.providers.google.cloud.operators.datafusion import CloudDataFusionStartPipelineOperator

YESTERDAY = datetime.datetime.now() - datetime.timedelta(days=1)

default_args = {
    'owner': 'Composer Example',
    'depends_on_past': False,
    'email': [''],
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': datetime.timedelta(minutes=5),
    'start_date': YESTERDAY,
}

with airflow.DAG(
        'trigger_df',
        'catchup=False',
        default_args=default_args,
        schedule_interval=datetime.timedelta(days=1)) as dag:

    start_pipeline = CloudDataFusionStartPipelineOperator(
        location='us-central1',
        pipeline_name='DataFusionQuickstart',
        instance_name='test',
        task_id="start_pipeline",
    )

    start_pipeline

成功的“图表视图”：

日志：

【讨论】：

云数据融合的好例子。谢谢。但我正在寻找数据融合的本地实例 CDAP。一个类似的例子会很有帮助。当我说前提条件时，连接需要主机、端口和用户身份验证