函数计算支持DataHub作为触发源。当有请求到达Connector设置为函数计算的阿里云流式数据服务DataHub时,DataHub会触发函数的执行。函数计算会将执行结果返回给DataHub。本文介绍如何配置DataHub触发函数执行,包括配置函数的入口参数、编写函数代码并测试等。

前提条件

使用限制

DataHub为单向触发器,只支持在事件源端配置触发器。操作路径如下。

登录DataHub服务控制台,在项目列表页面,单击目标项目名称进入详情页面。在新建Topic面板,成功配置相关信息后,在Topic列表页签,单击已创建的Topic名称进入详情页面。sc_create_topic_for_datahub_trigger

然后在Topic详情页面,单击右上角的同步,通过创建函数计算Connector的方式配置数据同步功能,便可以在函数计算中使用DataHub数据。

sc_datahub_trigger

步骤一:配置函数入口测试参数

DataHub触发函数执行时,同步的信息以event的形式作为输入参数,通过同步调用的方式传给函数。您可以将DataHub传入的event信息作为参数,调试函数代码编写是否正确。

  1. 登录函数计算控制台,在左侧导航栏,单击服务及函数
  2. 在顶部菜单栏,选择地域,然后在服务列表页面,单击目标服务。
  3. 函数管理页面,单击目标函数名称。
  4. 在函数详情页面,单击函数代码页签,然后单击xialatubiao图标,从下拉列表中,选择配置测试参数
  5. 配置测试参数面板,选择创建新测试事件编辑已有测试事件页签,填写事件名称和事件内容。然后单击确定
    event格式示例如下所示。参数的详细信息,请参见event结构定义
    {
      "eventSource": "acs:datahub",
      "eventName": "acs:datahub:putRecord",
      "eventSourceARN": "/projects/test_project_name/topics/test_topic_name",
      "region": "cn-hangzhou",
      "records": [
        {
          "eventId": "0:12345",
          "systemTime": 1463000123000,
          "data": "[\"col1's value\",\"col2's value\"]"
        },
        {
          "eventId": "0:12346",
          "systemTime": 1463000156000,
          "data": "[\"col1's value\",\"col2's value\"]"
        }
      ]
    }

步骤二:编写函数代码并测试

  1. 登录函数计算控制台,在左侧导航栏,单击服务及函数
  2. 在顶部菜单栏,选择地域,然后在服务列表页面,单击目标服务。
  3. 函数管理页面,单击目标函数名称。
  4. 在函数详情页面,单击函数代码页签,在代码编辑器中编写代码,然后单击部署代码
    本文以Python 3.6为例。
    # -*- coding: utf-8 -*-
    import logging
    import json
    
    # To enable the initializer feature
    # please implement the initializer function as below:
    # def initializer(context):
    #   logger = logging.getLogger()
    #   logger.info('initializing')
    
    def handler(event, context):
      logger = logging.getLogger()
      logger.info('hello world')
      # parse and handle event here
      # evt = json.loads(event) 
      logger.info(event)
      return 'hello world'
  5. 单击函数代码页签的测试函数
    执行完成后,您可以在函数代码页签的上方查看执行结果。