安排Google Cloud Dataflow作业的最简便方法

Question

我只需要每天运行一个数据流管道,但在我看来,建议像App Engine Cron Service这样需要构建整个Web应用程序的解决方案似乎有点太多了.我正在考虑从Compute Engine Linux VM中的cron作业运行管道,但这可能太简单了:).这样做有什么问题,为什么不是任何人(除了我,我猜)建议它？

Answer 1

这就是我使用 Cloud Functions、PubSub 和 Cloud Scheduler 的方式（假设您已经创建了一个 Dataflow 模板并且它存在于您的 GCS 存储桶中的某处）

1. 在 PubSub 中创建一个新主题。这将用于触发云功能

2. 创建从模板启动 Dataflow 作业的 Cloud Functions。我发现从 CF 控制台创建它是最简单的。确保您选择的服务帐号有权创建数据流作业。该函数的 index.js 看起来像：

const google = require('googleapis');

exports.triggerTemplate = (event, context) => {
  // in this case the PubSub message payload and attributes are not used
  // but can be used to pass parameters needed by the Dataflow template
  const pubsubMessage = event.data;
  console.log(Buffer.from(pubsubMessage, 'base64').toString());
  console.log(event.attributes);

  google.google.auth.getApplicationDefault(function (err, authClient, projectId) {
  if (err) {
    console.error('Error occurred: ' + err.toString());
    throw new Error(err);
  }

  const dataflow = google.google.dataflow({ version: 'v1b3', auth: authClient });

  dataflow.projects.templates.create({
        projectId: projectId,
        resource: {
          parameters: {},
          jobName: 'SOME-DATAFLOW-JOB-NAME',
          gcsPath: 'gs://PATH-TO-YOUR-TEMPLATE'
        }
      }, function(err, response) {
        if (err) {
          console.error("Problem running dataflow template, error was: ", err);
        }
        console.log("Dataflow template response: ", response);
      });
  });
};

package.json 看起来像

{
  "name": "pubsub-trigger-template",
  "version": "0.0.1",
  "dependencies": {
    "googleapis": "37.1.0",
    "@google-cloud/pubsub": "^0.18.0"
  }
}

3. 转到 PubSub 和您创建的主题，手动发布消息。这应该会触发 Cloud Function 并启动 Dataflow 作业

4. 使用 Cloud Scheduler 按计划发布 PubSub 消息 https://cloud.google.com/scheduler/docs/tut-pub-sub

Answer 2

使用 cron 作业启动 Dataflow 管道绝对没有错。我们一直在为我们的生产系统做这件事，无论是我们的 Java 还是 Python 开发的管道。

尽管如此，我们正试图摆脱 cron 作业，更多地使用 AWS Lambdas（我们运行多云）或 Cloud Functions。不幸的是，Cloud Functions还没有调度。AWS Lambda可以。

Answer 3

该问题有一个常见问题解答：https : //cloud.google.com/dataflow/docs/resources/faq#is_there_a_built-in_scheduling_mechanism_to_execute_pipelines_at_given_time_or_interval

• 您可以使用Google App Engine（仅适用于Flexible Environment）或Cloud Functions自动执行管道。
• 您可以使用Apache Airflow的Dataflow Operator，它是Cloud Composer工作流程中的几种Google Cloud Platform Operators之一。
• 您可以在Compute Engine上使用自定义（cron）作业流程。

Cloud Function方法被描述为“ Alpha”，仍然没有调度（不等同于AWS cloudwatch调度事件），只有发布/订阅消息，云存储更改，HTTP调用仍然是事实。

云作曲家看起来是一个不错的选择。有效地重新标记了Apache Airflow，它本身就是一个很棒的编排工具。绝对不是像cron那样“太简单” :)