E-MapReduce:E-MapReduce data migration solution

Migration scenarios

EMR supports the following migration paths:

Supported incremental upstream data sources for Hadoop Distributed File System (HDFS) include ApsaraDB RDS incremental data and Apache Flume.

Prerequisites

Before you begin, make sure that you have:

• Network connectivity between the source and target clusters (see Establish network connectivity)

• The NameNode IP address of the source cluster (use the active NameNode if HA is configured)

• The ACL configuration status of the source cluster (dfs.permissions.enabled and dfs.namenode.acls.enabled)

• The Hive version running on the source cluster

• The Hive metastore database type (MySQL) and connection credentials

Migration phases

Establish network connectivity

Choose a connectivity method based on your source cluster environment.

Migrate HDFS data

Use DistCp (Distributed Copy) to migrate full and incremental HDFS data. Run distcp after the source and target cluster networks are interconnected.

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

Migrate Hive metadata

Hive metadata is typically stored in MySQL. During migration, two adjustments are required:

• Location replacement: Update the HDFS nameservices prefix in all table and partition locations.

• Hive version alignment: Upgrade the metastore schema if the source cluster runs an earlier Hive version.

Migrate Flume streams

After HDFS data and Hive metadata are migrated, set up Flume to write to the new cluster.

Migrate jobs

If the version differences between the source and target clusters are significant (Hadoop, Hive, Spark, or MapReduce), rebuild jobs as needed.

export HADOOP_HEAPSIZE=512

export HADOOP_HEAPSIZE=1024

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

Migrate Presto configuration

If the source cluster uses Presto, update the Hive connector configuration files on the EMR cluster. For more information, see Presto Hive connector documentation.

Update the following Hive properties:

Validate data

After all migration phases are complete, verify the migration by generating and reviewing reports that compare the data in the source and target clusters.

Appendix: Hive upgrade script example (1.2 to 2.3)

The following example shows the SQL statements executed when upgrading Hive from version 1.2 to 2.3. These scripts are located at /usr/lib/hive-current/scripts/metastore/upgrade/mysql/ on the EMR cluster.

-- Step 1: Run 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);

-- Step 2: Run 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;

-- Step 3: Run 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;

-- Step 4: Run 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);