All Products
Search

This Product

    ApsaraMQ for Kafka:Create a MaxCompute sink connector

    Document Center

    ApsaraMQ for Kafka:Create a MaxCompute sink connector

    Last Updated:Jun 21, 2026

    This topic describes how to create a MaxCompute sink connector to export data from a topic in an ApsaraMQ for Kafka instance to a MaxCompute table.

    Prerequisites

    For detailed steps, see Prerequisites for sink connectors.

    Notes

    To use the MaxCompute partitioning feature, you must create an additional partition column named time with the STRING data type when you create a table.

    Step 1: Create the destination resource

    Create a table using the MaxCompute client. For more information, see Create a table.

    This tutorial uses a table named kafka_to_maxcompute as an example. The table contains three columns and uses the partitioning feature. The following SQL statement creates the table:

    CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT) PARTITIONED by (time STRING);

    If you do not use the partitioning feature, use the following statement:

    CREATE TABLE IF NOT EXISTS kafka_to_maxcompute(topic STRING,valueName STRING,valueAge BIGINT);

    After the statement runs successfully, the following result is returned:

    On the Tables page, view the information for the created kafka_to_maxcompute table. Its partition type is partitioned table, and its table type is internal table. The table schema contains three fields: topic (string), valueName (string), and valueAge (bigint). None of them are primary keys. The partition field is time (string).

    Step 2: Create and start the connector

    1. Log on to the ApsaraMQ for Kafka console. In the Resource Distribution section of the Overview page, select the region where the ApsaraMQ for Kafka instance that you want to manage resides.

    2. In the left-side navigation pane, choose Connector Ecosystem Integration > Tasks.

    3. On the Tasks page, click Create Task.

    4. In the Create Task page, configure the Task Name and Description parameters. Then, follow the on-screen instructions to configure other parameters.

      • Task Creation

        1. In the Source step, set the Data Provider parameter to ApsaraMQ for Kafka and follow the on-screen instructions to configure other parameters. Then, click Next Step. The following table describes the parameters.

          Parameter

          Description

          Example

          Region

          The region where the ApsaraMQ for Kafka instance resides.

          China (Hangzhou)

          ApsaraMQ for Kafka Instance

          The ID of the ApsaraMQ for Kafka instance in which the data that you want to route are produced.

          alikafka_post-cn-9hdsbdhd****

          Topic

          The topic on the ApsaraMQ for Kafka instance in which the data that you want to route are produced.

          guide-sink-topic

          Group ID

          The ID of the group on the ApsaraMQ for Kafka instance in which the data that you want to route are produced.

          • Quickly Create: The system automatically creates a group whose ID is in the GID_EVENTBRIDGE_xxx format.

          • Use Existing Group: Select the ID of an existing group that is not in use. If you select an existing group that is in use, the publishing and subscription of existing messages are affected.

          Use Existing Group

          Consumer Offset

          • Latest Offset: Messages are consumed from the latest offset.

          • Earliest Offset: Messages are consumed from the earliest offset.

          Latest Offset

          Network Configuration

          If cross-border data transmission is required, select Self-managed Internet. In other cases, select Basic Network.

          Basic Network

          Data Format

          The data format feature is used to encode binary data delivered from the source into a specific data format. Multiple data formats are supported. If you do not have special requirements on encoding, specify Json as the value.

          • Json: Binary data is encoded into JSON-formatted data based on UTF-8 encoding and then put into the payload. This is the default value.

          • Text: Binary data is encoded into strings based on UTF-8 encoding and then put into the payload.

          • Binary: Binary data is encoded into strings based on Base64 encoding and then put into the payload.

          Json

          Messages

          The maximum number of messages that can be sent in each function invocation. Requests are sent only when the number of messages in the backlog reaches the specified value. Valid values: 1 to 10000.

          2000

          Interval (Unit: Seconds)

          The time interval at which the function is invoked. The system sends the aggregated messages to Function Compute at the specified time interval. Valid values: 0 to 15. Unit: seconds. The value 0 indicates that messages are sent immediately after aggregation.

          3

        2. In the Filtering step, define an event pattern to filter requests. For more information, see Event patterns.

        3. In the Transformation step, configure data cleansing to implement complex data processing capabilities such as splitting, mapping, enrichment, and dynamic routing. For more information, see Use Function Compute to clean message data.

        4. In the Sink step, set Service Type to acs.maxcompute and configure the following parameters.

          Parameter

          Description

          Example

          AccessKey ID

          The AccessKey ID for your Alibaba Cloud account, used to access the MaxCompute service.

          yourAccessKeyID

          AccessKey Secret

          The AccessKey secret for your Alibaba Cloud account.

          yourAccessKeySecret

          MaxCompute Project Name

          Select an existing MaxCompute project.

          test_compute

          MaxCompute Table Name

          Select an existing MaxCompute table.

          kafka_to_maxcompute

          MaxCompute Table Input Parameter

          After you select a table, its column names and data types are displayed. You only need to configure a value extraction rule for each column. The following code shows a message example. In this example, the value for the topic column is extracted from the topic field. Therefore, the value extraction rule is defined as $.topic.

          {
  'data': {
    'topic': 't_test',
    'partition': 2,
    'offset': 1,
    'timestamp': 1717048990499,
    'headers': {
      'headers': [],
      'isReadOnly': False
    },
    'key': 'MaxCompute-K1',
    'value': 'MaxCompute-V1'
  },
  'id': '9b05fc19-9838-4990-bb49-ddb942307d3f-2-1',
  'source': 'acs:alikafka',
  'specversion': '1.0',
  'type': 'alikafka:Topic:Message',
  'datacontenttype': 'application/json; charset=utf-8',
  'time': '2024-05-30T06:03:10.499Z',
  'aliyunaccountid': '1413397765616316'
}

          topic: $.data.topic

          valuename: $.data.value

          valueage: $.data.offset

          Partition Dimension

          • Close: Disables the partitioning feature.

          • Enable: Enables the partitioning feature.

            If you enable partitioning, you must configure parameters such as the partition value:

            • The partition value supports the time variables {yyyy}, {MM}, {dd}, {HH}, and {mm}, which represent the year, month, day, hour, and minute, respectively. Time variables are case-sensitive.

            • The partition value can also be a constant.

          Enable

          {yyyy}-{MM}-{dd}.{HH}:{mm}.suffix

          Network configuration

          • VPC: Deliver Kafka messages to MaxCompute through a VPC.

          • Public Network: Deliver Kafka messages to MaxCompute over the internet.

          internet

          VPC

          Select a VPC ID. This parameter is required only if you set Network Configuration to VPC.

          vpc-bp17fapfdj0dwzjkd****

          vSwitch

          Select a vSwitch ID. This parameter is required only if you set Network Configuration to VPC.

          vsw-bp1gbjhj53hdjdkg****

          Security Group

          Select a security group. This parameter is required only if you set Network Configuration to VPC.

          test_group

      • Task Properties

        Configure retry policies and dead-letter queues to handle delivery errors. For more information, see Retry policies and dead-letter queues.

    5. After you complete the preceding configurations, click Save. On the Tasks page, find the MaxCompute sink connector task that you created. The Status column shows Starting. When the status changes to Running, the creation process is complete.

    Step 3: Test the connector

    1. On the Tasks page, find the MaxCompute sink connector and click the source topic in the Event Source column.

    2. On the topic details page, click Send Test Message.

    3. In the Start to Send and Consume Message panel, configure the message content as follows, and then click OK.

      On the Console tab, set Message key to MaxCompute-K1, set Message body to MaxCompute-V1, and set Send to specified partition to No.

    4. Go to the MaxCompute console and run the following SQL statement to view partition information.

      show PARTITIONS kafka_to_maxcompute;

      The following result is returned:

      OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
OK
time=2024-05-31.16:37.suffix
OK
2024-05-31 16:42:49 INFO ==================================================================
2024-05-31 16:42:49 INFO Exit code of the Shell command 0
2024-05-31 16:42:49 INFO --- Invocation of Shell command completed ---
2024-05-31 16:42:49 INFO Shell run successfully!
2024-05-31 16:42:49 INFO Current task status: FINISH
2024-05-31 16:42:49 INFO Cost time is: 1.411s

    5. Based on the partition information, run the following statement to view the data in the partition.

      SELECT * FROM kafka_to_maxcompute WHERE time="2024-05-31.16:37.suffix";

      The query returns one data record with the following column values: topic is xxx (masked), valueName is MaxCompute-V1, valueAge is 4, and time is 2024-05-31.16:37.suffix. This indicates that data was successfully written from ApsaraMQ for Kafka to the partitioned table in MaxCompute.