数据融合中的管道依赖关系

问题描述

我在数据融合中有3个管道，分别是A，B和C。我希望在管道A和B都完成后执行以触发管道C。管道触发器仅将依赖关系放在一个管道上。可以在数据融合中实现吗？

解决方法

您可以使用Google Cloud Composer [1]完成此操作。为了执行此操作，首先需要在Google Cloud Composer [2]中创建一个新的环境，完成后，您需要在环境[3]中安装一个新的Python程序包，并且需要该程序包安装是[4]“ apache-airflow-backport-providers-google”。

安装此软件包后，您将可以使用这些操作[5]，您需要执行的操作是[6]“启动DataFusion管道”，这样您就可以从Airflow启动新的管道。 / p>

python代码的示例如下：

import airflow
import datetime
from airflow import DAG
from airflow import models
from airflow.operators.bash_operator import BashOperator
from datetime import timedelta
from airflow.providers.google.cloud.operators.datafusion import (
    CloudDataFusionStartPipelineOperator
)

default_args = {
   'start_date': airflow.utils.dates.days_ago(0),'retries': 1,'retry_delay': timedelta(minutes=5)
}

with models.DAG(
    'composer_DF',schedule_interval=datetime.timedelta(days=1),default_args=default_args) as dag:

    # the operations.
    A = CloudDataFusionStartPipelineOperator(
            location="us-west1",pipeline_name="A",instance_name="instance_name",task_id="start_pipelineA",)
    B = CloudDataFusionStartPipelineOperator(
            location="us-west1",pipeline_name="B",task_id="start_pipelineB",)
    C = CloudDataFusionStartPipelineOperator(
            location="us-west1",pipeline_name="C",task_id="start_pipelineC",)
    # First A then B and then C
    A >> B >> C

您可以通过查看Airflow文档来设置时间间隔。

将这些代码另存为.py文件后，请将其保存到您环境的Google Cloud Storage DAG文件夹中。

DAG启动时，它将执行任务A，完成时将执行任务B，依此类推。

[1] https://cloud.google.com/composer

[2] https://cloud.google.com/composer/docs/how-to/managing/creating#:~:text=In%20the%20Cloud%20Console%2C%20open%20the%20Create%20Environment%20page.&text=Under%20Node%20configuration%2C%20click%20Add%20environment%20variable.&text=The%20From%3A%20email%20address%2C%20such,%40%20.&text=Your%20SendGrid%20API%20key。

[3] https://cloud.google.com/composer/docs/how-to/using/installing-python-dependencies

[4] https://pypi.org/project/apache-airflow-backport-providers-google/

[5] https://airflow.readthedocs.io/en/latest/_api/airflow/providers/google/cloud/operators/datafusion/index.html

[6] https://airflow.readthedocs.io/en/latest/howto/operator/google/cloud/datafusion.html#start-a-datafusion-pipeline

我没有想到的直接方法，只有两个解决方法

解决1 。将管道A和B合并到管道AB中，然后触发管道C（AB> C）。

管道A-（GCS复制>解压缩），管道B-（GCS2> thrashsad）

BigQueryExecute以减轻错误：无效的DAG。有一个由阶段组成的岛屿。

在BigQueryExecute中，有效和伪查询。

将两个管道合并为一个，可能会扰乱管道测试。为了克服这个问题，您可以添加一个虚拟条件来一次运行管道。

在BigQueryExecute中，将查询更改为“ Select $ {flag}”，并在运行时参数中传递flag的值，或者选择1作为标志，然后将“ Row As Arguments”打钩为true。
在BigQueryExecute之后添加条件插件，并将条件运行时['flag'] = 1
条件插件有两个插座，将它们连接到管道A和管道B。

解决方法2 ：将两个管道（A和B）的标志存储在BiqQuery表中，创建两个流A> C和B> C来触发管道C。这将两次触发管道C但是只有在BigQuery表中两个标志均可用时，才能使用BigQueryExecute和condition插件。

如何？

在管道A和B中将输出（一行）写入BigQuery表“ Pipeline_Run”
在管道C中，添加BigQueryExecute并从ds.Pipeline_Run查询'select count（*）作为Cnt，然后将“行作为参数”勾选为true。
在管道C中，添加Condition插件，并检查cnt的值是否为2（运行时['cnt'] = 2），并将其余管道的插件连接到其“是”插座。

您可以探索通过CDAP REST API设置的“时间表”。这样就可以并行执行管道，并且不依赖云编写器（工作流中第一个管道的基于文件的触发器除外。为此，您将需要云功能或可能是云编写器文件传感器）

cdap google-cloud-data-fusion