2

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

4

3 回答 3

6

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

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

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", 
            instance_name="instance_name", task_id="start_pipelineB",
        )
    C = CloudDataFusionStartPipelineOperator(
            location="us-west1", pipeline_name="C", 
            instance_name="instance_name", 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=在%20Node%20configuration%2C%20click%20Add%20environment%20variable.&text=%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

于 2020-08-28T10:56:42.197 回答
1

我想不出直接的方法，但有两种解决方法

围绕 1 工作。将流水线 A 和 B 合并到流水线​​ AB 中，然后触发流水线 C (AB > C)。

管道 A - (GCS Copy > Decompress), 管道 B - (GCS2 > thrashsad)

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

在此处输入图像描述

在 BigQueryExecute 中，有效查询和虚拟查询。

在此处输入图像描述

将两条流水线合二为一，可能会使流水线测试感到不安。为了克服这个问题，您可以添加一个虚拟条件来运行管道一次。

  1. 在 BigQueryExecute 中，将查询更改为“选择 ${flag}”并在运行时参数中传递标志的值或选择 1 作为标志并将“作为参数的行”勾选为真。
  2. 在 BigQueryExecute 之后添加条件插件并将条件 runtime['flag'] = 1
  3. Condition 插件有两个出口，分别连接到管道 A 和管道 B。

在此处输入图像描述

解决方法 2：将两个管道（A 和 B）的标志存储在 BiqQuery 表中，创建两个流 A>C 和 B>C 以触发管道 C。这将触发管道 C 两次，但使用 BigQueryExecute 和条件插件将仅在以下情况下运行这两个标志在 BigQuery 表中都可用。

如何？

  1. 在管道 A 和 B 中将输出（一行）写入 BigQuery 表“Pipeline_Run”
  2. 在管道 C 中，添加 BigQueryExecute 并查询 'select count(*) as Cnt from ds.Pipeline_Run' 并将“Row As Arguments”勾选为 true。
  3. 在管道 C 中，添加 Condition 插件并检查 cnt 的值是否为 2 (runtime['cnt'] = 2) 并将管道的其余插件连接到其“是”出口。
于 2020-09-08T16:35:41.303 回答
0

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

于 2020-09-02T11:20:22.750 回答