google-cloud-platform - 如何从谷歌云功能调用谷歌数据处理工作

Question

每当有新文件上传到云存储桶时触发云函数。此函数应调用用 pyspark 编写的数据处理作业来读取文件并将其加载到 BigQuery。

我想知道如何从云函数调用 google dataproc 作业。请提出建议。

Answer 1

我能够创建一个简单的 Cloud Function，它在 GCS 创建文件事件上触发 Dataproc 作业。在此示例中，GCS 中的文件包含要执行的 Pig 查询。不过你可以关注Dataproc API创建 PySpark 版本的文档。

index.js:

exports.submitJob = (event, callback) => {

  const google = require('googleapis');

  const projectId = 'my-project'
  const clusterName = 'my-cluster'

  const file = event.data;
  if (file.name) {
    google.auth.getApplicationDefault(function (err, authClient, projectId) {
      if (err) {
        throw err;
      }

      const queryFileUri = "gs://" + file.bucket + "/" + file.name
      console.log("Using queryFileUri: ", queryFileUri);

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

      const dataproc = google.dataproc({ version: 'v1beta2', auth: authClient });

      dataproc.projects.regions.jobs.submit({
          projectId: projectId,
          region: "global",
          resource: {
            "job": {
              "placement": {"clusterName": clusterName},
              "pigJob": {
                "queryFileUri": queryFileUri,
              }
            }
          }
        }, function(err, response) {
          if (err) {
            console.error("Error submitting job: ", err);
          }
          console.log("Dataproc response: ", response);
          callback();
        });

    });
  } else {
    throw "Skipped processing file!";
  }

  callback();
};

确保将 Function to execute 设置为 submitJob。

package.json:

{
  "name": "sample-cloud-storage",
  "version": "0.0.1",
  "dependencies":{ "googleapis": "^21.3.0" }
}

下面的博文给了我很多如何开始的想法: https://cloud.google.com/blog/big-data/2016/04/scheduling-dataflow-pipelines-using-app-engine-cron-service-or-cloud-functions