DataWorks:HBase data source

HBase Reader

HBase Reader supports normal mode and multiVersionFixedColumn mode.

• normal mode: Treats an HBase table as a standard two-dimensional table and retrieves the latest version of data.

  hbase(main):017:0> scan 'users'
  ROW                                   COLUMN+CELL
  lisi                                 column=address:city, timestamp=1457101972764, value=beijing
  lisi                                 column=address:contry, timestamp=1457102773908, value=china
  lisi                                 column=address:province, timestamp=1457101972736, value=beijing
  lisi                                 column=info:age, timestamp=1457101972548, value=27
  lisi                                 column=info:birthday, timestamp=1457101972604, value=1987-06-17
  lisi                                 column=info:company, timestamp=1457101972653, value=baidu
  xiaoming                             column=address:city, timestamp=1457082196082, value=hangzhou
  xiaoming                             column=address:contry, timestamp=1457082195729, value=china
  xiaoming                             column=address:province, timestamp=1457082195773, value=zhejiang
  xiaoming                             column=info:age, timestamp=1457082218735, value=29
  xiaoming                             column=info:birthday, timestamp=1457082186830, value=1987-06-17
  xiaoming                             column=info:company, timestamp=1457082189826, value=alibaba
  2 row(s) in 0.0580 seconds

  The following table displays the output.

  rowKey	address:city	address:contry	address:province	info:age	info:birthday	info:company
  lisi	beijing	china	beijing	27	1987-06-17	baidu
  xiaoming	hangzhou	china	zhejiang	29	1987-06-17	alibaba

• multiVersionFixedColumn mode: Treats the HBase table as a vertical table. Each record consists of four columns: rowKey, family:qualifier, timestamp, and value. You must specify the columns to read. This mode treats each cell value as a separate record. If a cell has multiple versions, the mode generates a separate record for each version.

  hbase(main):018:0> scan 'users',{VERSIONS=>5}
  ROW                                   COLUMN+CELL
  lisi                                 column=address:city, timestamp=1457101972764, value=beijing
  lisi                                 column=address:contry, timestamp=1457102773908, value=china
  lisi                                 column=address:province, timestamp=1457101972736, value=beijing
  lisi                                 column=info:age, timestamp=1457101972548, value=27
  lisi                                 column=info:birthday, timestamp=1457101972604, value=1987-06-17
  lisi                                 column=info:company, timestamp=1457101972653, value=baidu
  xiaoming                             column=address:city, timestamp=1457082196082, value=hangzhou
  xiaoming                             column=address:contry, timestamp=1457082195729, value=china
  xiaoming                             column=address:province, timestamp=1457082195773, value=zhejiang
  xiaoming                             column=info:age, timestamp=1457082218735, value=29
  xiaoming                             column=info:age, timestamp=1457082178630, value=24
  xiaoming                             column=info:birthday, timestamp=1457082186830, value=1987-06-17
  xiaoming                             column=info:company, timestamp=1457082189826, value=alibaba
  2 row(s) in 0.0260 seconds

  The following table displays the output.

  rowKey	column:qualifier	timestamp	Value
  lisi	address:city	1457101972764	beijing
  lisi	address:contry	1457102773908	china
  lisi	address:province	1457101972736	beijing
  lisi	info:age	1457101972548	27
  lisi	info:birthday	1457101972604	1987-06-17
  lisi	info:company	1457101972653	baidu
  xiaoming	address:city	1457082196082	hangzhou
  xiaoming	address:contry	1457082195729	china
  xiaoming	address:province	1457082195773	zhejiang
  xiaoming	info:age	1457082218735	29
  xiaoming	info:age	1457082178630	24
  xiaoming	info:birthday	1457082186830	1987-06-17
  xiaoming	info:company	1457082189826	alibaba

HBase Writer

• The HBase Writer can generate a rowKey by concatenating multiple columns from the source.

• The HBase Writer can set the version (timestamp) of the data in the following ways:

  • Using the current time.

  • Using a value from a source column.

  • Using a user-specified time.

Batch read

• The following table lists the data type mappings for HBase Reader.

  Type	Data Integration column type	Database data type
  Integer	long	short, int, and long
  Floating-point	double	float and double
  String	string	binary_string and string
  Date and Time	date	date
  Byte	bytes	bytes
  Boolean	boolean	boolean

• HBase20xsql Reader supports most, but not all, Phoenix data types. Verify that your data types are supported before use.

• The following table lists the type mappings used by HBase20xsql Reader for Phoenix data types.

  DataX internal type	Phoenix data type
  long	INTEGER, TINYINT, SMALLINT, and BIGINT
  double	FLOAT, DECIMAL, and DOUBLE
  string	CHAR and VARCHAR
  date	DATE, TIME, and TIMESTAMP
  bytes	BINARY and VARBINARY
  boolean	BOOLEAN

Batch write

The following table lists the data type mappings for HBase Writer.

Single-table offline synchronization task

• For instructions, see Configure a task in the codeless UI and Configure a task in the code editor.

  By default, Wizard Mode does not display the Field Mapping section because HBase is a schemaless data source. You must configure the field mapping manually:

  • When HBase is the data source, configure Source Field in the following format: data_type|column_family:column_name.

  • When HBase is the data destination, configure both Destination Field and rowkey. For Destination Field, use the format source_field_index|data_type|column_family:column_name. For rowkey, use the format source_primary_key_index|data_type.

  Note

  Each field must be on a separate line.

• For a complete list of parameters and script examples in Script Mode, see Appendix: Script demos 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 Configure a task in the code editor. 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.

HBase Reader script demo

{
    "type":"job",
    "version":"2.0",// The version number.
    "steps":[
        {
            "stepType":"hbase",// The plugin name.
            "parameter":{
                "mode":"normal",// The mode for reading data from HBase. Valid values: normal and multiVersionFixedColumn.
                "scanCacheSize":"256",// The number of rows the client reads from the server per RPC.
                "scanBatchSize":"100",// The number of columns the client reads from the server per RPC.
                "hbaseVersion":"094x/11x",// The HBase version.
                "column":[// The columns to read.
                    {
                        "name":"rowkey",// The column name.
                        "type":"string"// The data type.
                    },
                    {
                        "name":"columnFamilyName1:columnName1",
                        "type":"string"
                    },
                    {
                        "name":"columnFamilyName2:columnName2",
                        "format":"yyyy-MM-dd",
                        "type":"date"
                    },
                    {
                        "name":"columnFamilyName3:columnName3",
                        "type":"long"
                    }
                ],
                "range":{// The rowkey range for the HBase Reader.
                    "endRowkey":"",// The end rowkey.
                    "isBinaryRowkey":true,// Specifies how to convert startRowkey and endRowkey to byte arrays. The default value is false.
                    "startRowkey":""// The start rowkey.
                },
                "maxVersion":"",// The number of versions to read in multi-version mode.
                "encoding":"UTF-8",// The encoding format.
                "table":"",// The table name.
                "hbaseConfig":{// Connection configuration for the HBase cluster, in JSON format.
                    "hbase.zookeeper.quorum":"hostname",
                    "hbase.rootdir":"hdfs://ip:port/database",
                    "hbase.cluster.distributed":"true"
                }
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// The maximum number of error records allowed.
        },
        "speed":{
            "throttle":true,// Enables throttling. If true, throttling is enabled based on the mbps value. If false, the mbps parameter is ignored.
            "concurrent":1,// The number of concurrent tasks for the job.
            "mbps":"12"// The throttling rate. In this example, 1 mbps is equal to 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase Reader parameters

HBase Writer script demo

{
    "type":"job",
    "version":"2.0",// The version number.
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"hbase",// The plugin name.
            "parameter":{
                "mode":"normal",// The mode for writing data to HBase.
                "walFlag":"false",// Specifies whether to write to the Write-Ahead Log (WAL). A value of false disables it.
                "hbaseVersion":"094x",// The HBase version.
                "rowkeyColumn":[// The columns to use for the rowkey.
                    {
                        "index":"0",// The serial number.
                        "type":"string"// The data type.
                    },
                    {
                        "index":"-1",
                        "type":"string",
                        "value":"_"
                    }
                ],
                "nullMode":"skip",// Specifies how to handle null values.
                "column":[// The HBase columns to write to.
                    {
                        "name":"columnFamilyName1:columnName1",// The column name.
                        "index":"0",// The index number.
                        "type":"string"// The data type.
                    },
                    {
                        "name":"columnFamilyName2:columnName2",
                        "index":"1",
                        "type":"string"
                    },
                    {
                        "name":"columnFamilyName3:columnName3",
                        "index":"2",
                        "type":"string"
                    }
                ],
                "encoding":"utf-8",// The encoding format.
                "table":"",// The table name.
                "hbaseConfig":{// Connection configuration for the HBase cluster, in JSON format.
                    "hbase.zookeeper.quorum":"hostname",
                    "hbase.rootdir":"hdfs://ip:port/database",
                    "hbase.cluster.distributed":"true"
                }
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// The maximum number of error records allowed.
        },
        "speed":{
            "throttle":true,// Enables throttling. If true, throttling is enabled based on the mbps value. If false, the mbps parameter is ignored.
            "concurrent":1, // The number of concurrent tasks for the job.
            "mbps":"12"// The throttling rate.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase Writer parameters

"rowkeyColumn": [
          {
            "index":0,
            "type":"string"
          },
          {
            "index":-1,
            "type":"string",
            "value":"_"
          }
      ]

HBase20xsql Reader script demo

{
    "type":"job",
    "version":"2.0",// The version number.
    "steps":[
        {
            "stepType":"hbase20xsql",// The plugin name.
            "parameter":{
                "queryServerAddress": "http://127.0.0.1:8765",  // The Phoenix QueryServer address.
                "serialization": "PROTOBUF",  // The QueryServer serialization format.
                "table": "TEST",    // The table to read.
                "column": ["ID", "NAME"],   // The columns to read.
                "splitKey": "ID"    // The split column, which must be the primary key of the table.
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// The maximum number of error records allowed.
        },
        "speed":{
            "throttle":true,// Enables throttling. If true, throttling is enabled based on the mbps value. If false, the mbps parameter is ignored.
            "concurrent":1,// The number of concurrent tasks for the job.
            "mbps":"12"// The throttling rate. In this example, 1 mbps is equal to 1 MB/s.
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase20xsql Reader parameters

HBase11xsql Writer script demo

{
  "type": "job",
  "version": "1.0",
  "configuration": {
    "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
            "throttle":true,// Enables throttling. If true, throttling is enabled based on the mbps value. If false, the mbps parameter is ignored.
            "concurrent":1, // The number of concurrent tasks for the job.
            "mbps":"1"// The throttling rate. In this example, 1 mbps is equal to 1 MB/s.
      }
    },
    "reader": {
      "plugin": "odps",
      "parameter": {
        "datasource": "",
        "table": "",
        "column": [],
        "partition": ""
      }
    },
    "plugin": "hbase11xsql",
    "parameter": {
      "table": "The target HBase table name, which is case-sensitive.",
      "hbaseConfig": {
        "hbase.zookeeper.quorum": "The ZooKeeper server address of the target HBase cluster.",
        "zookeeper.znode.parent": "The znode of the target HBase cluster."
      },
      "column": [
        "columnName"
      ],
      "batchSize": 256,
      "nullMode": "skip"
    }
  }
}

HBase11xsql Writer parameters