从 Cloud Function 触发 Cloud Dataflow 管道 - 函数超时 [英] Triggering Cloud Dataflow pipeline from Cloud Function - function times out

问题描述

我正在尝试从 Cloud Function 触发 Dataflow 管道，该管道本身是在 GCS 存储桶中上传新文件时触发的.当我上传文件时，云功能被正确触发，但几秒钟后超时，没有触发任何数据流.下面是我的功能代码:

I am trying to trigger a Dataflow pipeline from a Cloud Function which itself is triggered upon upload of a new file in a GCS bucket. When I upload a file, the Cloud function gets triggered properly but timesout after few seconds without any Dataflow being triggered. Below is my function code:

const google = require('googleapis');
const projectId = "iot-fitness-198120";
 
exports.moveDataFromGCStoPubSub = 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'
       ]);
     }
     console.log("File exists and client function is authenticated");
     console.log(file);
     
 
     const dataflow = google.dataflow({ version: 'v1b3', auth: authClient });
     console.log(`Incoming data: ${file.name}`);
 
     dataflow.projects.templates.create({
       projectId: projectId,
       resource: {
         parameters: {
           inputFile: `gs://${file.bucket}/${file.name}`,
           outputTopic: `projects/iot-fitness-198120/topics/MemberFitnessData`
         },
         jobName: 'CStoPubSub',
         gcsPath: 'gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub',
         staginglocation: 'gs://fitnessanalytics-tmp/tmp'
       }
     }, function(err, response) {
       if (err) {
         console.error("problem running dataflow template, error was: ", err);
       }
       console.log("Dataflow template response: ", response);
       callback();
     });
 
   });
 }
};

执行甚至没有记录以下行，console.log("File exists and client function is authenticated");这告诉我它甚至没有走那么远.

The execution doesn't even log the following line, console.log("File exists and client function is authenticated"); which tells me it is not even getting that far.

这是执行期间的日志输出:

Here's the log output during execution:

2018-03-20 04:56:43.283 GST数据流触发函数52957909906492函数执行耗时 60097 毫秒，状态为:'timeout'

2018-03-20 04:56:43.283 GST DataflowTriggeringFunction 52957909906492 Function execution took 60097 ms, finished with status: 'timeout'

2018-03-20 04:55:43.188 GST数据流触发函数52957909906492函数执行开始

2018-03-20 04:55:43.188 GST DataflowTriggeringFunction 52957909906492 Function execution started

知道为什么它没有触发数据流但没有抛出错误消息吗?

Any idea why it's not triggering the Dataflow and yet not throwing an error message ?

推荐答案

我终于修改了代码.从 GCP 支持获得了一些帮助.下面是正确的语法:

I have finally modified the code. Got some help from GCP support. Below is the right syntax that works:

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

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


const file = event.data;
const context = event.context;

console.log(`Event ${context.eventId}`);
console.log(`  Event Type: ${context.eventType}`);
console.log(`  Bucket: ${file.bucket}`);
console.log(`  File: ${file.name}`);
console.log(`  Metageneration: ${file.metageneration}`);
console.log(`  Created: ${file.timeCreated}`);
console.log(`  Updated: ${file.updated}`);

  google.auth.getApplicationDefault(function (err, authClient, projectId) {
     if (err) {
       throw err;
     }

 console.log(projectId);

 const dataflow = google.dataflow({ version: 'v1b3', auth: authClient });
 console.log(`gs://${file.bucket}/${file.name}`);
 dataflow.projects.templates.create({
  gcsPath: 'gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub', 
  projectId: projectId,
   resource: {
    parameters: {
        inputFilePattern: `gs://${file.bucket}/${file.name}`,
        outputTopic: 'projects/iot-fitness-198120/topics/MemberFitnessData2'
      },
    environment: {
      tempLocation: 'gs://fitnessanalytics-tmp/tmp'
    },
      jobName: 'CStoPubSub',
      //gcsPath: 'gs://dataflow-templates/latest/GCS_Text_to_Cloud_PubSub',    
    }
 }, function(err, response) {
   if (err) {
     console.error("problem running dataflow template, error was: ", err);
   }
   console.log("Dataflow template response: ", response);
   callback();
 });

   });

 callback();
};

这篇关于从 Cloud Function 触发 Cloud Dataflow 管道 - 函数超时的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！