What Lindorm read and write capabilities does DataWorks support? - DataWorks

DataWorks Data Integration supports bidirectional data transfer through the Lindorm Reader and Lindorm Writer plug-ins. This topic describes how DataWorks reads data from and writes data to Lindorm.

Scope

The wide table engine supports Serverless resource groups (recommended) and exclusive resource groups for Data Integration.
The compute engine supports only Serverless resource groups.

Note

Lindorm is a multimodal database. For more information, see Lindorm documentation. DataWorks currently supports only the wide table engine and the compute engine.

Supported field types

Lindorm Reader and Lindorm Writer support most Lindorm data types, but some types are unsupported. Check your data types before use.

The following table lists the Lindorm type mappings for Lindorm Reader and Lindorm Writer.

Type category	Data type
Integer	INT, LONG, SHORT
Floating-point	DOUBLE, FLOAT, DOUBLE
String	STRING
Date and time	DATE
Boolean	BOOLEAN
Binary	BINARYSTRING

Develop data synchronization tasks

For information about the entry point for and the procedure of configuring a synchronization task, see the following configuration guides.

Single-table batch synchronization

Supported data sources: All source types supported by the Data Integration module
Configuration guide: Offline Synchronization Capabilities
For all script mode parameters and sample scripts, see Appendix: Sample scripts and parameters.

Single-table real-time synchronization

Supported data sources: Kafka, LogHub, and Hologres
Configuration guide: Single-Table Real-Time Synchronization Capabilities

Full-database real-time synchronization

Supported data source: PostgreSQL
Configuration guide: Configure real-time full-database synchronization tasks

Appendix: Sample scripts and parameters

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 sample scripts

Configure a job that extracts data from a Lindorm SQL Table in the wide table engine to local storage.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "lindorm",
            "parameter": {
                "mode": "FixedColumn",
                "caching": 128,
                "column": [
                   "id",
                  "value" 
                ],
                "envType": 1,
                "datasource": "lindorm",
                "tableMode": "tableService",
                "table": "lindorm_table"
            },
            "name": "lindormreader",
            "category": "reader"
        },
        {
            "stepType": "mysql",
            "parameter": {
                "postSql": [],
                "datasource": "lindorm",
                "session": [],
                "envType": 1,
                "column": [
                    "id",
                    "value" 
                ],
                "socketTimeout": 3600000,
                "writeMode": "insert",
                "batchSize": 1024,
                "encoding": "UTF-8",
                "table": "",
                "preSql": []
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "jvmOption": "",
        "executeMode": null,
        "errorLimit": {
            "record": "0"
        },
        "speed": {
        //Set the transfer speed in bytes per second. DataX tries to reach this speed without exceeding it.
        "byte": 1048576
      }
      //Error limits
      "errorLimit": {
        //Maximum number of error records. The job fails when this limit is exceeded.
        "record": 0,
        //Maximum percentage of error records. 1.0 means 100%, and 0.02 means 2%
        "percentage": 0.02
      }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Configure a job that extracts data from a Lindorm HBaseLike (WideColumn) table in the wide table engine to local storage.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "lindorm",
            "parameter": {
                "mode": "FixedColumn",
                "column":  [
                     "STRING|rowkey",
                      "INT|f:a"
                ],
                "envType": 1,
                "datasource": "lindorm",
                "tableMode": "wideColumn",
                "table":"lindorm_table"
            },
            "name": "lindormreader",
            "category": "reader"
        },
        {
            "stepType": "mysql",
            "parameter": {
                "postSql": [],
                "datasource": "_IDB.TAOBAO",
                "session": [],
                "envType": 1,
                "column": [
                    "id",
                    "value"
                ],
                "socketTimeout": 3600000,
                "guid": "",
                "writeMode": "insert",
                "batchSize": 1024,
                "encoding": "UTF-8",
                "table": "",
                "preSql": []
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "jvmOption": "",
        "executeMode": null,
        "errorLimit": {
            "record": "0"
        },
        "speed": {
        //Set the transfer speed in bytes per second. DataX tries to reach this speed without exceeding it.
        "byte": 1048576
      }
        //Error limits
        "errorLimit": {
        //Maximum number of error records. The job fails when this limit is exceeded.
        "record": 0,
        //Maximum percentage of error records. 1.0 means 100%, and 0.02 means 2%。
        "percentage": 0.02
      }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Configure a job that extracts data from a compute engine table to local storage.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "lindorm",
            "parameter": {
               "datasource": "lindorm_datasource",
                "column": [
                   "id",
                   "value"
                ],
                "tableComment": "",
                "where": "",
                "session": [],
                "splitPk": "id",
                "table": "auto_ob_149912212480"
            },
            "name": "lindormreader",
            "category": "reader"
        },
        {
            "stepType": "mysql",
            "parameter": {
                "postSql": [],
                "datasource": "_IDB.TAOBAO",
                "session": [],
                "envType": 1,
                "column": [
                    "id",
                    "value"
                ],
                "socketTimeout": 3600000,
                "guid": "",
                "writeMode": "insert",
                "batchSize": 1024,
                "encoding": "UTF-8",
                "table": "",
                "preSql": []
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "jvmOption": "",
        "executeMode": null,
        "errorLimit": {
            "record": "0"
        },
        "speed": {
        //Set the transfer speed in bytes per second. DataX tries to reach this speed without exceeding it.
        "byte": 1048576
      }
        //Error limits
        "errorLimit": {
        //Maximum number of error records. The job fails when this limit is exceeded.
        "record": 0,
        //Maximum percentage of error records. 1.0 means 100%, and 0.02 means 2%。
        "percentage": 0.02
      }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Reader script parameters

Parameter

Description

Required

Default value

mode

Specific to the wide table engine. Specifies the data read mode: fixed-column mode FixedColumn or dynamic-column mode DynamicColumn.

Yes

FixedColumn

tableMode

Specific to the wide table engine. Specifies either standard table SQL mode table or wide-column mode wideColumn. The default is table. If you select table mode, you can leave this parameter empty.

Empty by default

table

The name of the Lindorm table to read. Lindorm table names are case-sensitive.

Yes

None

encoding

Specific to the wide table engine. The encoding format, UTF-8 or GBK. Typically used to convert the Lindorm byte[] type stored as binary data to String.

UTF-8

caching

Specific to the wide table engine. The number of records retrieved in each batch. This can greatly reduce network interactions between the data synchronization system and Lindorm and increase overall throughput. An excessively large value can overload the Lindorm server or cause an out-of-memory (OOM) error in the synchronization process.

100

selects

Specific to the wide table engine. Table-type data currently does not support automatic shard splitting and runs with one concurrent task by default. Therefore, configure selects to split the data manually. Example:

selects": [
    "where(compare(\"id\", LESS, 5))",
    "where(and(compare(\"id\", GREATER_OR_EQUAL, 5), compare(\"id\", LESS, 10)))",
    "where(compare(\"id\", GREATER_OR_EQUAL, 10))"
    ],

Limits:

Only primary key columns and indexed columns may be used in query conditions. Conditions on regular columns trigger a full-table scan and affect the stability of the source cluster.

When a table has multiple primary key columns, query conditions must follow the leftmost-prefix rule: the first n-1 consecutive primary key columns must all use equality conditions. For example, suppose the table has primary key [id, order_time], regular columns [type, data], and a secondary index on the type column.

Recommended syntax examples:

SQL syntax	Plug-in syntax
where id >= 1 and id < 100	`where(and(compare(\"id\", GREATER_OR_EQUAL, 1), compare(\"id\", LESS, 100)))`
where id = 1 and order_time > 1234567	`where(and(compare(\"id\", EQUAL, 1), compare(\"order_time\", GREATER, 1234567)))`
where type = 'pay'	`where(compare(\"type\", EQUAL, \"pay\"))`

Not recommended syntax examples:

SQL syntax	Plug-in syntax	Reason
where order_time >= 1234567 and order_time < 5678910	`where(and(compare(\"order_time\", GREATER_OR_EQUAL, 1234567), compare(\"order_time\", LESS, 5678910)))`	Missing the leftmost primary key column `id`.
where id > 1 and order_time > 1234567	`where(and(compare(\"id\", GREATER, 1), compare(\"order_time\", GREATER, 1234567)))`	The leftmost primary key column `id` does not use an equality condition.
where data > 'xxx'	`where(compare(\"data\", GREATER, \"xxx\"))`	The data field is not a primary key column.

None

session

Specific to the compute engine. Session-level job parameters, such as set hive.execution.engine=tez.

None

splitPk

Specific to the compute engine. The split key for reading a compute engine table. If specified, splitPk identifies the field used to split the data, and data synchronization starts concurrent tasks to improve efficiency.

If splitPk is not specified, including when splitPk is absent or splitPk is empty, data synchronization reads the table over a single channel.
Currently, splitPk supports splitting only on integer data, not strings, floating-point numbers, dates, or other types.

None

columns

The list of fields to read. You can select a subset of columns and reorder them. Column selection exports only selected columns; reordering exports columns in an order different from the table schema.

For a table-type table, enter only the column names. The schema is obtained automatically from the table metadata. Example:
```
[
    "id",
    "name",
    "age",
    "birthday",
    "gender"
]
```
For an HBaseLike (widecolumn) table, use the following format:
```
[
    "STRING|rowkey",
    "INT|f:a",
    "DOUBLE|f:b"
]
```

Yes

None

Writer sample scripts

Configure a job that writes data from MySQL to a Lindorm SQL Table in the wide table engine.

{
  "type": "job",
  "version": "2.0",
  "steps": [
    {
      "stepType": "mysql",
      "parameter": {
        "checkSlave": true,
        "datasource": " ",
        "envType": 1,
        "column": [
          "id",
          "value"
        ],
        "socketTimeout": 3600000,
        "masterSlave": "slave",
        "connection": [
          {
            "datasource": " ",
            "table": []
          }
        ],
        "where": "",
        "splitPk": "",
        "encoding": "UTF-8",
        "print": true
      },
      "name": "mysqlReader",
      "category": "reader"
    },
    {
      "stepType": "lindorm",
      "parameter": {
        "nullMode": "skip",
        "datasource": "lindorm_datasource",
        "envType": 1,
        "column": [
          "id",
          "value"
        ],
        "dynamicColumn": "false",
        "table": "lindorm_table",
        "encoding": "utf8"
      },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
    "jvmOption": "",
    "executeMode": null,
    "speed": {
      //Set the transfer speed in bytes per second. DataX tries to reach this speed without exceeding it.
      "byte": 1048576
    },
    //Error limits
    "errorLimit": {
      //Maximum number of error records. The job fails when this limit is exceeded.
      "record": 0,
      //Maximum percentage of error records. 1.0 means 100%, and 0.02 means 2%。
      "percentage": 0.02
    }
  },
  "order": {
    "hops": [
      {
        "from": "Reader",
        "to": "Writer"
      }
    ]
  }
}

Configure a job that writes data from MySQL to a Lindorm HBaseLike (WideColumn) table in the wide table engine.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "mysql",
            "parameter": {
                "envType": 0,
                "datasource": " ",
                "column": [
                     "id",
                     "value"
                ],
                "connection": [
                    {
                        "datasource": " ",
                        "table": []
                    }
                ],
                "where": "",
                "splitPk": "",
                "encoding": "UTF-8"
            },
            "name": "Reader",
            "category": "reader"

        },
      {
          "stepType": "lindorm",
          "parameter": {
            "datasource": "lindorm_datasource",
            "table": "xxxxxx",
            "encoding": "utf8",
            "nullMode": "skip",
            "dynamicColumn": "false",
            "caching": 128,
            "column": [  //Map source fields in field order
                  "ROW|STRING", //Row key. This fixed configuration maps the first source field to the row key, such as id in this example.
                  "cf:name|STRING" //cf is the editable column family name; name is the editable destination column name
            ]
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting": {
        "jvmOption": "",
                "errorLimit": {
                        "record": "0"
                },
                "speed": {
                    "concurrent": 3,
                    "throttle": false
                }
    },
    "order": {
            "hops": [
               {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
  }

Configure a job that writes data from MySQL to a compute engine table.

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "mysql",
            "parameter": {
                "envType": 0,
                "datasource": " ",
                "column": [
                     "id",
                     "value"
                ],
                "connection": [
                    {
                        "datasource": " ",
                        "table": []
                    }
                ],
                "where": "",
                "splitPk": "",
                "encoding": "UTF-8"
            },
            "name": "Reader",
            "category": "reader"

        },
      {
          "stepType": "lindorm",
          "parameter": {
              "datasource": "lindorm_datasource",
              "table": "xxxxxx",
              "column": [ 
                 "id",
                "value"
               ],
              "formatType": "ICEBERG"
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting": {
        "jvmOption": "",
                "errorLimit": {
                        "record": "0"
                },
                "speed": {
                    "concurrent": 3,
                    "throttle": false
                }
    },
    "order": {
            "hops": [
               {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
  }

Writer script parameters

Parameter	Description	Required	Default value
table	The name of the Lindorm table to write to. Lindorm table names are case-sensitive.	Yes	None
encoding	Specific to the wide table engine. The encoding format, UTF-8 or GBK. Typically used to convert the Lindorm byte[] type stored as binary data to String.	No	UTF-8
columns	The list of fields to write. You can select a subset of columns and reorder them. Column selection exports only selected columns; reordering exports columns in an order different from the table schema. For a table-type table, enter only the column names. The schema is obtained automatically from the table metadata. For a widecolumn-type or table-type table.	Yes	None
nullMode	Specific to the wide table engine. When a value read from the source is null, set Lindorm Writer's nullMode parameter to specify how to handle it. SKIP: Do not write this column to Lindorm. EMPTY_BYTES: If the field value is null, write an empty byte array to the corresponding Lindorm field. NULL: Write a null value. DELETE: If the field value is null, delete the corresponding Lindorm field.	No	EMPTY_BYTES
formatType	Specific to the compute engine. Specifies the type of the table to synchronize. Valid values: iceberg parquet orc	No	None