EMR で Apache Sqoop を使用して OSS-HDFS データを読み書きする方法 - OSS

E-MapReduce (EMR) クラスターで Apache Sqoop を使用して、MySQL データベースから OSS-HDFS にデータをインポートします。Sqoop は EMR に組み込まれている Hadoop-OSS コネクタを介して OSS-HDFS にアクセスします。そのため、OSS-HDFS のエンドポイント形式は標準の Object Storage Service (OSS) とは異なり、oss://bucket/path の代わりに oss://bucket.region.oss-dls.aliyuncs.com/path を使用する必要があります。

前提条件

開始する前に、以下を確認してください。

EMR V3.42.0 以降、または EMR V5.8.0 以降を実行している EMR クラスター。手順については、「クラスターの作成」をご参照ください。
バケットで OSS-HDFS が有効になっており、アクセス権限が付与されていること。手順については、「OSS-HDFS の有効化とアクセス権限の付与」をご参照ください。

EMR クラスターへの接続

EMR コンソールにログインします。左側のナビゲーションウィンドウで、EMR on ECS をクリックします。
EMR クラスターをクリックします。
[ノード] タブをクリックし、ノードグループの左側にあるプラスアイコン () をクリックします。
ECS インスタンスの ID をクリックします。インスタンスページで、インスタンス ID の横にある [接続] をクリックします。

Windows または Linux から SSH キーペアまたは SSH パスワードでログインする手順については、「クラスターへのログイン」をご参照ください。

MySQL から OSS-HDFS へのデータインポート

次のコマンドを実行して、MySQL テーブルから OSS-HDFS ディレクトリにデータをインポートします。

sudo sqoop import \
  --connect <dburi>/<dbname> \
  --username <username> \
  --password <password> \
  --table <tablename> \
  --target-dir <oss-dir> \
  -m <count>

パラメーター

パラメーター	必須	説明
`dburi`	はい	データベースの JDBC URI。例：`jdbc:mysql://192.168.xxx.xxx:3306/`
`dbname`	はい	データベースの名前。
`username`	はい	データベースにログインするためのユーザー名。
`password`	はい	ユーザーのパスワード。
`tablename`	はい	インポートする MySQL テーブルの名前。
`oss-dir`	はい	OSS-HDFS のターゲットディレクトリ。OSS-HDFS エンドポイント形式を使用します：`oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/dir/`
`count`	いいえ	MapReduce タスクの数。
`format`	いいえ	出力ファイル形式。有効な値：`avrodatafile`、`sequencefile`、`textfile` (デフォルト)、`parquetfile`。`--as-<format>` を使用して指定します。例：`--as-parquetfile`。

OSS-HDFS エンドポイント形式 (oss://bucket.region.oss-dls.aliyuncs.com/path) は、標準の OSS とは異なります。必ず OSS-HDFS 固有のエンドポイントを使用してください。

例

次のコマンドは、sqoop_test データベースの src_kv テーブルを Parquet 形式で OSS-HDFS にインポートします。

sudo sqoop import \
  --connect jdbc:mysql://master-1-1/sqoop_test \
  --username root \
  --password password1 \
  --table src_kv \
  -m 1 \
  --target-dir oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_kv \
  --as-parquetfile

データの増分インポート

増分インポートの場合は、コマンドに --incremental、--check-column、--last-value を追加します。append と lastmodified の 2 つのモードがサポートされています。

Append モード

append モードは、テーブルが新しい行のみ (挿入のみ、更新なし) を受け取る場合に使用します。Sqoop は、チェック列の値が --last-value より大きい行をインポートします。

append モードを使用する場合、--temporary-rootdir が必要です。Sqoop はまず一時ディレクトリにデータを書き込み、その後ファイルをターゲットディレクトリに移動します。ターゲットディレクトリが既に存在する場合、Sqoop はその中のデータをインポートしたり上書きしたりしません。

sudo sqoop import \
  --connect <dburi>/<dbname> \
  --username <username> \
  --password <password> \
  --table <tablename> \
  --target-dir <oss-dir> \
  --temporary-rootdir <oss-tmpdir> \
  --incremental append \
  --check-column <col> \
  --last-value <value> \
  -m <count>

増分パラメーター

パラメーター	必須	説明
`oss-tmpdir`	はい (増分モード)	一時的な OSS-HDFS ディレクトリ。例：`oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_tmp/`
`col`	はい	新しい行を識別するために使用される列。
`value`	はい	前回のインポートからのチェック列の最大値。Sqoop は、より高い値を持つ行のみをインポートします。

例

sudo sqoop import \
  --connect jdbc:mysql://master-1-1/sqoop_test \
  --username root \
  --password password1 \
  --table src_kv \
  --target-dir oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_kv \
  --temporary-rootdir oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_tmp \
  --incremental append \
  --check-column id \
  --last-value 100 \
  -m 1

Lastmodified モード

lastmodified モードは、テーブルが挿入と更新の両方を受け取る場合に使用します。Sqoop は、チェック列 (タイムスタンプ) が --last-value よりも新しい行をインポートします。このモードでは、更新された行を既存のデータとマージするために --merge-key が必要です。

lastmodified モードを使用する場合、--temporary-rootdir が必要です。

sudo sqoop import \
  --connect <dburi>/<dbname> \
  --username <username> \
  --password <password> \
  --table <tablename> \
  --target-dir <oss-dir> \
  --temporary-rootdir <oss-tmpdir> \
  --incremental lastmodified \
  --check-column <col> \
  --merge-key <key-col> \
  --last-value <value> \
  -m <count>

例

sudo sqoop import \
  --connect jdbc:mysql://master-1-1/sqoop_test \
  --username root \
  --password password1 \
  --table src_kv \
  --target-dir oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_kv \
  --temporary-rootdir oss://examplebucket.cn-hangzhou.oss-dls.aliyuncs.com/tmp/sqoop_tmp \
  --incremental lastmodified \
  --check-column updated_at \
  --merge-key id \
  --last-value "2024-01-01 00:00:00" \
  -m 1