使用 pubsub 推送触发器运行云功能

Posted 2023-02-16

【中文标题】使用 pubsub 推送触发器运行云功能

【英文标题】：Running a cloud function with a pubsub push trigger

【发布时间】：2020-08-30 20:24:57

【问题描述】：

我设置了一个 Python 脚本，该脚本将从一个数据集中获取某些 bigquery 表，使用 SQL 查询清理它们，并将清理后的表添加到新数据集中。该脚本工作正常。我想将其设置为每天午夜触发的云功能。

我还使用云调度程序在每天午夜向 pubsub 主题发送消息。我已经验证这可以正常工作。我是 pubsub 新手，但我按照文档中的教程进行操作，并设法设置了一个测试云功能，当它从 pubsub 收到推送通知时打印出 hello world。

但是，我的问题是，当我尝试将两者结合并自动化我的脚本时 - 我收到一条日志消息，表明执行崩溃：

Function execution took 1119 ms, finished with status: 'crash'

为了帮助你理解我在做什么，这是我的 main.py 中的代码：

# Global libraries
import base64

# Local libraries
from scripts.one_minute_tables import helper

def one_minute_tables(event, context):

    # Log out the message that triggered the function
    print("""This Function was triggered by messageId  published at 
    """.format(context.event_id, context.timestamp))

    # Get the message from the event data
    name = base64.b64decode(event['data']).decode('utf-8')

    # If it's the message for the daily midnight schedule, execute function
    if name == 'midnight':
        helper.format_tables('raw_data','table1')
    else:
        pass

为了方便，这是我的python脚本的简化版：

# Global libraries
from google.cloud import bigquery
import os

# Login to bigquery by providing credentials
credential_path = 'secret.json'
os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = credential_path

def format_tables(dataset, list_of_tables):

    # Initialize the client
    client = bigquery.Client()

    # Loop through the list of tables
    for table in list_of_tables:

        # Create the query object
        script = f"""
            SELECT *
            FROM dataset.table
        """

        # Call the API
        query = client.query(script)

        # Wait for job to finish
        results = query.result()

        # Print
        print('Data cleaned and updated in table: .'.format(dataset, table))

这是我的文件夹结构：

而我的requirements.txt 文件中只有一个条目：google-cloud-bigquery==1.24.0

非常感谢您帮助我弄清楚我需要修复什么才能使用 pubsub 触发器运行此脚本，而不会收到显示执行已崩溃的日志消息。

编辑：基于 cmets，这是函数崩溃的日志

  "textPayload": "Function execution took 1078 ms, finished with status: 'crash'",
  "insertId": "000000-689fdf20-aee2-4900-b5a1-91c34d7c1448",
  "resource": 
    "type": "cloud_function",
    "labels": 
      "function_name": "one_minute_tables",
      "region": "us-central1",
      "project_id": "PROJECT_ID"
    
  ,
  "timestamp": "2020-05-15T16:53:53.672758031Z",
  "severity": "DEBUG",
  "labels": 
    "execution_id": "x883cqs07f2w"
  ,
  "logName": "projects/PROJECT_ID/logs/cloudfunctions.googleapis.com%2Fcloud-functions",
  "trace": "projects/PROJECT_ID/traces/f391b48a469cbbaeccad5d04b4a704a0",
  "receiveTimestamp": "2020-05-15T16:53:53.871051291Z"

【问题讨论】：

看你的云函数的日志，回溯错误是什么？我假设您发布的第一个 python 脚本 (def one_minute_tables) 是由 pubsub 触发的，对吗？

你的函数的配置是什么？您是否创建了 trigger-http 函数和对 PubSub 的 http 推送订阅？或者你创建一个--trigger-topic？

一个想法是尝试捕获 Cloud Functions 有时会抑制的堆栈跟踪。使用the Approach 2 in this answer 作为指导

@guillaumeblaquiere 我已经设置了一个触发主题gcloud functions deploy one_minute_tables --runtime python37 --trigger-topic scheduled_updates

@MajorHonda 是的，函数one_minute_tables是由pubsub触发的。我查看了日志，我将编辑我的问题以添加函数崩溃的日志

【参考方案1】：

问题来自list_of_tables 属性。你这样调用你的函数

    if name == 'midnight':
        helper.format_tables('raw_data','table1')

然后你迭代你的 'table1' 参数

执行此操作，它应该可以工作

    if name == 'midnight':
        helper.format_tables('raw_data',['table1'])