E-MapReduce:E-MapReduce データ移行ソリューション

移行シナリオ

EMR は、以下の移行パスをサポートしています。

HDFS でサポートされている増分アップストリームデータソースには、ApsaraDB RDS の増分データと Apache Flume が含まれます。

前提条件

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

• ソースクラスターとターゲットクラスター間のネットワーク接続 (ネットワーク接続の確立をご参照ください)

• ソースクラスターの NameNode の IP アドレス (HA が設定されている場合はアクティブな NameNode を使用)

• ソースクラスターの ACL 設定ステータス (dfs.permissions.enabled および dfs.namenode.acls.enabled)

• ソースクラスターで実行されている Hive のバージョン

• Hive メタストアのデータベースタイプ (MySQL) と接続認証情報

移行フェーズ

ネットワーク接続の確立

ソースクラスターの環境に基づいて接続方法を選択します。

HDFS データの移行

DistCp (分散コピー) を使用して、完全および増分 HDFS データを移行します。ソースクラスターとターゲットクラスターのネットワークが相互接続された後、distcp を実行します。

hadoop distcp -pbugpcax -m 1000 -bandwidth 30 -update -delete hdfs://oldclusterip:8020/user/hive/warehouse /user/hive/warehouse

Hive メタデータの移行

Hive メタデータは通常 MySQL に保存されます。移行中、2 つの調整が必要です。

• ロケーションの置換： すべてのテーブルとパーティションのロケーションで、HDFS の nameservices プレフィックスを更新します。

• Hive バージョンの整合： ソースクラスターが古い Hive バージョンを実行している場合、メタストアスキーマをアップグレードします。

Flume ストリームの移行

HDFS データと Hive メタデータが移行された後、新しいクラスターに書き込むように Flume を設定します。

ジョブの移行

ソースクラスターとターゲットクラスターのバージョン差 (Hadoop、Hive、Spark、または MapReduce) が大きい場合は、必要に応じてジョブを再構築します。

export HADOOP_HEAPSIZE=512

export HADOOP_HEAPSIZE=1024

set mapreduce.map.java.opts=-Xmx3072m;
set mapreduce.map.memory.mb=3840;

Presto 設定の移行

ソースクラスターが Presto を使用している場合、EMR クラスター上の Hive コネクタ設定ファイルを更新します。詳細については、「Presto Hive コネクタドキュメント」をご参照ください。

次の Hive プロパティを更新します。

データの検証

すべての移行フェーズが完了したら、ソースクラスターとターゲットクラスターのデータを比較するレポートを生成してレビューし、移行を検証します。

付録：Hive アップグレードスクリプトの例 (1.2 から 2.3)

次の例は、Hive をバージョン 1.2 から 2.3 にアップグレードする際に実行される SQL ステートメントを示しています。これらのスクリプトは、EMR クラスターの /usr/lib/hive-current/scripts/metastore/upgrade/mysql/ にあります。

-- ステップ 1: upgrade-1.2.0-to-2.0.0.mysql.sql を実行
source /usr/lib/hive-current/scripts/metastore/upgrade/mysql/upgrade-1.2.0-to-2.0.0.mysql.sql

CREATE TABLE COMPACTION_QUEUE (
  CQ_ID bigint PRIMARY KEY,
  CQ_DATABASE varchar(128) NOT NULL,
  CQ_TABLE varchar(128) NOT NULL,
  CQ_PARTITION varchar(767),
  CQ_STATE char(1) NOT NULL,
  CQ_TYPE char(1) NOT NULL,
  CQ_WORKER_ID varchar(128),
  CQ_START bigint,
  CQ_RUN_AS varchar(128),
  CQ_HIGHEST_TXN_ID bigint,
  CQ_META_INFO varbinary(2048),
  CQ_HADOOP_JOB_ID varchar(32)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

CREATE TABLE TXNS (
  TXN_ID bigint PRIMARY KEY,
  TXN_STATE char(1) NOT NULL,
  TXN_STARTED bigint NOT NULL,
  TXN_LAST_HEARTBEAT bigint NOT NULL,
  TXN_USER varchar(128) NOT NULL,
  TXN_HOST varchar(128) NOT NULL,
  TXN_AGENT_INFO varchar(128),
  TXN_META_INFO varchar(128),
  TXN_HEARTBEAT_COUNT int
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

CREATE TABLE HIVE_LOCKS (
  HL_LOCK_EXT_ID bigint NOT NULL,
  HL_LOCK_INT_ID bigint NOT NULL,
  HL_TXNID bigint,
  HL_DB varchar(128) NOT NULL,
  HL_TABLE varchar(128),
  HL_PARTITION varchar(767),
  HL_LOCK_STATE char(1) NOT NULL,
  HL_LOCK_TYPE char(1) NOT NULL,
  HL_LAST_HEARTBEAT bigint NOT NULL,
  HL_ACQUIRED_AT bigint,
  HL_USER varchar(128) NOT NULL,
  HL_HOST varchar(128) NOT NULL,
  HL_HEARTBEAT_COUNT int,
  HL_AGENT_INFO varchar(128),
  HL_BLOCKEDBY_EXT_ID bigint,
  HL_BLOCKEDBY_INT_ID bigint,
  PRIMARY KEY(HL_LOCK_EXT_ID, HL_LOCK_INT_ID),
  KEY HIVE_LOCK_TXNID_INDEX (HL_TXNID)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

CREATE INDEX HL_TXNID_IDX ON HIVE_LOCKS (HL_TXNID);

-- ステップ 2: upgrade-2.0.0-to-2.1.0.mysql.sql を実行
source /usr/lib/hive-current/scripts/metastore/upgrade/mysql/upgrade-2.0.0-to-2.1.0.mysql.sql

CREATE TABLE TXN_COMPONENTS (
  TC_TXNID bigint,
  TC_DATABASE varchar(128) NOT NULL,
  TC_TABLE varchar(128),
  TC_PARTITION varchar(767),
  FOREIGN KEY (TC_TXNID) REFERENCES TXNS (TXN_ID)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

-- ステップ 3: upgrade-2.1.0-to-2.2.0.mysql.sql を実行
source /usr/lib/hive-current/scripts/metastore/upgrade/mysql/upgrade-2.1.0-to-2.2.0.mysql.sql

CREATE TABLE IF NOT EXISTS `NOTIFICATION_LOG` (
  `NL_ID` BIGINT(20) NOT NULL,
  `EVENT_ID` BIGINT(20) NOT NULL,
  `EVENT_TIME` INT(11) NOT NULL,
  `EVENT_TYPE` varchar(32) NOT NULL,
  `DB_NAME` varchar(128),
  `TBL_NAME` varchar(128),
  `MESSAGE` mediumtext,
  PRIMARY KEY (`NL_ID`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

CREATE TABLE IF NOT EXISTS `PARTITION_EVENTS` (
  `PART_NAME_ID` bigint(20) NOT NULL,
  `DB_NAME` varchar(128) CHARACTER SET latin1 COLLATE latin1_bin DEFAULT NULL,
  `EVENT_TIME` bigint(20) NOT NULL,
  `EVENT_TYPE` int(11) NOT NULL,
  `PARTITION_NAME` varchar(767) CHARACTER SET latin1 COLLATE latin1_bin DEFAULT NULL,
  `TBL_NAME` varchar(128) CHARACTER SET latin1 COLLATE latin1_bin DEFAULT NULL,
  PRIMARY KEY (`PART_NAME_ID`),
  KEY `PARTITIONEVENTINDEX` (`PARTITION_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

CREATE TABLE COMPLETED_TXN_COMPONENTS (
  CTC_TXNID bigint NOT NULL,
  CTC_DATABASE varchar(128) NOT NULL,
  CTC_TABLE varchar(128),
  CTC_PARTITION varchar(767)
) ENGINE=InnoDB DEFAULT CHARSET=latin1;

-- ステップ 4: upgrade-2.2.0-to-2.3.0.mysql.sql を実行
source /usr/lib/hive-current/scripts/metastore/upgrade/mysql/upgrade-2.2.0-to-2.3.0.mysql.sql

CREATE TABLE NEXT_TXN_ID (
  NTXN_NEXT bigint NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
INSERT INTO NEXT_TXN_ID VALUES(1);

CREATE TABLE NEXT_LOCK_ID (
  NL_NEXT bigint NOT NULL
) ENGINE=InnoDB DEFAULT CHARSET=latin1;
INSERT INTO NEXT_LOCK_ID VALUES(1);