触发 DataPrep 数据流作业的云函数

Question

我有一个小管道正在尝试执行：

1. 文件放入 GCS Bucket > 2. 当文件放入 GCS Bucket 时，Cloud Function 触发 Dataflow 作业（不工作）> 3. 写入 Big Query 表（这部分工作）

我通过 Dataprep 创建了一个 Dataflow 作业，因为它有很好的 UI 可以在写入 BigQuery 表之前完成我的所有转换（写入 BigQuery 工作正常），并且当文件上传到 GCS 存储桶时会触发 Cloud 函数.但是，Cloud Function 不会触发 Dataflow 作业（我在 Dataprep 中编写）。

请看一下我的云函数下面的示例代码，如果我能得到任何关于为什么 Dataflow 作业没有触发的指示。

/**
 * Triggered from a message on a Cloud Storage bucket.
 *
 * @param {!Object} event The Cloud Functions event.
 * @param {!Function} The callback function.
 */
exports.processFile = (event, callback) => {
  console.log('Processing file: ' + event.data.name);
  callback();

  const google = require('googleapis');

 exports.CF_GCStoDataFlow_v2 = function(event, callback) {
  const file = event.data;
  if (file.resourceState === 'exists' && file.name) {
    google.auth.getApplicationDefault(function (err, authClient, projectId) {
      if (err) {
        throw err;
      }

      if (authClient.createScopedRequired && authClient.createScopedRequired()) {
        authClient = authClient.createScoped([
          'https://www.googleapis.com/auth/cloud-platform',
          'https://www.googleapis.com/auth/userinfo.email'
        ]);
      }

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

      dataflow.projects.templates.create({
        projectId: projectId,
        resource: {
          parameters: {
            inputFile: `gs://${file.bucket}/${file.name}`,
            outputFile: `gs://${file.bucket}/${file.name}`
          },
          jobName: 'cloud-dataprep-csvtobq-v2-281345',
          gcsPath: 'gs://mygcstest-pipeline-staging/temp/'
        }
      }, function(err, response) {
        if (err) {
          console.error("problem running dataflow template, error was: ", err);
        }
        console.log("Dataflow template response: ", response);
        callback();
      });

    });
  }
 };
};

Answer 1

这个 sn-p 可能会有所帮助，它使用数据流 api（启动）的不同方法，它对我有用，请注意您需要指定模板的 url 并检查元数据文件（您可以在同一个文件中找到它通过dataprep接口执行时作为模板的目录）文件包含正确的参数

dataflow.projects.templates.launch({
   projectId: projectId,
   location: location,
   gcsPath: jobTemplateUrl,
   resource: {
     parameters: {
       inputLocations : `{"location1" :"gs://${file.bucket}/${file.name}"}`,
       outputLocations: `{"location1" : "gs://${destination.bucket}/${destination.name}"}"}`,
     },
      environment: {
        tempLocation: `gs://${destination.bucket}/${destination.tempFolder}`,
        zone: "us-central1-f"
     },
     jobName: 'my-job-name',

   }
 }

Answer 2

您是否提交了 Dataproc 作业？它开始运行了吗？ 下面的文档可以提供一些入门的想法！

https://cloud.google.com/dataproc/docs/concepts/jobs/life-of-a-job

Answer 3

看起来您将CF_GCStoDataFlow_v2 放入processFile，因此代码的数据流部分没有执行。

你的函数应该是这样的：

/**
 * Triggered from a message on a Cloud Storage bucket.
 *
 * @param {!Object} event The Cloud Functions event.
 * @param {!Function} The callback function.
 */
exports.CF_GCStoDataFlow_v2 = (event, callback) => {

  const google = require('googleapis');

  if (file.resourceState === 'exists' && file.name) {
    google.auth.getApplicationDefault(function (err, authClient, projectId) {
      if (err) {
        throw err;
      }

      if (authClient.createScopedRequired && authClient.createScopedRequired()) {
        authClient = authClient.createScoped([
          'https://www.googleapis.com/auth/cloud-platform',
          'https://www.googleapis.com/auth/userinfo.email'
        ]);
      }

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

      dataflow.projects.templates.create({
        projectId: projectId,
        resource: {
          parameters: {
            inputFile: `gs://${file.bucket}/${file.name}`,
            outputFile: `gs://${file.bucket}/${file.name}`
          },
          jobName: '<JOB_NAME>',
          gcsPath: '<BUCKET_NAME>'
        }
      }, function(err, response) {
        if (err) {
          console.error("problem running dataflow template, error was: ", err);
        }
        console.log("Dataflow template response: ", response);
        callback();
      });

    });
  }

  callback();
};

确保将“Function to execute”下的值更改为 CF_GCStoDataFlow_v2