All Products
Search

This Product

    DataHub:Create a MaxCompute sync connector

    Document Center

    DataHub:Create a MaxCompute sync connector

    Last Updated:Aug 25, 2026

    Prerequisites

    Create a MaxCompute table

    Use a partitioned table to simplify data processing in MaxCompute.

    DataHub supports syncing data from TUPLE and BLOB topics to MaxCompute tables.

    • For TUPLE topics, the data types in the target MaxCompute table must be compatible with the DataHub topic schema. The following table shows the data type mappings.

      MaxCompute

      DataHub

      BIGINT

      BIGINT

      STRING

      STRING

      BOOLEAN

      BOOLEAN

      DOUBLE

      DOUBLE

      DATETIME

      TIMESTAMP

      DECIMAL

      DECIMAL

      TINYINT

      TINYINT

      SMALLINT

      SMALLINT

      INT

      INTEGER

      FLOAT

      FLOAT

      MAP

      Not supported

      ARRAY

      Not supported

      When creating the MaxCompute table schema, use only data types that DataHub supports.

    • For BLOB topics, the MaxCompute table schema must contain only a single STRING column. DataHub automatically syncs data to this column.

      DataHub

      MaxCompute

      BLOB

      STRING

    • To facilitate data tracing and troubleshooting, add a column named __rowkey__ STRING when you create the MaxCompute table. DataHub automatically syncs trace information to this column to aid data investigation.

    Account and permissions

    Timestamp unit

    • The TimestampUnit parameter specifies the unit for converting TIMESTAMP data before writing it to a date-related data type, such as DATETIME, in the destination system.

    • If the TIMESTAMP column contains values in seconds, select SECOND for TimestampUnit when creating the connector. If it contains values in milliseconds, select MILLISECOND. If it contains values in microseconds, select MICROSECOND.

    Important

    Due to MaxCompute's write standards, a high number of partitions slows down data synchronization from DataHub. When you create a MaxCompute sync task, limit the number of partitions as much as possible, especially in the USER_DEFINE synchronization mode.

    • Keep data within the same partition as contiguous as possible. Avoid frequent switching between partitions.

    • When the synchronization mode controls partition creation, do not create an excessive number of partitions.

    Note

    When the allowlist feature is enabled for a MaxCompute project, only devices on the allowlist can access the project. After enabling the MaxCompute IP allowlist, you must add the service to the allowlist to ensure the sync service can access the project. For details, see Overview.

    Synchronization modes

    Append mode

    Data is appended to the target table. This mode is suitable for scenarios where data only needs to be added and not updated.

    Upsert mode

    Upsert is a combination of Update and Insert with the following logic:

    • If a record with the same primary key exists in the target table, the existing record is updated.

    • If no record with the same primary key exists in the target table, a new record is inserted.

    The upsert mode provides a flexible way to handle data updates and insertions, keeping the data in the target table current.

    Note

    For more information about the Upsert feature in MaxCompute, see Basic concepts.

    Use cases

    • Updating data based on a primary key: Data may change over time, requiring existing records to be updated based on their primary key.

    • Maintaining data uniqueness in the target table: This mode prevents duplicate data by enforcing record uniqueness in the target table.

    • Handling duplicate data: Deduplicates large volumes of data based on a specific primary key.

    Configuration

    1. DataHub Topic Type: Must be a TUPLE topic.

    2. DataHub Topic Schema: The following two types are supported:

      1. DTS format: The schema type used when DTS syncs data to DataHub.

      2. Custom format: A user-created schema where you must select a column of the String type to serve as the operation column.

    3. ODPS Target Table: Must be a Transaction 2.0 table.

    Synchronization rules

    DTS format

    For data formats synced from DTS to DataHub, DataHub uses the fixed columns operation_flag, before_flag, and after_flag in the schema to determine how to sync data to the ODPS target table based on the following rules:

    operation_flag

    before_flag

    after_flag

    Operation type

    Action

    I

    *

    *

    UPSERT

    Updates the record in the target table based on the primary key.

    U

    Y

    N

    DELETE

    Deletes the record from the target table based on the primary key.

    U

    N

    Y

    UPSERT

    Updates the record in the target table based on the primary key.

    D

    *

    *

    DELETE

    Deletes the record from the target table based on the primary key.

    Custom format

    For user-created data, DataHub uses the selected operation column to determine how to sync data to the ODPS target table.

    Operation value

    Operation type

    Action

    U

    UPSERT

    Updates the record in the target table based on the primary key.

    D

    DELETE

    Deletes the record from the target table based on the primary key.

    Create a sync connector

    1. In the DataHub console, go to the details page of your topic.

    2. In the upper-right corner of the topic details page, click Sync to create a sync task.

    3. Select the MaxCompute job type to open the Create Connector page.

      Parameters:

      Parameter

      Option

      Required

      Description

      Project name

      /

      Yes

      The name of the MaxCompute project. You can select the project from the drop-down list. If you do not have permission to retrieve the project list, enter the name manually.

      Schema

      /

      No

      The name of the MaxCompute schema.

      Note

      To use the Schema feature, you must enable schema syntax development. For information about how to enable this and for more details about schemas, see Schema operations.

      Table

      /

      Yes

      The name of the MaxCompute table. You can select the table from the drop-down list. If you do not have permission to retrieve the table list, enter the name manually.

      Note

      If you use upsert mode, the target table must be a Transaction 2.0 table.

      Synchronization mode

      Append

      Yes

      Appends data to the target MaxCompute table.

      Upsert

      Updates or deletes data in the target MaxCompute table based on the primary key.

      For details, see the Synchronization modes section.

      Authentication method

      AK

      Authenticates by using an AccessKey pair.

      DataHub Default Role

      If you select this option, the datahub__access__role role is automatically authorized for the project. The permission policy for this role is:

      {
  "Statement": [
    {
      "Action": [
        "odps:CreateInstance",
        "odps:CreateTable",
        "odps:Describe",
        "odps:Alter",
        "odps:Update"
      ],
      "Effect": "Allow",
      "Resource": [
        "*"
      ]
    }
  ],
  "Version": "1"
}

      Custom Role

      A custom role that you can create and manage in the RAM console.

      Upsert method

      SYNC_CUSTOM 

      Required if Synchronization mode is set to Upsert. Not applicable if Synchronization mode is set to Append.

      Uses a custom upsert operation field.

      SYNC_NONE

      All data is written to the target table by using the upsert operation.

      SYNC_DTS

      Applies to scenarios where data is written from DTS to DataHub and the new DTS attachment column rules are enabled.

      SYNC_DTS_OLD  

      Applies to scenarios where data is written from DTS to DataHub and the old DTS attachment column rules are enabled.

      Primary key fields

      /

      The primary key columns of the target table. This is required for upsert synchronization.

      Upsert operation field

      /

      Required if Upsert method is set to SYNC_CUSTOM .

      Select a column of the String type to use as the Operation field. This field indicates whether the current data is synced to the downstream table as an Upsert or Delete operation.

    4. For more information about the upsert mode, see the Upsert mode section in this topic.

      • Fields to Import: You can configure DataHub to sync only a subset of columns to the MaxCompute table.

      • The partition mode determines the MaxCompute partition to which data is written, and DataHub supports the following partitioning methods:

        Partition mode

        Partition basis

        Topic types

        Description

        USER_DEFINE

        The value of the partition column in the record. The column must have the same name as the partition field in MaxCompute.

        TUPLE

        • The DataHub schema must include the MaxCompute partition fields.

        • The column value must be a UTF-8 string. The value can be empty, which indicates that the data is not partitioned.

        SYSTEM_TIME

        The time when the record is written to DataHub.

        TUPLE / BLOB

        • In the partition configuration, set the time conversion format for the MaxCompute partition.

        • Set the time zone information.

        EVENT_TIME

        The value of the event_time(TIMESTAMP) column in the record.

        TUPLE

        • In the partition configuration, set the time conversion format for the MaxCompute partition.

        • Set the time zone information.

        META_TIME

        The value of the __dh_meta_time__ attribute field in the record.

        TUPLE / BLOB

        • In the partition configuration, set the time conversion format for the MaxCompute partition.

        • Set the time zone information.

        The SYSTEM_TIME, EVENT_TIME, and META_TIME modes all convert timestamps to create MaxCompute partitions based on the timestamp and time zone configuration. The default unit is microseconds.

      • The partition configuration specifies how to convert timestamps to MaxCompute partitions. The console uses a default, fixed format for MaxCompute partitions. The configuration is as follows:

        Partition

        Time format

        Description

        ds

        %Y%m%d

        Day

        hh

        %H

        Hour

        mm

        %M

        Minute

        • Partition Interval: The time interval for converting timestamps into MaxCompute partitions. The range is from 15 minutes to 1440 minutes (1 day), with an increment of 15 minutes.

        • The time zone information (TimeZone) determines the time zone used to convert MaxCompute partitions based on timestamps.

        • Delimiter: When you sync BLOB data, you can specify a hexadecimal delimiter to split the data before syncing it to MaxCompute. For example, 0A represents the newline character (\n).

        • Base64 Encoding: DataHub stores BLOB data as binary data by default, while the corresponding column in MaxCompute is of the STRING type. Therefore, when you create a sync connector in the console, the data is Base64 encoded by default. For more customization options, use the SDK.

    View a sync connector

    Go to the connector's details page to view information such as its running status and checkpoint. You can also perform operations such as restarting and stopping the connector.

    Edit a sync connector

    On the sync task page, click Edit to modify the authentication method, imported fields, partition interval, time zone, and TimestampUnit value.

    Synchronization examples

    USER_DEFINE mode

    1. Create a DataHub topic.

      The topic schema must include the MaxCompute partition fields, and their type must be STRING.

    2. Write data to the DataHub topic by using the DataHub SDK.

      For testing, use the SDK to write several records with [ds,hh,mm] values of [20210304,01,15] and [20210304,02,15].

    3. Create a sync connector.

      In USER_DEFINE partition mode, you can set the partition configuration fields during synchronization. If the corresponding table does not exist in MaxCompute, it can be created automatically. In the import fields configuration, specify that fields f1 and f2 are synced, and exclude field f3.

    4. Confirm the synced data.

      You can view sync information for the connector in the DataHub console and query the data results in MaxCompute.

      In USER_DEFINE mode, DataHub syncs data to the corresponding partition based on the value of the MaxCompute grouping field.

    SYSTEM_TIME mode

    1. Create a DataHub topic.

      Because partitions are calculated based on the time data is written to DataHub, the topic schema only needs to contain data fields, not partition fields.

    2. Write data to the DataHub topic by using the DataHub SDK.

      For testing, use the SDK to write several records. Assume the current write time to DataHub is 2021-03-04 14:02:45.

    3. Create a sync connector.

      Ensure that the partition configuration matches the partitions in the MaxCompute table.

    4. Confirm the synced data.

      You can view sync information, such as the DoneTime, for the connector in the DataHub console and query the data results in MaxCompute.

      In SYSTEM_TIME mode, DataHub syncs data to the corresponding partition based on the time the data was written to DataHub.

    FAQ

    • Why does a timestamp field in MaxCompute show a date like 1970-01-19 after synchronization?

      • Cause: The default time unit for DataHub to MaxCompute synchronization is microseconds, but the timestamp written to DataHub was in milliseconds.

      • Solution: Ensure timestamps are written to DataHub in microseconds.