如何使用云功能触发数据流？ (Python SDK)

Question

我有一个由云 Pub/Sub 触发的云功能。我想要使​​用 Python SDK 的相同函数触发数据流。这是我的代码：

import base64
def hello_pubsub(event, context):   
    if 'data' in event:
        message = base64.b64decode(event['data']).decode('utf-8')
    else:
        message = 'hello world!'
    print('Message of pubsub : {}'.format(message))

我是这样部署函数的：

gcloud beta functions deploy hello_pubsub  --runtime python37 --trigger-topic topic1

Answer 1

您必须将管道 python 代码嵌入到您的函数中。当你的函数被调用时，你只需调用管道 python 主函数，它会在你的文件中执行管道。

如果您在 Cloud Shell 中开发并试用了您的管道，并且您已经在 Dataflow 管道中运行了它，那么您的代码应该具有以下结构：

def run(argv=None, save_main_session=True):
  # Parse argument
  # Set options
  # Start Pipeline in p variable
  # Perform your transform in Pipeline
  # Run your Pipeline
  result = p.run()
  # Wait the end of the pipeline
  result.wait_until_finish()

因此，使用正确的参数（尤其是 runner=DataflowRunner）调用此函数，以允许 python 代码在 Dataflow 服务中加载管道。

删除最后的result.wait_until_finish()，因为您的函数不会长期存在所有数据流过程。

如果你愿意，你也可以使用模板。

Answer 2

您可以使用Cloud Dataflow templates 启动您的工作。您将需要编写以下步骤：

• 检索凭据
• 生成数据流服务实例
• 获取 GCP PROJECT_ID
• 生成模板正文
• 执行模板

这是一个使用您的基本代码的示例（请随意拆分为多个方法以减少 hello_pubsub 方法中的代码）。

from googleapiclient.discovery import build
import base64
import google.auth
import os

def hello_pubsub(event, context):   
    if 'data' in event:
        message = base64.b64decode(event['data']).decode('utf-8')
    else:
        message = 'hello world!'

    credentials, _ = google.auth.default()
    service = build('dataflow', 'v1b3', credentials=credentials)
    gcp_project = os.environ["GCLOUD_PROJECT"]

    template_path = gs://template_file_path_on_storage/
    template_body = {
        "parameters": {
            "keyA": "valueA",
            "keyB": "valueB",
        },
        "environment": {
            "envVariable": "value"
        }
    }

    request = service.projects().templates().launch(projectId=gcp_project, gcsPath=template_path, body=template_body)
    response = request.execute()

    print(response)

在 template_body 变量中，参数值是将发送到您的管道的参数，环境值由 Dataflow 服务（服务帐户、工作人员和网络配置）使用。

LaunchTemplateParameters documentation

RuntimeEnvironment documentation