Hadoop User Guide

NOTICE: This page refers to the Peel Hadoop Cluster which was retired in 2023 and is in the process of being archived. Its contents should not be referenced for Dataproc Courses.

What is Hadoop?

Hadoop is an open-source software framework for storing and processing big data in a distributed/parallel fashion on large clusters of commodity hardware. Essentially, it accomplishes two tasks: massive data storage and faster processing. The core Hadoop consists of HDFS - the Hadoop File System - and Hadoop implementation of MapReduce.

What is MapReduce?

MapReduce is a programming model and an associated implementation for processing and generating large data sets with a parallel, distributed algorithm on a cluster.

A MapReduce job splits a large data set into independent chunks and organizes them into key-value pairs for parallel processing. The mapping and reducing functions receive not just values, but (key, value) pairs. 

Every MapReduce job consists of at least two parts:

• The Mapper

• The Reducer

Mapping Phase: Takes input as <key,value> pairs, processes them, and produces another set of intermediate <key,value> pairs as output.

Reducing Phase: Reducing lets you aggregate values together. A reducer function receives an iterator of input values from an input list. It then combines these values together, returning a single output value.

MapReduce Data Flow for WordCount Problem

To Trigger the Job 

Where <jarfilename>.jar is the jar file from Mapper, Reducer and DriverClass.

Accessing the Peel Hadoop Cluster

All active HPC users have an account on Peel. If you need an account, review the HPC Getting and Renewing Accounts page for instructions on how to get an account. 

The Peel login nodes can be reached directly using the following command (NYU VPN required):

For more details about logging into the Peel cluster read the Accessing HPC Systems page. 

YARN Scheduler

YARN is the resource manager and job scheduler in the Peel cluster. YARN allows you to use various data processing engines for batch, interactive, and real-time stream processing of data stored in HDFS.

Job Queues

The memory available to users’ Yarn containers is 4.45 TB in total. There is a queue named 'q1' created to accommodate users with a large memory requirement. Majority of users are in the 'default' queue which is guaranteed resources. If you want to be placed in 'q1', please contact us. To check which queue you are using, run a Spark application then go to the page All Yarn Applications, and look at the 'Queue' column for your application.

Application status and logs

Please find the list of current running apps using 'Yarn' script. Running the yarn script without any arguments prints the description for all commands.

To kill a currently running app because the submitted app started malfunctioning or in worst case scenario, it's stuck in an infinite loop. Get the app ID and then kill it as given below

To download an app logs for examination on the command line

What is Spark?

Spark is a framework for performing general data analytics on a distributed computing cluster like Hadoop. It provides in-memory computations for increased speed and data processing over mapreduce. It runs on top of existing hadoop cluster and access hadoop data store (HDFS), can also process structured data in Hive and Streaming data from HDFS, Flume, Kafka, Twitter.

Running Spark Jobs

The default deploy mode for Spark on Peel is set to 'cluster'. When running Spark jobs, please make sure Spark is in the 'cluster' deploy mode, which helps reduce load on the login nodes, which are not for computing. To get an interactive session for debugging, run the following:

  or

If you need Python 3 for Spark, use the setup:

Add the option '--conf spark.yarn.submit.waitAppCompletion=false' to your Spark command, if you want to exit from a terminal after applications submitted. 

Running PySpark with Conda Env

The section is based on a post in the Cloudera Community. You may read the article before trying it out. We modified the recipe, tailored it for Peel and tested it. And it works great. The example demonstrates the use of Conda env to transport a python environment with a PySpark application needed to be executed. This sample application uses the NLTK package with the additional requirement of making tokenizer and tagger resources available to the application as well. So it is a bit complicated than those utilizing many other packages. Users are encouraged to make own Conda env by following the steps as demonstrated here. The sample application 'spark_nltk_sample.py':

Below are the commands to run, to make the magic happening. 

# Using vi, emacs or nano editor, create the application script with content as shown previously

# At this point we should have generated these files and directories 

# Run spark-submit

Once the zip files with Conda env folded in,  we may re-use them to submit many applications requiring same packages. Additionally we can put the zip files to a HDFS directory, and tell Spark to grab them there by giving option e.g.:

Using Hive

Apache Hive is a Data Warehouse software that facilitates querying and managing large datasets residing in a distributed storage (Example: HDFS). Hive provides a mechanism to project structure onto this data and query the data using a SQL-like language called HiveQL. The Hive Query Language (HiveQL or HQL) for MapReduce processes structured data using Hive. 

We configure Sentry so that by default a regular user can see only your own database which is pre-created. We can grant permission to your project members' database upon request. 

Using the Impala Shell

Apache Impala is and open source parallel processing SQL query engine running on all compute nodes (hc[01-18].nyu.cluster).

To connect to Impala use hostname where impala daemon is running. In fact, impala daemons are running on all computing nodes on Peel 'hc[01-18].nyu.cluster'. So, connecting to any compute nodes, impala will work.

Here is the process to connect to Impala-shell.

Zeppelin

Apache Zeppelin is a web-based interactive computational environment that could use Apache Spark as a backend. In some sense it is like the IPython Notebook. Zeppelin is installed on Peel. 

Peel is a YARN cluster, not a standalone Spark cluster.  Here is the process to work with Zeppelin on Peel.

See [users] section in $HOME/zeppelin/conf/shiro.ini for the default user/password and the instruction to change password. The procedures to start and stop a Zeppelin server on Peel login nodes are:

# please remember to clean up when you are done, not to leave outdated processes hanging around

There are two Peel login nodes. Verify which login node the Zeppelin server is just started by using command 'hostname' at command line. Suppose that is 'hlog-1.hpc.nyu.edu', and NYU VPN is established. Open a new terminal (or PuTTY session) on your computer, and enable SSH port forwarding by running command "ssh -L 4321:localhost:9178 <net_id>@216.165.13.149", or equivalent in PuTTY.  Note that 216.165.13.149 is the external IP address for the node 'hlog-1'. The node 'hlog-2' external IP address is 216.165.13.150. Also note that the Zeppelin port is randomized, it could be different next time you start a new server. The number 9178 is generated as shown in the example above.  

Now it is time to open a new web browser tab, input the following address to get the Zeppelin UI:

http://localhost:4321

File Permissions and Access Control Lists

Users can share files with others using ACL's.

An access control list (or ACL) gives per-file, per-directory and per-user control over who have permissions on the files. You can see the ACL for a file or directory with the getfacl command:

To open subdirectory permission to others, you need to open each higher level directory's navigation permission too:

Hadoop File System (HDFS)

HDFS stands for Hadoop Distributed File System. HDFS is a highly fault-tolerant file system and is designed to be deployed on low-cost hardware. HDFS provides high throughput access to application data and is suitable for applications that have large data sets.

TO UPLOAD DATA TO HDFS

HDFS stands for Hadoop Distributed File System. HDFS is a highly fault-tolerant file system and is designed to be deployed on low-cost hardware. HDFS provides high throughput access to application data and is suitable for applications that have large data sets.

To Upload Data to HDFS

To Get Data from HDFS

Peel Account Access

All HPC users should have an account on Peel. If you are enrolled in a class using the clusters, you may already have an account, try logging in first to check. See the HPC 'Getting and Renewing Accounts' page for instructions on how to get an account.  (NYU VPN is required). 

Your home directory on Peel is the same as that on the HPC Greene cluster. To see which cluster you are on, on the command prompt run 'echo $CLUSTER'.

Transferring Local Files

To replicate files from your computer to a Peel HDFS directory, first use 'scp' to copy from your computer to $HOME or $SCRATCH, then run 'hdfs dfs -put' to move them to the HDFS directory.

Transferring Big Data Files

Use Globus to transfer the data to /scratch first. The same /scratch folder is also available on Peel login nodes. Then copy the data from /scratch to your HDFS directory in Peel. /home is shared too between multiple clusters i.e., Greene and Peel. 

See instructions Data Transfers page to transfer files to and from NYU HPC storage.