DataWorks:HBase データソース

HBase リーダー

normal モードおよび multiVersionFixedColumn モードがサポートされています。設定方法については、「HBase フィールドマッピング設定ガイド」をご参照ください。

• normal モード：HBase テーブルを通常の二次元テーブル（ワイドテーブル）として読み取り、最新バージョンのデータを取得します。

  hbase:007:0> scan 'student'
  ROW                                   COLUMN+CELL
  s001                                 column=basic:age, timestamp=2026-03-09T14:41:40.240, value=20
  s001                                 column=basic:name, timestamp=2026-03-09T14:41:40.214, value=Tom
  s001                                 column=score:english, timestamp=2026-03-09T14:41:40.333, value=90
  s001                                 column=score:math, timestamp=2026-03-09T14:41:40.277, value=85
  1 row(s) in 0.0580 seconds 

  以下の表は、読み取られたデータの例を示しています。

  Rowkey	basic:age	basic:name	score:english	score:math
  s001	20	Tom	90	85

• multiVersionFixedColumn モードでは、HBase テーブルを狭いテーブル（ナローテーブル）として読み取ります。各返却レコードは、rowKey、family:qualifier、timestamp、および value の 4 つのカラムで構成されます。読み取るカラムを明示的に指定する必要があります。各セルの値は個別のレコードとして返却されます。セルに複数のバージョンがある場合、複数のレコードが返却されます。

  hbase:007:0> scan 'student',{VERSIONS=>5}
  ROW                                   COLUMN+CELL
  s001                                 column=basic:age, timestamp=2026-03-09T14:41:40.240, value=20
  s001                                 column=basic:age, timestamp=2026-03-09T14:30:00.100, value=19
  s001                                 column=basic:name, timestamp=2026-03-09T14:41:40.214, value=Tom
  s001                                 column=score:english, timestamp=2026-03-09T14:41:40.333, value=90
  s001                                 column=score:math, timestamp=2026-03-09T14:41:40.277, value=85
  1 row(s) in 0.0260 seconds }

  以下の表は、結果として得られる 4 カラム形式のデータの例を示しています。

  Rowkey	column:qualifier	Timestamp	Value
  s001	basic:age	2026-03-09T14:41:40.240	20
  s001	basic:age	2026-03-09T14:30:00.100	19
  s001	basic:name	2026-03-09T14:41:40.214	Tom
  s001	score:english	2026-03-09T14:41:40.333	90
  s001	score:math	2026-03-09T14:41:40.277	85

HBase ライター

• rowkey の生成ルール：現在、HBase ライターは、ソースから複数のフィールドを連結して HBase テーブルの rowkey を生成することをサポートしています。

• HBase へのデータ書き込み時に使用可能なタイムスタンプ（version）のオプションは以下のとおりです：

  • 現在時刻をバージョンとして使用します。

  • ソースカラムの値をバージョンとして使用します。

  • 固定時刻をバージョンとして指定します。

バッチ読み取り

• 以下の表は、HBase リーダーのデータ型マッピングを示しています。

  カテゴリ	データ統合カラム型	データベース型
  整数	long	short、int、および long
  浮動小数点	double	float および double
  文字列	string	binary_string および string
  日付および時刻	date	date
  バイト	bytes	bytes
  ブール値	boolean	boolean

• HBase20xsql リーダーは、ほとんどの Phoenix データ型をサポートしていますが、すべてのデータ型をサポートしているわけではありません。ご使用のデータ型がサポートされているか確認してください。

• 以下の表は、HBase20xsql リーダーのデータ型マッピングを示しています。

  DataX 内部型	Phoenix データ型
  long	INTEGER、TINYINT、SMALLINT、BIGINT
  double	FLOAT、DECIMAL、DOUBLE
  string	CHAR、VARCHAR
  date	DATE、TIME、TIMESTAMP
  bytes	BINARY、VARBINARY
  boolean	BOOLEAN

バッチ書き込み

以下の表は、HBase ライターのデータ型マッピングを示しています。

単一テーブルのバッチ同期

• 設定方法については、「コードレス UI の使用」および「コードエディタの使用」をご参照ください。

  コードレス UI を使用する場合、HBase は柔軟なスキーマを持つため、フィールドマッピングはデフォルトで表示されません。マッピングは手動で設定する必要があります。

  HBase をソースとして使用する場合、まず 出力モード を選択する必要があります：ワイドテーブル（normal モード）またはナローテーブル（multiVersionFixedColumn モード）。

  2 つのモード間でフィールドマッピングの設定方法は異なります：

  • ワイドテーブル（normal モード）：これはデフォルトモードです。HBase テーブルを標準的な二次元テーブルとして読み取り、最新バージョンのデータを取得します。HBase をソースとして使用する場合、ソースフィールド と ターゲットフィールド の間のマッピングを設定する必要があります。図に示すように、ソースとターゲットのフィールド間には 1 対 1 のマッピングが存在します。ソーステーブルには固定フィールドがないため、デフォルトでは行単位でフィールドがマッピングされます。このマッピングを変更するには、フィールド順序を手動で編集する必要があります。

    新しいデータ開発

    

    従来のデータ開発

    

    ターゲットテーブルの結果：

    

  • ナローテーブル（multiVersionFixedColumn モード）：各出力レコードは 4 つのカラム（rowKey、family:qualifier、timestamp、value）で構成され、複数のデータバージョンを読み取ることができます。ソースフィールドの ソースフィールド を ColumnFamily:Qualifier 形式（例：basic:age）で設定します。ターゲットは、4 カラム（row_key、cf、timestamp_col、value）で構成される固定テーブルです。このモードではフィールドマッピングは不要です。

    新しいデータ開発

    

    従来のデータ開発

    

    ターゲットテーブルの結果：

    

  • HBase をデータ送信先として使用する場合（normal モードのみ対応）、ターゲットフィールド および rowkey を別々に設定する必要があります。rowkey フィールドは、複数のソースフィールドを連結したものにすることができます。

• コードエディタのパラメーター一覧およびサンプルスクリプトについては、「付録：スクリプトのデモおよびパラメーター説明」をご参照ください。

コードエディタを使用したバッチ同期タスクの設定

コードエディタを使用してバッチ同期タスクを設定する場合、統一されたスクリプトフォーマット要件に基づき、関連するパラメーターをスクリプト内に設定する必要があります。詳細については、「コードエディタの使用」をご参照ください。以下に、コードエディタを使用してバッチ同期タスクを設定する際に、データソースに対して設定する必要のあるパラメーターについて説明します。

HBase リーダースクリプトの例

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"hbase",// プラグイン名。
            "parameter":{
                "mode":"normal",// HBase からデータを読み取るモード。有効な値：normal および multiVersionFixedColumn。
                "scanCacheSize":"256",// HBase クライアントがリモートプロシージャコール（RPC）ごとにサーバーから読み取る行数。
                "scanBatchSize":"100",// HBase クライアントが RPC ごとにサーバーから読み取るカラム数。
                "hbaseVersion":"094x/11x",// HBase のバージョン。
                "column":[// 読み取るカラム。
                    {
                        "name":"rowkey",// カラム名。
                        "type":"string"// データ型。
                    },
                    {
                        "name":"basic:age",
                        "type":"string"
                    },
                    {
                        "name":"basic:name",
                        "type":"string"
                    },
                    {
                        "name":"score:english",
                        "type":"string"
                    },
                    {
                        "name":"score:math",
                        "type":"string"
                    }
                ],
                "range":{// HBase リーダーの rowkey 範囲。
                    "endRowkey":"",// 終了 rowkey。
                    "isBinaryRowkey":true,// 構成済みの startRowkey および endRowkey を byte 配列に変換する方法を指定します。デフォルト値：false。
                    "startRowkey":""// 開始 rowkey。
                },
                "maxVersion":"",// multiVersionFixedColumn モードで読み取るバージョン数。
                "encoding":"UTF-8",// エンコード形式。
                "table":"student",// テーブル名。
                "hbaseConfig":{// HBase クラスターへの接続構成（JSON 形式）。
                    "hbase.zookeeper.quorum":"hostname",
                    "hbase.rootdir":"hdfs://ip:port/database",
                    "hbase.cluster.distributed":"true"
                }
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"odps",// 送信先プラグイン名。この例では MaxCompute を使用していますが、他のライタープラグインに置き換え可能です。
            "parameter":{
                "partition":"",// 送信先テーブルのパーティション情報。非パーティション化テーブルの場合は、このパラメーターを設定する必要はありません。
                "truncate":true,// 書き込み操作の前に、送信先テーブルまたはパーティション内の既存データをクリアするかどうかを指定します。true を指定すると、データがクリアされます。
                "datasource":"odps_datasource",// MaxCompute データソースの名前。
                "column":[// 送信先カラムのリスト。
                    "rowkey",
                    "basic_age",
                    "basic_name",
                    "score_english",
                    "score_math"
                ],
                "table":"student_target"// 送信先 MaxCompute テーブルの名前。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// 許容される最大エラーレコード数。
        },
        "speed":{
            "throttle":true,// 速度制限を有効にするかどうかを指定します。true を設定すると、`mbps` で指定されたレートでデータが転送されます。false を設定すると、`mbps` パラメーターは無視され、レート制限は適用されません。
            "concurrent":1,// 最大並行スレッド数。
            "mbps":"12"// 最大転送レート。単位：MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase リーダースクリプトの例（multiVersionFixedColumn モード）

以下の例は、multiVersionFixedColumn モードを使用して HBase からデータを読み取り、MaxCompute に書き込む完全なスクリプト構成を示しています。このモードでは、HBase の各セルの値が個別のレコードに変換されます。各レコードは、rowkey、family:qualifier、timestamp、および value の 4 つのカラムで構成されます。

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"hbase",// プラグイン名。
            "parameter":{
                "mode":"multiVersionFixedColumn",// HBase からデータを読み取るモード。この例では multiVersionFixedColumn モードを使用しています。
                "scanCacheSize":"256",// HBase クライアントが RPC ごとにサーバーから読み取る行数。
                "scanBatchSize":"100",// HBase クライアントが RPC ごとにサーバーから読み取るカラム数。
                "hbaseVersion":"20x",// HBase のバージョン。
                "datasource":"hbase_datasource",// HBase データソースの名前。
                "column":[// 読み取るカラム。最初のカラムは rowkey でなければなりません。その他のすべてのカラムの名前は "ColumnFamily:Qualifier" 形式である必要があります。
                    {
                        "name":"rowkey",// rowkey カラム。
                        "type":"string"
                    },
                    {
                        "name":"basic:age",// basic カラムファミリーの age カラム。
                        "type":"string"
                    },
                    {
                        "name":"basic:name",// basic カラムファミリーの name カラム。
                        "type":"string"
                    },
                    {
                        "name":"score:english",// score カラムファミリーの english カラム。
                        "type":"string"
                    },
                    {
                        "name":"score:math",// score カラムファミリーの math カラム。
                        "type":"string"
                    }
                ],
                "range":{
                    "isBinaryRowkey":false
                },
                "maxVersion":"-1",// すべてのバージョンのデータを読み取ります。multiVersionFixedColumn モードでは必須パラメーターです。-1 を指定すると、すべてのバージョンが読み取られます。
                "encoding":"UTF-8",// エンコード形式。
                "table":"student"// HBase テーブルの名前。
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"odps",// 送信先プラグイン名。この例では MaxCompute を使用しています。
            "parameter":{
                "partition":"",// 送信先テーブルのパーティション情報。非パーティション化テーブルの場合は、このパラメーターを設定する必要はありません。
                "truncate":true,// 書き込み操作の前に、送信先テーブルまたはパーティション内の既存データをクリアするかどうかを指定します。true を指定すると、データがクリアされます。
                "datasource":"odps_datasource",// MaxCompute データソースの名前。
                "column":[// 4 つの送信先カラムは、rowkey、family:qualifier、timestamp、value に対応します。
                    "rowkey",
                    "cf",
                    "timestamp_col",
                    "value"
                ],
                "table":"hbase_multiversion_target"// 送信先 MaxCompute テーブルの名前。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// 許容される最大エラーレコード数。
        },
        "speed":{
            "throttle":false,// 速度制限を無効化します。
            "concurrent":2// 最大並行スレッド数。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase リーダーパラメーター

HBase ライタースクリプトの例

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"hbase",// プラグイン名。
            "parameter":{
                "mode":"normal",// HBase へのデータ書き込みモード。
                "walFlag":"false",// false に設定すると、Write-Ahead Log（WAL）機能が無効になります。
                "hbaseVersion":"094x",// HBase のバージョン。
                "rowkeyColumn":[// HBase 行キーの構成に使用するソースカラムを定義します。
                    {
                        "index":"0",// ソースカラムのインデックス。
                        "type":"string"// データ型。
                    },
                    {
                        "index":"-1",// 定数値の場合は -1 を使用します。
                        "type":"string",
                        "value":"_"
                    }
                ],
                "nullMode":"skip",// null 値の処理方法を指定します。
                "column":[// HBase に書き込むカラム。
                    {
                        "name":"columnFamilyName1:columnName1",// カラム名。
                        "index":"0",// ソースカラムのインデックス。
                        "type":"string"// データ型。
                    },
                    {
                        "name":"columnFamilyName2:columnName2",
                        "index":"1",
                        "type":"string"
                    },
                    {
                        "name":"columnFamilyName3:columnName3",
                        "index":"2",
                        "type":"string"
                    }
                ],
                "encoding":"UTF-8",// エンコード形式。
                "table":"YOUR_TABLE_NAME",// テーブル名。
                "hbaseConfig":{// HBase クラスターへの接続構成（JSON 形式）。
                    "hbase.zookeeper.quorum":"hostname",
                    "hbase.rootdir":"hdfs://ip:port/database",
                    "hbase.cluster.distributed":"true"
                }
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// 許容される最大エラーレコード数。
        },
        "speed":{
            "throttle":true,// 速度制限を有効にするかどうかを指定します。true を設定すると、`mbps` で指定されたレートでデータが転送されます。false を設定すると、`mbps` パラメーターは無視され、レート制限は適用されません。
            "concurrent":1, // 最大並行スレッド数。
            "mbps":"12"// 最大転送レート。単位：MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase ライターパラメーター

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

HBase20xsql リーダースクリプトの例

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"hbase20xsql",// プラグイン名。
            "parameter":{
                "queryServerAddress": "http://127.0.0.1:8765",  // Phoenix QueryServer のエンドポイント。
                "serialization": "PROTOBUF",  // QueryServer のシリアル化形式。
                "table": "TEST",    // 読み取るテーブルの名前。
                "column": ["ID", "NAME"],   // 読み取るカラムの名前。
                "splitKey": "ID"    // スプリットキー。これはテーブルのプライマリキーである必要があります。
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"// 許容される最大エラーレコード数。
        },
        "speed":{
            "throttle":true,// 速度制限を有効にするかどうかを指定します。true を設定すると、`mbps` で指定されたレートでデータが転送されます。false を設定すると、`mbps` パラメーターは無視され、レート制限は適用されません。
            "concurrent":1,// 最大並行スレッド数。
            "mbps":"12"// 最大転送レート。単位：MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase20xsql リーダーパラメーター

HBase11xsql ライタースクリプトの例

{
  "type": "job",
  "version": "1.0",
  "configuration": {
    "setting": {
      "errorLimit": {
        "record": "0" // 許容される最大エラーレコード数。
      },
      "speed": {
            "throttle":true, // 速度制限を有効にするかどうかを指定します。
            "concurrent":1, // 最大並行スレッド数。
            "mbps":"1" // 最大転送レート。単位：MB/s。
      }
    },
    "reader": {
      "plugin": "odps",
      "parameter": {
        "datasource": "your_odps_datasource_name",
        "table": "your_source_table_name",
        "column": [],
        "partition": ""
      }
    },
    "writer": {
        "plugin": "hbase11xsql",
        "parameter": {
          "table": "PHOENIX_TABLE_NAME", // 送信先 Phoenix テーブルの名前。
          "hbaseConfig": {
            "hbase.zookeeper.quorum": "zookeeper-ip1,zookeeper-ip2", // 送信先 HBase クラスターの ZooKeeper エンドポイント。
            "zookeeper.znode.parent": "/hbase" // 送信先 HBase クラスターの znode。
          },
          "column": [
            "ID",
            "NAME" // 送信先カラム名のリスト。
          ],
          "batchSize": 256, // バッチ書き込みの最大行数。
          "nullMode": "skip" // null 値の処理方法を指定します。
        }
    }
  }
}

HBase11xsql ライターパラメーター