DataWorks:HBase データソース

HBase Reader

HBase Reader は normal モードおよび multiVersionFixedColumn モードをサポートします。

• normal モード: HBase テーブルを標準的な 2 次元テーブルとして扱い、最新バージョンのデータを取得します。

  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

  以下の表に、出力結果を示します。

  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 モード: HBase テーブルを縦方向のテーブルとして扱います。各レコードは rowKey、family:qualifier、timestamp、および value の 4 つの列で構成されます。読み取る列を明示的に指定する必要があります。このモードでは、各セルの値を個別のレコードとして扱います。セルに複数のバージョンがある場合、各バージョンに対して個別のレコードが生成されます。

  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

  以下の表に、出力結果を示します。

  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

• HBase Writer は、ソースの複数の列を連結することで rowKey を生成できます。

• HBase Writer は、以下の方法でデータの version（タイムスタンプ）を設定できます：

  • 現在時刻を使用する。

  • ソース列の値を使用する。

  • ユーザーが指定した時刻を使用する。

バッチ読み取り

• 以下の表に、HBase Reader のデータ型マッピングを示します。

  型	Data Integration の列型	データベースのデータ型
  整数	long	short、int、および long
  浮動小数点数	double	float および double
  文字列	string	binary_string および string
  日付および時刻	date	date
  バイト	bytes	bytes
  ブール値	boolean	boolean

• HBase20xsql Reader は、Phoenix のデータ型のうち、ほとんどの型をサポートしていますが、すべての型をサポートしているわけではありません。使用前に、対象のデータ型がサポートされていることを確認してください。

• 以下の表に、HBase20xsql Reader が Phoenix のデータ型に対して使用する型マッピングを示します。

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

バッチ書き込み

以下の表に、HBase Writer のデータ型マッピングを示します。

単一テーブルのオフライン同期タスク

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

  デフォルトでは、HBase はスキーマレスのデータソースであるため、ウィザードモードでは [フィールドマッピング] セクションが表示されません。そのため、フィールドマッピングを手動で設定する必要があります：

  • HBase をデータソースとして使用する場合、ソースフィールド を次の形式で設定します：data_type|column_family:column_name。

  • HBase をデータ宛先として使用する場合、宛先フィールド および rowkey の両方を設定します。宛先フィールド には、source_field_index|data_type|column_family:column_name の形式を使用します。rowkey には、source_primary_key_index|data_type の形式を使用します。

  説明

  各フィールドは別行に記述する必要があります。

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

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

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

HBase Reader スクリプトのデモ

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"hbase",// プラグイン名。
            "parameter":{
                "mode":"normal",// HBase からデータを読み取るモード。有効な値：normal および multiVersionFixedColumn。
                "scanCacheSize":"256",// RPC ごとにクライアントがサーバーから読み取る行数。
                "scanBatchSize":"100",// RPC ごとにクライアントがサーバーから読み取る列数。
                "hbaseVersion":"094x/11x",// HBase のバージョン。
                "column":[// 読み取る列。
                    {
                        "name":"rowkey",// 列名。
                        "type":"string"// データ型。
                    },
                    {
                        "name":"columnFamilyName1:columnName1",
                        "type":"string"
                    },
                    {
                        "name":"columnFamilyName2:columnName2",
                        "format":"yyyy-MM-dd",
                        "type":"date"
                    },
                    {
                        "name":"columnFamilyName3:columnName3",
                        "type":"long"
                    }
                ],
                "range":{// HBase Reader の rowkey 範囲。
                    "endRowkey":"",// 終了 rowkey。
                    "isBinaryRowkey":true,// startRowkey および endRowkey をバイト配列に変換する方法を指定します。デフォルト値は false です。
                    "startRowkey":""// 開始 rowkey。
                },
                "maxVersion":"",// マルチバージョンモードで読み取るバージョン数。
                "encoding":"UTF-8",// エンコード形式。
                "table":"",// テーブル名。
                "hbaseConfig":{// HBase クラスターへの接続構成（JSON 形式）。
                    "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"// 許容されるエラー レコードの最大数。
        },
        "speed":{
            "throttle":true,// 速度制限を有効化します。true の場合、mbps 値に基づいて速度制限が有効化されます。false の場合、mbps パラメーターは無視されます。
            "concurrent":1,// ジョブの同時実行タスク数。
            "mbps":"12"// 速度制限率。この例では、1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase Reader のパラメーター

HBase Writer スクリプトのデモ

{
    "type":"job",
    "version":"2.0",// バージョン番号。
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"hbase",// プラグイン名。
            "parameter":{
                "mode":"normal",// HBase へのデータ書き込みモード。
                "walFlag":"false",// Write-Ahead Log（WAL）への書き込みを有効化するかどうかを指定します。false を指定すると無効化されます。
                "hbaseVersion":"094x",// HBase のバージョン。
                "rowkeyColumn":[// rowkey の生成に使用する列。
                    {
                        "index":"0",// シリアル番号。
                        "type":"string"// データ型。
                    },
                    {
                        "index":"-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":"",// テーブル名。
                "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"// 速度制限率。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase Writer のパラメーター

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

HBase20xsql Reader スクリプトのデモ

{
    "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"// 速度制限率。この例では、1 mbps は 1 MB/s に相当します。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

HBase20xsql Reader のパラメーター

HBase11xsql Writer スクリプトのデモ

{
  "type": "job",
  "version": "1.0",
  "configuration": {
    "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
            "throttle":true,// 速度制限を有効化します。true の場合、mbps 値に基づいて速度制限が有効化されます。false の場合、mbps パラメーターは無視されます。
            "concurrent":1, // ジョブの同時実行タスク数。
            "mbps":"1"// 速度制限率。この例では、1 mbps は 1 MB/s に相当します。
      }
    },
    "reader": {
      "plugin": "odps",
      "parameter": {
        "datasource": "",
        "table": "",
        "column": [],
        "partition": ""
      }
    },
    "plugin": "hbase11xsql",
    "parameter": {
      "table": "ターゲット HBase テーブル名（大文字小文字を区別）。",
      "hbaseConfig": {
        "hbase.zookeeper.quorum": "ターゲット HBase クラスターの ZooKeeper サーバー アドレス。",
        "zookeeper.znode.parent": "ターゲット HBase クラスターの znode。"
      },
      "column": [
        "columnName"
      ],
      "batchSize": 256,
      "nullMode": "skip"
    }
  }
}

HBase11xsql Writer のパラメーター