DataWorks:MaxCompute data source

Batch read

• MaxCompute Reader supports reading data from partitioned and non-partitioned tables, but not from virtual views or external tables.

• When you perform a batch read from a MaxCompute partitioned table, you cannot directly configure field mapping for partition key columns. To synchronize partition key values, you can add a custom field, manually enter the partition name, and then configure the field mapping.

• You can specify partition values by using scheduling parameters to enable automatic replacement, so that data from the corresponding partition is synchronized based on the scheduling time.

  For example, a partitioned table named t0 contains the id and name columns. The level-1 partition key is pt and the level-2 partition key is ds. To read data from the partition where pt=<business date> and ds=hangzhou, you must specify the partition values as pt=${scheduling parameter} and ds=hangzhou when you configure the source. Then, you can configure the column mapping for the id and name columns.

• You can write partition key columns to a destination table by adding them as custom fields.

• MaxCompute Reader supports data filtering by using a WHERE clause.

Batch write

• MaxCompute Writer does not support the VARCHAR data type if the source data contains null values.

• If the destination table is a DeltaTable, expand Advanced Configuration and set Visible After Synchronization to Yes. Otherwise, the task reports an error if the concurrency is greater than 1.

• Synchronizing data from a source to a MaxCompute external table is not supported.

• If a column in the destination table is not mapped to a source column, Data Integration sets its value to null after synchronization, even if a default value was specified when the table was created.

• MaxCompute Writer does not support pre-import SQL statements (preSql) or post-import SQL statements (postSql). If you need to perform SQL operations on the destination table before or after data is written (for example, running DELETE statements to remove partial data for incremental writes), configure a SQL node upstream or downstream of the data synchronization task to handle the required pre-processing or post-processing operations.

Real-time write

• Real-time synchronization tasks support serverless resource groups.

• Real-time synchronization tasks do not support synchronizing tables without a primary key.

• Synchronizing data from a source to a MaxCompute external table is not supported.

• When you perform real-time synchronization to the default MaxCompute data source (typically odps_first), a temporary AccessKey pair is used by default. The temporary AccessKey pair expires after 7 days, which causes the task to fail. The platform automatically restarts the task when it detects that the failure is caused by an expired temporary AccessKey pair. If you have configured monitoring rules for this type of alert, you will receive an alert notification.

• For one-click real-time synchronization to MaxCompute tasks, only historical full data can be queried on the day of configuration. Incremental data can be queried in MaxCompute only after the merge is completed the next day.

• One-click real-time synchronization to MaxCompute tasks generates a full partition every day. To prevent excessive storage usage, MaxCompute tables automatically created by these tasks have a default lifecycle of 30 days. If this does not meet your business requirements, you can click the corresponding MaxCompute table name when configuring the synchronization task to modify the lifecycle.

• Data Integration uses the MaxCompute engine synchronization data channel to upload and download data. For the SLA details of the synchronization data channel, see MaxCompute Tunnel overview. Evaluate your data synchronization technology choices based on the SLA of the MaxCompute engine synchronization data channel.

• For one-click real-time synchronization to MaxCompute in instance mode, the exclusive resource group for Data Integration must have a minimum specification of 8C16G.

• Only custom MaxCompute data sources in the same region as the current workspace are supported. Cross-region MaxCompute projects may pass the connectivity test, but during task execution, an error indicating that the engine does not exist is reported during the table creation phase in MaxCompute.

• When MaxCompute is used as the destination for full-database synchronization, if the table type is a regular table, only one-click real-time synchronization to MaxCompute and the incremental streaming mode for full-database real-time synchronization are supported. If the table type is Delta Table, both full-database real-time synchronization and one-click real-time synchronization to MaxCompute are supported.

  Note

  When you use a custom MaxCompute data source, the DataWorks project must still be associated with a MaxCompute engine. Otherwise, you cannot create MaxCompute SQL nodes, which causes the creation of the done-flag node for full synchronization to fail.

Columns supported by 1.0 data types

Columns supported by 2.0 data types and Hive-compatible data types

Data type conversion

The following table describes the data type conversions that MaxCompute Reader supports.

Connect to MaxCompute and enable project-level settings

• Log on to the MaxCompute client. For more information, see MaxCompute client.

• Enable MaxCompute project-level settings: Make sure that you have the required permissions. You can use a Project Owner account to perform related operations. For more information about MaxCompute permissions, see MaxCompute permissions.

Enable the ACID property

You can use a Project Owner account to run the following command on the client to enable the ACID property. For more information about ACID semantics in MaxCompute, see ACID semantics.

setproject odps.sql.acid.table.enable=true;

(Optional) Enable 2.0 data types

If you need to use the TIMESTAMP type in MaxCompute 2.0 data types, you can use a Project Owner account to run the following command on the client to enable 2.0 data types.

setproject odps.sql.type.system.odps2=true;

(Optional) Grant access to accounts

When you associate a MaxCompute compute resource with a workspace, a MaxCompute data source is created by default in DataWorks. You can use this data source for data synchronization in the current workspace. If you want to synchronize data from this MaxCompute data source in another workspace, make sure that the access account specified for the data source in the other workspace has the required permissions to access the MaxCompute project. For cross-account authorization, see Cross-account authorization.

Configure a single-table batch synchronization task

• For the procedure, see Configure a batch synchronization task in wizard mode and Configure a batch synchronization task in script mode.

• For a complete list of parameters and script demos for script mode, see Appendix: Script demos and parameter descriptions.

Configure a single-table real-time synchronization task

For the procedure, see Configure a single-table real-time synchronization task.

Configure a full-database synchronization task

For the procedure, see Synchronize data from a full database in batch mode, Synchronize data from a full database in real-time mode, and One-click real-time synchronization to MaxCompute.

Configure a batch synchronization task by using the code editor

If you want to configure a batch synchronization task by using the code editor, you must configure the related parameters in the script based on the unified script format requirements. For more information, see Script mode configuration. The following information describes the parameters that you must configure for data sources when you configure a batch synchronization task by using the code editor.

Reader script demo

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"odps",//The plug-in name.
            "parameter":{
                "partition":[],//The partition from which data is read.
                "isCompress":false,//Specifies whether to compress data.
                "datasource":"",//The data source.
                "column":[//The column information of the source table.
                    "id"
                ],
                "where": "",//The specific WHERE clause content when data filtering by using WHERE is enabled.
                "enableWhere":false,//Specifies whether to enable data filtering by using WHERE.
                "table":""//The table name.
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//The error count.
        },
        "speed":{
            "throttle":true,//If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
            "concurrent":1, //The concurrency.
            "mbps":"12"//The throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

If you want to specify the Tunnel Endpoint for MaxCompute, you can manually configure the data source in script mode. Replace "datasource":"", in the preceding example with the specific parameters of the data source. Example:

"accessId":"*******************",
"accessKey":"*******************",
"endpoint":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api",
"odpsServer":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api", 
"tunnelServer":"http://dt.eu-central-1.maxcompute.aliyun.com", 
"project":"*****", 

Reader script parameters

Writer script demo

The following is a script configuration example.

{
    "type":"job",
    "version":"2.0",//The version number.
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"odps",//The plug-in name.
            "parameter":{
                "partition":"",//The partition information.
                "truncate":true,//The cleanup rule.
                "isCompress":false,//Specifies whether to compress data.
                "datasource":"odps_first",//The data source name.
            "column": [//The source column names.
                "id",
                "name",
                "age",
                "sex",
                "salary",
                "interest"
                ],
                "table":""//The table name.
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//The error count, which specifies the maximum number of tolerable dirty data records.
        },
        "speed":{
            "throttle":true,//If throttle is set to false, the mbps parameter does not take effect and throttling is disabled. If throttle is set to true, throttling is enabled.
            "concurrent":1, //The concurrency.
            "mbps":"12"//The throttling rate. 1 mbps = 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

If you want to specify the Tunnel Endpoint for MaxCompute, you can manually configure the data source in script mode. Replace "datasource":"", in the preceding example with the specific parameters of the data source. Example:

"accessId":"<yourAccessKeyId>",
 "accessKey":"<yourAccessKeySecret>",
 "endpoint":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api",
 "odpsServer":"http://service.eu-central-1.maxcompute.aliyun-inc.com/api", 
"tunnelServer":"http://dt.eu-central-1.maxcompute.aliyun.com", 
"project":"**********", 

Writer script parameters