如何运行创建 Dataproc 集群、运行作业、从 Cloud Function 删除集群

我想启动 Dataproc 作业来响应到达 GCS 存储桶中的日志文件。我也不想让持久集群保持运行，因为新的日志文件每天只到达几次，并且大多数时间都是空闲的。

此处提供了本教程的较新版本: https://cloud.google.com/dataproc/docs/tutorials/workflow-function

我可以用工作流程模板 API为我管理集群生命周期。使用 Dataproc 工作流程，我无需轮询要创建的集群、创建的作业，或执行任何错误处理。

这是我的云功能。设置Cloud Storage bucket触发Finalize/Create event:

索引.js：

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

  const {
    google
  } = require('googleapis');

  const region = 'global'
  const zone = 'us-central1-a'
  const clusterName = 'my-cluster'

  const file = event.data;
  console.log("Event: ", file);

  if (!file.name) {
    throw "Skipped processing file!";
  }

  const queryFileUri = "gs://" + file.bucket + "/" + file.name

  console.log("Creating auth client: ");
  google.auth.getApplicationDefault(
    (err, authClient, projectId) => {
      if (authClient.createScopedRequired && authClient.createScopedRequired()) {
        authClient = authClient.createScoped([
          'https://www.googleapis.com/auth/cloud-platform',
          'https://www.googleapis.com/auth/userinfo.email'
        ]);
      }

      const request = {
        parent: "projects/" + projectId + "/regions/" + region,
        resource: {
          "placement": {
            "managedCluster": {
              "clusterName": clusterName,
              "config": {
                "gceClusterConfig": {
                  "zoneUri": zone, // Can be omitted if using regional endpoint (like us-central1-a, not global)
                }
              }
            }
          },
          "jobs": [{
            "stepId": "step1",
            "pigJob": {
              "queryFileUri": queryFileUri,
            },
            "prerequisiteStepIds": [],
          }]
        }
      };

      const dataproc = google.dataproc({
        version: 'v1beta2',
        auth: authClient
      });
      dataproc.projects.regions.workflowTemplates.instantiateInline(
        request, (err, result) => {
          if (err) {
            throw err;
          }
          console.log(result);
          callback();
        });
    });
};

确保将要执行的功能设置为startWorkflow.

包.json：

{
  "name": "dataproc-workflow",
  "version": "1.0.0",
  "dependencies":{ "googleapis": "30.0.0"}
}