ApsaraMQ for Kafka:Create a MaxCompute sink connector

Step 1: Create a MaxCompute table

Create a destination table on the MaxCompute client. For details, see Create a table.

With partitioning -- add a partition key column named time with type STRING:

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

Without partitioning:

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

Tip: Partitioning organizes data into time-based segments, which improves query performance and reduces scan costs for large tables. Enable partitioning if you expect high data volumes or need to query by time range. Skip partitioning for small tables or when time-based filtering is not needed.

After the statement runs successfully, verify the table on the Tables page in the MaxCompute console.

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 your instance resides.

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

3. On the Tasks page, click Create Task.

4. On the Create Task page, enter a Task Name and Description, then configure the following sections.

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

Step 3: Verify data delivery

Send a test message and confirm that it arrives in the MaxCompute table.

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

2. On the Topic Details page, click Send Message.

3. In the Start to Send and Consume Message panel, enter a test message and click OK.

   

4. In the MaxCompute console, query the partitions to confirm data arrived:

      show PARTITIONS kafka_to_maxcompute;

   

5. Query data in the target partition. Replace the time value with the actual partition value from the previous step: If the query returns the test message data, the connector is working correctly.

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