DataHub:Create a Function Compute sync task

Create a Function Compute sync task

Step 1: Create a Function Compute function

1.1 Create a service

Create a service in the Function Compute console. Skip this step if you have already created a function.

1.2 Create a function

In the service you just created, click Create Function and enter your code. For detailed instructions, see Function overview. On the trigger page, click Skip, and then click Finish.

Step 2: Authorize the service-linked role

Selecting the STS authentication mode automatically creates a DataHub service-linked role. DataHub uses this role to access Function Compute to sync data.

Step 3: Create a DataHub topic

See Quick start (sync example).

Step 4: Create a DataHub Function Compute connector

4.1 Go to the topic details page

In the upper-right corner of the topic details page, click the + Sync button.

4.2 Select the Function Compute connector

In the list of sync types, select Function Compute.

4.3 Configure the settings

In the Create Connector dialog box, in addition to endpoint, service, function, and Start Time, configure the fields to import (select the topic fields to sync, such as field1, field2, and field3) and the authentication mode (for example, STS). When you finish, click Create.

• Endpoint: The endpoint for the Function Compute service. You must provide an internal service address in the format https://<account_id>.fc.<region>.aliyuncs.com. For example, the endpoint for Function Compute in the China (Shanghai) region is https://12423423992.fc.cn-shanghai-internal.aliyuncs.com. For more information, see Service endpoints.

• Service: The name of the destination Function Compute service.

• Function: The name of the destination function.

• Start Time: The position in the DataHub topic from which Function Compute starts processing records.

  Event structure

  Data sent to Function Compute has the following structure.

{
  "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\"]"
    }
  ]
}

Where:

• eventSource: The event source. The value is always acs:datahub.

• eventName: The event name. For data from DataHub, the value is acs:datahub:putRecord.

• eventSourceARN: The event source identifier. It includes the DataHub project and topic name, such as /projects/test_project_name/topics/test_topic_name.

• region: The region ID of the source DataHub service, such as cn-hangzhou.

• records: The list of records included in the event.

  • eventId: The record ID, in the format shardId:SequenceNumber.

  • systemTime: The timestamp in milliseconds when DataHub stored the event.

  • data: The data payload of the event. If the topic type is Tuple, this field is a list where each string element corresponds to a field in the topic. If the topic type is Blob, this field is a string.

    Notes

• You must use an internal service domain. The specified Function Compute service and function must exist.

• DataHub supports only invoking functions in synchronous mode to ensure that data is processed in order.

• If a function encounters a runtime error, DataHub retries the invocation after 1 second. If the number of failures reaches 512, DataHub suspends the task.

• You can view the task status, checkpoint, and detailed error information in the web console.