google-cloud-platform - 从 Cloud Functions 启动 Cloud Dataflow

Question

如何从 Google Cloud Function 启动 Cloud Dataflow 作业?我想使用 Google Cloud Functions 作为启用跨服务组合的机制。

Answer 1

我已经包含了 WordCount 的一个非常基本的示例。下面的示例。请注意，您需要在 Cloud Function 部署中包含 java 二进制文件的副本，因为它不在默认环境中。同样，您还需要将部署 jar 与 Cloud Function 一起打包。

module.exports = {
  wordcount: function (context, data) {
    const spawn = require('child_process').spawn;
    const child = spawn(
            'jre1.8.0_73/bin/java',
            ['-cp',
             'MY_JAR.jar',
             'com.google.cloud.dataflow.examples.WordCount',
             '--jobName=fromACloudFunction',
             '--project=MY_PROJECT',
             '--runner=BlockingDataflowPipelineRunner',
             '--stagingLocation=gs://STAGING_LOCATION',
             '--inputFile=gs://dataflow-samples/shakespeare/*',
             '--output=gs://OUTPUT_LOCATION'
            ],
            { cwd: __dirname });

    child.stdout.on('data', function(data) {
      console.log('stdout: ' + data);
    });
    child.stderr.on('data', function(data) {
      console.log('error: ' + data);
    });
    child.on('close', function(code) {
      console.log('closing code: ' + code);
    });
    context.success();
  }
}

您可以使用 non-blocking runner 进一步增强此示例。并让函数返回作业 ID，以便您可以单独轮询作业完成情况。这种模式也应该对其他 SDK 有效，只要它们的依赖项可以打包到 Cloud Function 中。