Sqoop Imports from Oracle

Sqoop is a tool designed to transfer data between Hadoop and relational databases. You can use Sqoop to import data from a relational database management system (RDBMS) such as MySQL or Oracle into the Hadoop Distributed File System (HDFS), transform the data in Hadoop MapReduce, and then export the data back into an RDBMS.

JDBC ORACLE: Examples for Import:

sqoop-import --connect jdbc:oracle:thin:@db.test.com:PORT:INSTANCE_NAME --table DW_DATAMART.HCM_EMPLOYEE_D --fields-terminated-by '\t' --lines-terminated-by '\n' --username RAHUL -P


sqoop-import --connect jdbc:oracle:thin:@db.test.com:PORT:INSTANCE_NAME --table DW_DATAMART.HCM_EMPLOYEE_D --fields-terminated-by '\t' --lines-terminated-by '\n' --username RAHUL -P

JDBC ORACLE: Example for Select:

The sqoop-eval tool allows users to quickly run simple SQL queries against a database; results are printed to the console. This allows users to preview their import queries to ensure they import the data they expect.



sqoop-eval --connect jdbc:oracle:thin:@db.test.com:PORT:INSTANCE_NAME --select * from DW_DATAMART.HCM_COMPANY_D

Sqoop Import to HBASE table:

sqoop-import --connect jdbc:oracle:thin:@db.test.com:PORT:INSTANCE_NAME --username RAHUL --P --table DW_DATAMART.PAY_PAY_CHK_OPTION_D --hbase-table DW_DATAMART.PAY_PAY_CHK_OPTION_D --column-family cf1 --hbase-create-table

If no primary key defined in the Oracle table

sqoop-import --connect jdbc:oracle:thin:@db.test.com:1725:hrlites --username RAHUL --P --table PSMERCHANTID --hbase-table PSMERCHANTID --column-family cf --hbase-row-key MERCHANTID --hbase-create-table --split-by MERCHANTID

sqoop-import --connect jdbc:oracle:thin:@db.test.com:PORT:INSTANCE_NAME --username RAHUL --P --table DW_DATAMART.PAY_PAYGROUP_D --hbase-table DW_DATAMART.PAY_PAYGROUP_D --column-family cf1 --hbase-create-table


sqoop-import --connect jdbc:oracle:thin:@db.test.com:1725:hrlites --username RAHUL --P --table PSMERCHANTID --hbase-table PSMERCHANTID --column-family cf --hbase-create-table --split-by MERCHANTID

DistCp (Distributed Copy)

Hadoop DistCp can be used to copy data between Hadoop clusters (and also within a Hadoop cluster). DistCp uses MapReduce to implement its distribution, error handling and reporting. It expands a list of files and directories into map tasks, each of which copies a partition of the files specified in the source list.

Common use of DistCp is an inter-cluster copy:

hadoop distcp hdfs://nn1:8020/source hdfs://nn2:8020/destination 

Specify multiple source directories:

hadoop distcp hdfs://nn1:8020/source/a hdfs://nn1:8020/source/b hdfs:// nn2:8020/destination

Specify multiple source directories from a file with the -f option:

hadoop distcp -f hdfs://nn1:8020/srclist hdfs://nn2:8020/destination

Note: srclist contains:
hdfs://nn1:8020/source/a
hdfs://nn1:8020/source/b

DistCp from HDP-1.3.x to HDP-2.x:

hadoop distcp hftp://<hdp 1.3.x namenode host>:50070/<folder path of source> hdfs://<hdp 2.x namenode host>/<folder path of target>

DistCp copy from HDP 1.3.0 to HDP-2.0:

hadoop distcp hftp://namenodehdp130.test.com:50070/apps/hive/warehouse/db/ hdfs://namenodehdp20.test.com/data/raw/

Update & Overwrite:

• The DistCp -update option is used to copy files from a source that do not exist at the target, or that have different contents. 
• The DistCp -overwrite option overwrites target files even if they exist at the source, or if they have the same contents.

Daily Activities of Hadoop Admin

1. Working with data delivery teams to setup new Hadoop users. This job includes setting up Linux users, setting up Kerberos principals and testing HDFS, Hive, Pig and MapReduce access for the new users.

2. Cluster maintenance as well as creation and removal of nodes using tools like Ambari and other tools.

3. Screen Hadoop cluster job performances and capacity planning

4. Monitor Hadoop cluster connectivity and security

5. Manage and review Hadoop log files.

6. File system management and monitoring.

7. Collaborating with application teams to install operating system and Hadoop updates, patches, version upgrades when required.

8. Performance tuning of Hadoop clusters and Hadoop MapReduce routines.

9. HDFS support and maintenance.

10. Disk space management.

11. Working on Snapshot.

12. Transfering data using DISTCP.

Resource Manager High Availability

This blog provides instructions on setting up the ResourceManager (RM) High Availability (HA) feature in a HDFS cluster. The Active and Standby ResourceManagers embed the Zookeeper-based ActiveStandbyElector to determine which ResourceManager should be active.

Note: This guide assumes that an existing HDP cluster has been manually installed and deployed. It provides instructions on how to manually enable ResourceManager HA on top of the existing cluster.

The ResourceManager was a single point of failure (SPOF) in an HDFS cluster. Each cluster had a single ResourceManager, and if that machine or process became unavailable, the entire cluster would be unavailable until the ResourceManager was either restarted or started on a separate machine. This situation impacted the total availability of the HDFS cluster in two major ways:

• In the case of an unplanned event such as a machine crash, the cluster would be unavailable until an operator restarted the ResourceManager.

• Planned maintenance events such as software or hardware upgrades on the ResourceManager machine would result in windows of cluster downtime.

The ResourceManager HA feature addresses these problems. This feature enables you to run redundant ResourceManagers in the same cluster in an Active/Passive configuration with a hot standby. This mechanism thus facilitates either a fast failover to the standby ResourceManager during machine crash or a graceful administrator-initiated failover during planned maintenance.

Hardware Resources

Ensure that you prepare the following hardware resources:

• ResourceManager machines: The machines where you run Active and Standby ResourceManagers should have exactly the same hardware. For recommended hardware for ResourceManagers, see "Hardware for Master Nodes" in the Cluster Planning Guide.

• Zookeeper machines: For automated failover functionality, there must be an existing Zookeeper cluster available. The Zookeeper service nodes can be co-located with other Hadoop daemons.

Deploy ResourceManager HA Cluster

HA configuration is backwards-compatible and works with your existing single ResourceManager configuration.

As described in the following sections, first configure manual or automatic ResourceManager failover. Then deploy the ResourceManager HA cluster.

Configure Manual or Automatic ResourceManager Failover

Complete the following prerequisites:

• Make sure that you have a working Zookeeper service. If you had an Ambari deployed HDP cluster with Zookeeper, you can use that Zookeeper service. If not, deploy ZooKeeper using the instructions provided in the Non-Ambari Cluster Installation Guide.

Note: In a typical deployment, ZooKeeper daemons are configured to run on three or five nodes. It is, however, acceptable to co-locate the ZooKeeper nodes on the same hardware as the HDFS NameNode and Standby Node. Many operators choose to deploy the third ZooKeeper process on the same node as the YARN ResourceManager. To achieve performance and improve isolation, Hortonworks recommends configuring the ZooKeeper nodes such that the ZooKeeper data and HDFS metadata is stored on separate disk drives.

• Shut down the cluster using the instructions provided in "Controlling HDP Services Manually," in the HDP Reference Guide.

Set Common ResourceManager HA Properties

The following properties are required for both manual and automatic ResourceManager HA. Add these properties to the etc/hadoop/conf/yarn-site.xml file:

Note: You can also set values for each of the following properties in yarnsite.xml:

yarn.resourcemanager.address.<rm#id>

yarn.resourcemanager.scheduler.address.<rm#id>

yarn.resourcemanager.admin.address.<rm#id>

yarn.resourcemanager.resource#tracker.address.<rm#id>

yarn.resourcemanager.webapp.address.<rm#id>

If these addresses are not explicitly set, each of these properties will use 

yarn.resourcemanager.hostname.<rm-id>:default_port

such as DEFAULT_RM_PORT, DEFAULT_RM_SCHEDULER_PORT, etc.

The following is a sample yarn-site.xml file with these common ResourceManager

HA properties configured:

<!-- RM HA Configurations-->

<property>

<name>yarn.resourcemanager.ha.enabled</name>

<value>true</value>

</property>

<property>

<name>yarn.resourcemanager.ha.rm-ids</name>

<value>rm1,rm2</value>

</property>

<property>

<name>yarn.resourcemanager.hostname.rm1</name>

<value>${rm1 address}</value>

</property>

<property>

<name>yarn.resourcemanager.hostname.rm2</name>

<value>${rm2 address}</value>

</property>

<property>

<name>yarn.resourcemanager.webapp.address.rm1</name>

<value>rm1_web_address:port_num</value>

<description>We can set rm1_web_address separately.

If not, it will use

${yarn.resourcemanager.hostname.rm1}:DEFAULT_RM_WEBAPP_PORT

</description>

</property>

<property>

<name>yarn.resourcemanager.webapp.address.rm2</name>

<value>rm2_web_address:port_num</value>

</property>

<property>

<name>yarn.resourcemanager.recovery.enabled</name>

<value>true</value>

</property>

<property>

<name>yarn.resourcemanager.store.class</name>

<value>org.apache.hadoop.yarn.server.resourcemanager.recovery.

ZKRMStateStore</value>

</property>

<property>

<name>yarn.resourcemanager.zk-address</name>

<value>${zk1.address,zk2.address}</value>

</property>

<property>

<name>yarn.client.failover-proxy-provider</name>

<value>org.apache.hadoop.yarn.client.

ConfiguredRMFailoverProxyProvider</value>

</property>

Configure Manual ResourceManager Failover

Automatic ResourceManager failover is enabled by default, so it must be disabled for manual failover.

To configure manual failover for ResourceManager HA, add the yarn.resourcemanager.ha.automatic-failover.enabled configuration property to the etc/hadoop/conf/yarn-site.xml file, and set its value to "false":

<property>

<name>yarn.resourcemanager.ha.automatic-failover.enabled</name>

<value>false</value>

</property>

Configure Automatic ResourceManager Failover

The preceding section described how to configure manual failover. In that mode, the system will not automatically trigger a failover from the active to the standby ResourceManager, even if the active node has failed. This section describes how to configure automatic failover.

1. Add the following configuration options to the yarn-site.xml file:

Example:

<property>

<name>yarn.resourcemanager.ha.automatic-failover.zk-base-path</name>

<value>/yarn-leader-election</value>

<description>Optional setting. The default value is

/yarn-leader-election</description>

</property>

<property>

<name>yarn.resourcemanager.cluster-id</name>

<value>yarn-cluster</value>

</property>

2. Automatic ResourceManager failover is enabled by default.

If you previously configured manual ResourceManager failover by setting the value of

yarn.resourcemanager.ha.automatic-failover.enabled to "false", you must

delete this property to return automatic failover to its default enabled state.

Deploy the ResourceManager HA Cluster

1. Copy the etc/hadoop/conf/yarn-site.xml file from the primary ResourceManager host to the standby ResourceManager host.

2. Make sure that the clientPort value set in etc/zookeeper/conf/zoo.cfg matches the port set in the following yarn-site.xml property:

<property>

<name>yarn.resourcemanager.zk-state-store.address</name>

<value>localhost:2181</value>

</property>

3. Start ZooKeeper. Execute this command on the ZooKeeper host machine(s):

su - zookeeper -c "export ZOOCFGDIR=/usr/hdp/current/zookeeper-server/

conf ; export ZOOCFG=zoo.cfg; source /usr/hdp/current/zookeeper-server/conf/

zookeeper-env.sh ; /usr/hdp/current/zookeeper-server/bin/zkServer.sh start"

4. Start HDFS using the instructions provided in "Controlling HDP Services Manually".

5. Start YARN using the instructions provided in "Controlling HDP Services Manually".

6. Set the active ResourceManager:

MANUAL FAILOVER ONLY: If you configured manual ResourceManager failover, you

must transition one of the ResourceManagers to Active mode. Execute the following CLI

command to transition ResourceManager "rm1" to Active:

yarn rmadmin -transitionToActive rm1

You can use the following CLI command to transition ResourceManager "rm1" to Standby mode:

yarn rmadmin -transitionToStandby rm1

AUTOMATIC FAILOVER: If you configured automatic ResourceManager failover, no

action is required -- the Active ResourceManager will be chosen automatically.

7. Start all remaining unstarted cluster services using the instructions provided in

"Controlling HDP Services Manually," in the HDP Reference Guide.

Minimum Settings for Automatic ResourceManager

HA Configuration

The minimum yarn-site.xml configuration settings for ResourceManager HA with automatic failover are as follows:

<property>

<name>yarn.resourcemanager.ha.enabled</name>

<value>true</value>

</property>

<property>

<name>yarn.resourcemanager.ha.rm-ids</name>

<value>rm1,rm2</value>

</property>

<property>

<name>yarn.resourcemanager.hostname.rm1</name>

<value>192.168.1.9</value>

</property>

<property>

<name>yarn.resourcemanager.hostname.rm2</name>

<value>192.168.1.10</value>

</property>

<property>

<name>yarn.resourcemanager.recovery.enabled</name>

<value>true</value>

</property>

<property>

<name>yarn.resourcemanager.store.class</name>

<value>org.apache.hadoop.yarn.server.resourcemanager.recovery.

ZKRMStateStore</value>

</property>

<property>

<name>yarn.resourcemanager.zk-address</name>

<value>192.168.1.9:2181,192.168.1.10:2181</value>

<description>For multiple zk services, separate them with comma</description>

</property>

<property>

<name>yarn.resourcemanager.cluster-id</name>

<value>yarn-cluster</value>

</property>

Testing ResourceManager HA on a Single Node

If you would like to test ResourceManager HA on a single node (launch more than one

ResourceManager on a single node), you need to add the following settings in yarnsite.

xml.

To enable ResourceManager "rm1" to launch:

<property>

<name>yarn.resourcemanager.ha.id</name>

<value>rm1</value>

<description>If we want to launch more than one RM in single node, we need

this configuration</description>

</property>

To enable ResourceManager rm2 to launch:

<property>

<name>yarn.resourcemanager.ha.id</name>

<value>rm2</value>

<description>If we want to launch more than one RM in single node, we need

this configuration</description>

</property>

You should also explicitly set values specific to each ResourceManager for the following properties separately in yarn-site.xml:

• yarn.resourcemanager.address.<rm-id>

• yarn.resourcemanager.scheduler.address.<rm-id>

• yarn.resourcemanager.admin.address.<rm-id>

• yarn.resourcemanager.resource#tracker.address.<rm-id>

• yarn.resourcemanager.webapp.address.<rm-id>

For example:

<!-- RM1 Configs -->

<property>

<name>yarn.resourcemanager.address.rm1</name>

<value>localhost:23140</value>

</property>

<property>

<name>yarn.resourcemanager.scheduler.address.rm1</name>

<value>localhost:23130</value>

</property>

<property>

<name>yarn.resourcemanager.webapp.address.rm1</name>

<value>localhost:23188</value>

</property>

<property>

<name>yarn.resourcemanager.resource-tracker.address.rm1</name>

<value>localhost:23125</value>

</property>

<property>

<name>yarn.resourcemanager.admin.address.rm1</name>

<value>localhost:23141</value>

</property>

<!-- RM2 configs -->

<property>

<name>yarn.resourcemanager.address.rm2</name>

<value>localhost:33140</value>

</property>

<property>

<name>yarn.resourcemanager.scheduler.address.rm2</name>

<value>localhost:33130</value>

</property>

<property>

<name>yarn.resourcemanager.webapp.address.rm2</name>

<value>localhost:33188</value>

</property>

<property>

<name>yarn.resourcemanager.resource-tracker.address.rm2</name>

<value>localhost:33125</value>

</property>

<property>

<name>yarn.resourcemanager.admin.address.rm2</name>

<value>localhost:33141</value>

</property>

HiveServer2 Architecture

HiveServer2 (HS2) is a service that enables clients to execute queries against Hive. HiveServer2 is the successor to HiveServer1 which has been deprecated. HS2 supports multi-client concurrency and authentication. It is designed to provide better support for open API clients like JDBC and ODBC.

HS2 is a single process running as a composite service, which includes the Thrift-based Hive service (TCP or HTTP) and a Jetty web server for web UI.

HS2 Architecture

The Thrift-based Hive service is the core of HS2 and responsible for servicing the Hive queries (e.g., from Beeline). Thrift is an RPC framework for building cross-platform services. Its stack consists of 4 layers: Server, Transport, Protocol, and Processor. You can find more details about the layers at

The usage of those layers in the HS2 implementation is described below.

Server

HS2 uses a TThreadPoolServer (from Thrift) for TCP mode, or a Jetty server for the HTTP mode.

The TThreadPoolServer allocates one worker thread per TCP connection. Each thread is always associated with a connection even if the connection is idle. So there is a potential performance issue resulting from a large number of threads due to a large number of concurrent connections. In the future, HS2 might switch to another server type for TCP mode, for example, TThreadedSelectorServer. Here is an article about a performance comparison between different Thrift Java servers.

Transport

HTTP mode is required when a proxy is needed between the client and server (for example, for load balancing or security reasons). That is why it is supported, as well as TCP mode. You can specify the transport mode of the Thrift service through the Hive configuration property hive.server2.transport.mode.

Protocol

The Protocol implementation is responsible for serialization and deserialization. HS2 is currently using TBinaryProtocol as its Thrift protocol for serialization. In the future, other protocols may be considered, such as TCompactProtocol, based on more performance evaluation.

Processor

Process implementation is the application logic to handle requests. For example, the ThriftCLIService.ExecuteStatement() method implements the logic to compile and execute a Hive query.

Dependencies of HS2

Metastore

The metastore can be configured as embedded (in the same process as HS2) or as a remote server (which is a Thrift-based service as well). HS2 talks to the metastore for the metadata required for query compilation. 

Hadoop cluster

HS2 prepares physical execution plans for various execution engines (MapReduce/Tez/Spark) and submits jobs to the Hadoop cluster for execution.

JDBC Client

The JDBC driver is recommended for the client side to interact with HS2. Note that there are some use cases (e.g., Hadoop Hue) where the Thrift client is used directly and JDBC is bypassed.

Here is a sequence of API calls involved to make the first query:

• The JDBC client (e.g., Beeline) creates a HiveConnection by initiating a transport connection (e.g., TCP connection) followed by an OpenSession API call to get a SessionHandle. The session is created from the server side.
• The HiveStatement is executed (following JDBC standards) and an ExecuteStatement API call is made from the Thrift client. In the API call, SessionHandle information is passed to the server along with the query information.
• The HS2 server receives the request and asks the driver (which is a CommandProcessor) for query parsing and compilation. The driver kicks off a background job that will talk to Hadoop and then immediately returns a response to the client. This is an asynchronous design of the ExecuteStatement API. The response contains an OperationHandle created from the server side.
• The client uses the OperationHandle to talk to HS2 to poll the status of the query execution.

YARN Application Execution

YARN application execution consists of the following steps:

• Application submission.
• Bootstrapping the Application Master instance for the application.
• Application execution managed by the Application Master instance.

Let’s walk through an application execution sequence (steps are illustrated in the diagram):

1. A client program submits the application, including the necessary specifications to launch the application-specific Application Master.
2. The Resource Manager assumes responsibility for negotiating a specified Container in which to start the Application Master, and then launches the Application Master.
3. On boot-up, the Application Master registers with the Resource Manager. The registration allows the client program to query the Resource Manager for details, which allows it to directly communicate with its own Application Master.
4. During normal operation, the Application Master negotiates appropriate resource Containers via the resource-request protocol.
5. Upon successful Container allocations, the Application Master launches the Container by providing the Container launch specification to the Node Manager. The launch specification typically includes the necessary information to allow the Container to communicate with the Application Master.
6. The application code executing within the Container then provides necessary information (progress, status etc.) to its Application Master via an application-specific protocol.
7. During the application execution, the client that submitted the program communicates directly with the Application Master to get status, progress updates, etc. via an application-specific protocol.
8. Once the application is complete, the Application Master de-registers with the Resource Manager and shuts down, allowing its own Container to be repurposed.

Snapshots

HDFS Snapshots are read-only point-in-time copies of the file system. Snapshots can be taken on a subtree of the file system or the entire file system. Some common use cases of snapshots are data backup, protection against user errors and disaster recovery.

The implementation of HDFS Snapshots is efficient:

• Snapshot creation is instantaneous: the cost is O(1) excluding the inode lookup time.
• Additional memory is used only when modifications are made relative to a snapshot: memory usage is O(M), where M is the number of modified files/directories.
• Blocks in datanodes are not copied: the snapshot files record the block list and the file size. There is no data copying.
• Snapshots do not adversely affect regular HDFS operations: modifications are recorded in reverse chronological order so that the current data can be accessed directly. The snapshot data is computed by subtracting the modifications from the current data.

Snapshottable Directories

Snapshots can be taken on any directory once the directory has been set as snapshottable. A snapshottable directory is able to accommodate 65,536 simultaneous snapshots. There is no limit on the number of snapshottable directories. Administrators may set any directory to be snapshottable. If there are snapshots in a snapshottable directory, the directory can be neither deleted nor renamed before all the snapshots are deleted.

Nested snapshottable directories are currently not allowed. In other words, a directory cannot be set to snapshottable if one of its ancestors/descendants is a snapshottable directory.

Snapshot Paths

For a snapshottable directory, the path component “.snapshot” is used for accessing its snapshots. Suppose /foo is a snapshottable directory, /foo/bar is a file/directory in /foo, and /foo has a snapshot s0. Then, the path /foo/.snapshot/s0/bar refers to the snapshot copy of /foo/bar. The usual API and CLI can work with the “.snapshot” paths. The following are some examples.

Allow Snapshots

Allowing snapshots of a directory to be created. If the operation completes successfully, the directory becomes snapshottable.

Command:
hdfs dfsadmin -allowSnapshot <path>

Arguments:
path: The path of the snapshottable directory.

Disallow Snapshots

Disallowing snapshots of a directory to be created. All snapshots of the directory must be deleted before disallowing snapshots.

Command:
hdfs dfsadmin -disallowSnapshot <path>

Arguments:
path: The path of the snapshottable directory.

Create Snapshots

Create a snapshot of a snapshottable directory. This operation requires owner privilege of the snapshottable directory.

Command:
hdfs dfs -createSnapshot <path> [<snapshotName>]

Arguments:
path: The path of the snapshottable directory.
snapshotName: The snapshot name, which is an optional argument. When it is omitted, a default name is generated using a timestamp with the format "'s'yyyyMMdd-HHmmss.SSS", e.g. "s20130412-151029.033".

Delete Snapshots

Delete a snapshot of from a snapshottable directory. This operation requires owner privilege of the snapshottable directory.

Command:
hdfs dfs -deleteSnapshot <path> <snapshotName>

Arguments:
path: The path of the snapshottable directory.
snapshotName: The snapshot name.

Rename Snapshots

Rename a snapshot. This operation requires owner privilege of the snapshottable directory.

Command:
hdfs dfs -renameSnapshot <path> <oldName> <newName>

Arguments:
path: The path of the snapshottable directory.
oldName: The old snapshot name.
newName: The new snapshot name.

Get Snapshottable Directory Listing

Get all the snapshottable directories where the current user has permission to take snapshtos.

Command:
hdfs lsSnapshottableDir

Get Snapshots Difference Report

Get the differences between two snapshots. This operation requires read access privilege for all files/directories in both snapshots.

Command:
hdfs snapshotDiff <path> <fromSnapshot> <toSnapshot>