Integrate with DataWorks data integration-PolarDB(PolarDB)-阿里云帮助中心

Using a generic MySQL data source to synchronize data from PolarDB-X 2.0 in DataWorks can lead to inefficient parallel reads, excessive database resource consumption, or task failures. To ensure efficient and stable data synchronization, use the dedicated data source type optimized for PolarDB-X 2.0. This topic explains how to configure and use the PolarDB-X 2.0 data source and introduces its unique optimization parameters.

Overview

The PolarDB-X 2.0 data source in DataWorks is designed for the distributed architecture of PolarDB-X. Unlike a generic MySQL data source, it can identify the topology of PolarDB-X partitioned tables, enabling parallel data reads by partition. This optimized read pattern maximizes the use of database resources, significantly improves data synchronization performance, and prevents inefficient scan methods from impacting production workloads.

Scope

This feature applies only to PolarDB-X 2.0 instances.

Note

If you are using a PolarDB-X 1.0 instance, select the DRDS (PolarDB-X 1.0) data source.

Procedure

Step 1: Configure account permissions

To run a data synchronization task, the database account must have the following permissions.

Offline synchronization:
- Offline read (Reader): The account must have SELECT permission on the destination database.
  
  Note
  Before execution, the Reader retrieves table topology information by running the SHOW TOPOLOGY FROM <table_name> command. Therefore, ensure the account has read permission on the destination database.
- Offline write (Writer): The account must have INSERT, DELETE, and UPDATE permissions on the destination table.
Real-time synchronization (entire database):
- A privileged account can read the binary log by default for real-time synchronization.
- A standard account must be granted SELECT, REPLICATION SLAVE, and REPLICATION CLIENT permissions by a privileged account.

Step 2: Create a data source

When you add a data source in the DataWorks console, you must select PolarDB-X 2.0. For information about other parameters and configurations, see Create a data source.

Log on to the DataWorks console. In the target region, click More > Management Center in the left-side navigation pane. Select a workspace from the drop-down list and click Go to Management Center.
In the navigation pane on the left, click Data Source to go to the Data Source List page.
In the upper-left corner of the page, click Add Data Source.

In the Add Data Source dialog box, click the Relational Databases category, and then select PolarDB-X 2.0.

Step 3: (Optional) Use custom endpoints for workload isolation

If many DataWorks tasks create high load on specific compute nodes, you can achieve workload isolation by routing DataWorks traffic to dedicated compute nodes using a custom endpoint.

Create a custom endpoint
1. Go to the PolarDB-X console and navigate to the target instance.
2. In the left-side navigation pane, click Configuration and Management > Connection Management.
3. Go to the Custom Endpoint tab and click Create Custom Connection to configure the parameters.
Note
- If your instance is a three-zone instance, it includes standby compute nodes. These standby compute nodes do not handle business traffic and are ideal for DataWorks. In the Compute node list on the instance details page, you can find inactive nodes, which are the standby nodes. On the Topology Information page, in the Compute Node List, nodes with the master role have an Activated status and nodes with the standby role have a Not Activated status. These roles are typically located in different zones.
- If your instance is not a three-zone instance, you can do one of the following:
  - Change the deployment from a single zone to three zones. A three-zone instance costs the same as a single-zone instance.
  - Create multiple custom endpoints for your business services and DataWorks.
Use the custom endpoint in DataWorks: When you create or modify a DataWorks data source, select Connection String Mode for Configuration Method and enter the previously created custom endpoint in the Host Address/IP field.

Step 4: Configure a data synchronization task

We recommend using the code editor to configure the data synchronization task, as some advanced parameters can only be set there. For detailed steps and instructions, see Configure an offline synchronization task by using the code editor.

Reader

The following table describes the core parameters for the Reader. Some advanced parameters are available only in the code editor.

Parameter	Description	Required	Default
`table`	The name of the source table to synchronize. Only a single table can be specified.	Yes	None
`column`	The columns to read, defined as a JSON array. To select all columns, use `[*]`. You can select a subset of columns to synchronize. You can reorder columns. The order does not need to follow the table schema. Supports any expression that can appear in a SQL `SELECT` statement. You must explicitly specify the columns to synchronize. This parameter cannot be empty. The following code shows a complex example: `["c1+c2", "FROM_UNIXTIME(c3)", "CAST(c4 AS SIGNED)", "c5", "c5", "c5", "IFNULL(c6,0)", "IF(c7>10,c7,c8)", ]`	Yes	None
`where`	The data filtering condition is directly appended to the `SELECT` statement's `WHERE` clause. Use this feature with caution to ensure that the statement has an efficient execution plan. For example: `gmt_create>='2025-09-01'`. Note The filter condition is not applied at the DataWorks layer but is appended directly to the SQL query. For example, the preceding example is equivalent to running the `SELECT * FROM tbl WHERE gmt_create>='2025-09-01'` query. You must check whether this query has performance issues: If no index is created on the `gmt_create` column, the query causes a full table scan. Even if an index exists on the `gmt_create` column, issues such as incorrect index selection by the database or high costs of table lookups may occur. We do not recommend that you use this parameter to specify filter conditions on a production database. If you must use this parameter, carefully evaluate the performance of the SQL query.	No	None
`onePartitionOneTask`	This is a core optimization parameter. For partitioned tables, we recommend setting this parameter to `true`. When this setting is enabled, DataWorks reads data one partition at a time. Each read query targets only a single partition, which helps distribute the workload evenly across read threads. Note When you use the `onePartitionOneTask` parameter, the default read timeout for a single shard is one hour. If synchronizing a large partition takes longer than this limit, the task fails.	No	`false`
`splitPk`	This parameter is deprecated.	No	None

Sample script

{
  "stepType": "polardbx20",
  "parameter": {
    "checkSlave": "false",
    "slaveDelayLimit": "300",
    "datasource": "ms_polardbx",
    "envType": 1,
    "column": ["c1+c2", 
      "FROM_UNIXTIME(c3)", 
      "CAST(c4 AS SIGNED)", 
      "c5", 
      "c5",
      "c5",
      "IFNULL(c6,0)",
      "IF(c7>10,c7,c8)",
      ],
    "tableComment": "",
    "onePartitionOneTask":true    // Recommended when reading data from a partitioned table.
  },
  "name": "Reader",
  "category": "reader"
}

Writer

The following table describes the core parameters for the Writer.

Parameter	Description	Required	Default
`table`	The name of the destination table.	Yes	None
`writeMode`	The import mode. The following modes are supported: `insert into`: If a primary key or unique key conflict occurs, the conflicting row is not written and is recorded as dirty data. `insert ignore`: A common mode. If a primary key or unique key conflict occurs, the conflicting row is ignored, and the existing record in the destination table is preserved. `replace into`: If a primary key or unique key conflict occurs, the existing row is deleted before the new row is inserted. The new row replaces all fields of the existing row. `upsert`: A common mode. This mode uses an `INSERT INTO ... ON DUPLICATE KEY UPDATE ...` statement. If a primary key or unique key conflict occurs, all fields are updated by default. You can use the `updateColumn` parameter to specify the columns to update. Default behavior (update all columns): The new row replaces all fields of the existing row, which is similar to the behavior of `replace into`. Update specific columns: Specify the columns to update in the updateColumn parameter. Example: `"writeMode":"upsert", "updateColumn":["c1","c2"]` In this case, the `INSERT INTO ... ON DUPLICATE KEY UPDATE c1=VALUES(c1), c2=VALUES(c2)` statement is generated. `upsert complex`: Advanced usage. If a primary key or unique key conflict occurs, this mode executes the complex update logic that is defined in the `updateColumn` parameter.	No	`insert into`
`updateColumn`	Used with the `upsert` or `upsert complex` mode in `writeMode` to specify the columns to update or the update logic.	No	None
`column`	The names of the destination table columns, separated by commas (,). Example: `"column": ["id", "name", "age"]`. Use `["*"]` to specify all columns.	Yes	None
`preSql`	An SQL statement to execute before the data synchronization task starts. The codeless UI allows only a single SQL statement, while the code editor supports multiple SQL statements.	No	None
`postSql`	An SQL statement to execute after the data synchronization task is complete. The codeless UI allows only a single SQL statement, while the code editor supports multiple SQL statements.	No	None
`batchSize`	The number of records to write per batch. Increasing this value can reduce the number of network interactions between the data synchronization system and PolarDB-X 2.0, which improves overall throughput. However, a value that is too high increases memory consumption and may cause out-of-memory errors.	No	256
`enableRetry`	Specifies whether to enable automatic retries for issues such as deadlocks or lock wait timeouts. If the table has a primary key or unique key, we recommend that you set this parameter to `true` to enable automatic retries. Note If the table does not have a primary key or unique key, retrying may cause data duplication.	No	`false`
`retryableException`	Used with `enableRetry`. In addition to the default list of exceptions, you can add other exceptions. In most cases, you do not need to modify this parameter.	No	["Lock wait timeout exceeded", "Deadlock found when trying to get lock", "Query execution was interrupted"]
`maxRetryTimes`	Used with `enableRetry`. This parameter specifies the maximum number of retries. In most cases, you do not need to modify this parameter.	No	100
`enableReConnect`	Specifies whether to automatically reconnect and retry a task after a connection is interrupted, for example, due to a transient network issue or a primary/standby failover. If the table has a primary key or unique key, we recommend that you set this parameter to `true` to enable automatic retries. Note If the table does not have a primary key or unique key, retrying may cause data duplication.	No	`false`

Sample script

{
  "stepType": "polardbx20",
  "parameter": {
    "postSql": [],
    "datasource": "ms_polardbx",
    "envType": 1,
    "column": [
      "id",
      "c1",
      "c2",
      "c3",
      "c4"
    ],
    "tableComment": "",
    "writeMode": "insert ignore",
    "batchSize": 256,
    "table": "tbl",
    "preSql": [],
    "enableRetry": true,	// Recommended if the table has a primary or unique key.
    "enableReConnect": true	// Recommended if the table has a primary or unique key.
  },
  "name": "Writer",
  "category": "writer"
}

FAQ

Why does my task fail after it runs for more than 8 hours?

This issue likely occurs because the task exceeded the default maximum transaction duration (8 hours) in PolarDB-X 2.0. Use a privileged account to run the following command to increase this value (unit: seconds):

Note

When this issue occurs, the task execution time is often exactly 8 hours but can also be much longer. If a task runs for more than 8 hours, check for this issue first.

SET GLOBAL max_trx_duration = 100000;

How do I resolve the XResult stream fetch result timeout error when I read a single large shard?

This error indicates that reading a single shard exceeded the default 1-hour limit. Use a privileged account to run the following command to increase the socket timeout (unit: milliseconds):

SET GLOBAL socket_timeout = 36000000;

How do I resolve the Application was streaming results when the connection failed error?

This is a generic connection error. Check and optimize your configuration as follows:

Confirm that the DataWorks data source type is PolarDB-X 2.0.
In the Reader configuration, set onePartitionOneTask=true for partitioned tables.
Increase the values of the max_trx_duration and socket_timeout parameters as needed.

How can I identify whether an SQL query originates from DataWorks?

Read queries from DataWorks are often slow SQL queries. This is normal and expected for data extraction operations. If you are using the PolarDB-X 2.0 data source, the SQL statement typically includes an optimizer hint such as /*+TDDL:SCAN(...)*/ at the beginning, which specifies the physical shard to scan. Example:

/*+TDDL:SCAN(TABLE="mocktable",REAL_TABLE=("t1_00"),NODE="GROUP_00")*/SELECT ...

Why can't I find some parameters such as onePartitionOneTask in the UI?

Some advanced parameters, such as onePartitionOneTask, are available only in the code editor. To access them, switch from the codeless UI to the code editor by clicking </> (Convert to Script) in the toolbar on the node editing page.