Sqoop Commands Complete Reference

SQOOP Commands - Complete reference


sqoop version

sqoop help

sqoop list-databases --connect jdbc:mysql://localhost/ --username root --password ram
 
/first time error/
Warning: /usr/lib/sqoop/../hcatalog does not exist! HCatalog jobs will fail.
Please set $HCAT_HOME to the root of your HCatalog installation.
16/01/15 22:57:28 INFO sqoop.Sqoop: Running Sqoop version: 1.4.6
16/01/15 22:57:28 WARN tool.BaseSqoopTool: Setting your password on the command-line is insecure. Consider using -P instead.
16/01/15 22:57:28 INFO manager.MySQLManager: Preparing to use a MySQL streaming resultset.
16/01/15 22:57:28 ERROR sqoop.Sqoop: Got exception running Sqoop: java.lang.RuntimeException: Could not load db driver class: com.mysql.jdbc.Driver
 

Error1 
Warning: /usr/lib/sqoop/../hcatalog does not exist! HCatalog jobs will fail.

Fix1:

gedit .bashrc
export HCAT_HOME=${HIVE_HOME}/hcatalog/
export PATH=$HCAT_HOME/bin:$PATH

source .bashrc

Error 2:
16/01/15 22:57:28 ERROR sqoop.Sqoop: Got exception running Sqoop: java.lang.RuntimeException: Could not load db driver class: com.mysql.jdbc.Driver

note:
1. download connector for MySQL here http://dev.mysql.com/downloads/file/?id=459312
2. Extract the .gz and copy the .jar file in /usr/lib/sqoop/lib folder

sqoop list-tables --connect jdbc:mysql://localhost/sales --username root --password ram
Mysql: setting up environment

mysql -u root -p
create database sales
use sales
create table employee (empno int, empname varchar(50), empcity varchar(50));
insert into employee values (1001, 'Karthik', 'Chennai');
insert into employee values (1002, 'Saravanan', 'NY');
select * from employee;


sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee
 
ERROR tool.ImportTool: Error during import: No primary key could be found for table employee.
Please specify one with --split-by or perform a sequential import with '-m 1'.

sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee -m1
hadoop fs -ls /user/ram/employee
Found 2 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 08:58 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         39 2015-11-14 08:58 /user/ram/employee/part-m-00000

ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m-00000
1001,Karthik,Chennai
1002,Saravanan,NY

When you run the same import again:
sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee -m1
ERROR tool.ImportTool: Encountered IOException running import job: org.apache.hadoop.mapred.FileAlreadyExistsException:
Output directory hdfs://localhost:9000/user/ram/employee already exists

Appending the records in hdfs:
 
sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee -m1 --append
or
sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee --autoreset-to-one-mapper --append

ram@hadoop:~$ hadoop fs -ls /user/ram/employee/
15/11/14 09:13:40 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 3 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 08:58 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         39 2015-11-14 08:58 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         39 2015-11-14 09:13 /user/ram/employee/part-m-00001


sqoop import --connect jdbc:mysql://localhost/sales --username root --password ram --table employee --split-by empno --append
15/11/14 09:17:45 INFO mapreduce.Job: Counters: 30
    File System Counters
        FILE: Number of bytes read=0
        FILE: Number of bytes written=267100
        FILE: Number of read operations=0
        FILE: Number of large read operations=0
        FILE: Number of write operations=0
        HDFS: Number of bytes read=221
        HDFS: Number of bytes written=39
        HDFS: Number of read operations=8
        HDFS: Number of large read operations=0
        HDFS: Number of write operations=4
    Job Counters
        Launched map tasks=2
        Other local map tasks=2
        Total time spent by all maps in occupied slots (ms)=12800
        Total time spent by all reduces in occupied slots (ms)=0
        Total time spent by all map tasks (ms)=12800
        Total vcore-seconds taken by all map tasks=12800
        Total megabyte-seconds taken by all map tasks=13107200
    Map-Reduce Framework
        Map input records=2
        Map output records=2
        Input split bytes=221
        Spilled Records=0
        Failed Shuffles=0
        Merged Map outputs=0
        GC time elapsed (ms)=142
        CPU time spent (ms)=2880
        Physical memory (bytes) snapshot=363220992
        Virtual memory (bytes) snapshot=1681674240
        Total committed heap usage (bytes)=191889408
    File Input Format Counters
        Bytes Read=0
    File Output Format Counters
        Bytes Written=39


ram@hadoop:~$ hadoop fs -ls /user/ram/employee/
15/11/14 09:19:18 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 5 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 08:58 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         39 2015-11-14 08:58 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         39 2015-11-14 09:13 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         21 2015-11-14 09:17 /user/ram/employee/part-m-00002
-rw-r--r--   1 ram supergroup         18 2015-11-14 09:17 /user/ram/employee/part-m-00003
ram@hadoop:~$

Note: two files are imported as two mappers executed (002 and 003)

sqoop --options-file sqoopconf.conf

sqoopconf.conf
list-tables
--connect
jdbc:mysql://localhost/sales
--username
root
--password-file
/usr/local/hadoop/password.txt

Error while loading password file: The provided password file .password does not exist!ram@hadoop:~$ cat .password
ram
ram@hadoop:~$

note: You should save the password in a file on the users home directory with 400 permissions and
specify the path to that file using the --password-file argument

<<Need to check>>


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram --table employee \
     --split-by empno --append \
     --target-dir /user/ram/employee_result \
     --query "select empno, empname from employee" \
    --where "where empno=1001"
Error: Cannot specify --query and --table together.


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --split-by empno --append \
     --target-dir /user/ram/employee_result \
     --query "select empno, empname from employee" \
    --where "where empno=1001"
Error:
ERROR tool.ImportTool: Encountered IOException running import job: java.io.IOException:
Query [select empno, empname from employee] must contain '$CONDITIONS' in WHERE clause.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --target-dir /user/ram/employee_result \
     --table employee \
     --split-by empno \
     --append \
      --where " empno=1001 and empname = 'karthik' "
sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --target-dir /user/ram/employee_result \
     --table employee \
     --columns "empno, empname" \
     --split-by empno \
     --append \
      --where " empno=1001 and empname = 'karthik' "
sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --target-dir /user/ram/employee_result2 \
     --query 'select empno, empname from employee where empname =  and \$CONDITIONS  \
     --num-mappers 1 \
     --append
Free-form Query Imports

When importing a free-form query, you must specify a destination directory with --target-dir.

If you want to import the results of a query in parallel, then each map task will need to execute a copy of the query, with results partitioned by bounding conditions inferred by Sqoop. Your query must include the token $CONDITIONS which each Sqoop process will replace with a unique condition expression. You must also select a splitting column with --split-by.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --target-dir /user/ram/employee_result2 \
     --query "select empno, empname from employee where empno < 100 and \$CONDITIONS " \
     --num-mappers 1 \
     --append

Note:
If you are issuing the query wrapped with double quotes ("), you will have to use \$CONDITIONS instead of just $CONDITIONS to disallow your shell from treating it as a shell variable. For example, a double quoted query may look like: "SELECT * FROM x WHERE a='foo' AND \$CONDITIONS"

The facility of using free-form query in the current version of Sqoop is limited to simple queries where there are no ambiguous projections and no OR conditions in the WHERE clause. Use of complex queries such as queries that have sub-queries or joins leading to ambiguous projections can lead to unexpected results.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --target-dir /user/ram/employee_result2 \
     --query 'select empno, empname from employee where $CONDITIONS ' \
     --append \
     --num-mappers 1 


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --num-mappers 1 \
     --fetch-size 1000 \
     --fields-terminated-by '|' \
     --lines-terminated-by '\n'

ram@hadoop:~$ hadoop fs -ls /user/ram/employeeFound 2 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 22:05 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         39 2015-11-14 22:05 /user/ram/employee/part-m-00000
ram@hadoop:~$


ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m-000001001|Karthik|Chennai
1002|Saravanan|NY
ram@hadoop:~$

Boundary Query and Null Handling:

Boundary Query:
By default sqoop will use query select min(<split-by>), max(<split-by>) from <table name> to find out boundaries for creating splits. In some cases this query is not the most optimal so you can specify any arbitrary query returning two numeric columns using --boundary-query argument.

Null handling:
Sqoop will by default import NULL values as string null. Hive is however using string \N to denote NULL values and therefore predicates dealing with NULL (like IS NULL) will not work correctly. You should append parameters --null-string and --null-non-string in case of import job or --input-null-string and --input-null-non-string in case of an export job if you wish to properly preserve NULL values.


mysql> insert into employee (empno, empname) values (1003,'kavitha');Query OK, 1 row affected (0.07 sec)

mysql> select * from employee;
+-------+-----------+---------+
| empno | empname   | empcity |
+-------+-----------+---------+
|  1001 | Karthik   | Chennai |
|  1002 | Saravanan | NY      |
|  1003 | kavitha   | NULL    |
+-------+-----------+---------+
3 rows in set (0.00 sec)


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --boundary-query "select min(empno), max(empno) from employee" \
     --split-by empno \
     --fetch-size 1000 \
     --fields-terminated-by '|' \
     --lines-terminated-by '\n' \
     --null-string '\\N' \
     --null-non-string '\\N'--null-string '\\N': The string to be interpreted as null for string columns.
--null-non-string '\\N' :  The string to be interpreted as null for non string columns.

note:select min(empno), max(empno) from employee -- will work
select min (empno), max (empno) from employee -- will throw function not found error


ram@hadoop:~$ hadoop fs -ls /user/ram/employee/Found 4 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 22:29 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         21 2015-11-14 22:29 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-14 22:29 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         16 2015-11-14 22:29 /user/ram/employee/part-m-00002

ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m-000001001|Karthik|Chennai
ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m-00002
1003|kavitha|\N

Relaxed Isolation:
By default Sqoop uses Read Comitted isolation level
--relaxed isolation to use read uncommitted (not supported by all rdbms)

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --boundary-query "select min(empno), max(empno) from employee" \
     --split-by empno \
     --fetch-size 1000 \
     --fields-terminated-by '|' \
     --lines-terminated-by '\n' \
     --null-string '\\N' \
     --null-non-string '\\N' \
     --relaxed-isolation ram@hadoop:~$ hadoop dfs -cat /user/ram/employee/part-m-*1001|Karthik|Chennai
1002|Saravanan|NY
1003|kavitha|\N
ram@hadoop:~$ 



 --as-avrodatafile:
Avro data files are a compact, efficient binary format that provides interoperability with applications written in other programming languages. Avro also supports versioning, so that when, e.g., columns are added or removed from a table, previously imported data files can be processed along with new ones.

By default, data is not compressed. You can compress your data by using the deflate (gzip) algorithm with the -z or --compress argument, or specify any Hadoop compression codec using the --compression-codec argument. This applies to SequenceFile, text, and Avro files.


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --as-avrodatafile

ram@hadoop:~$ hadoop dfs -cat /user/ram/employee/part-*Obj avro.schema� {"type":"record","name":"employee","doc":"Sqoop import of employee","fields":[{"name":"empno","type":["null","int"],"default":null,"columnName":"empno","sqlType":"4"},{"name":"empname","type":["null","string"],"default":null,"columnName":"empname","sqlType":"12"},{"namc =Obj avro.schema� {"type":"record","name":"employee","doc":"Sqoop import of employee","fields":[{"name":"empno","type":["null","int"],"default":null,"columnName":"empno","sqlType":"4"},{"name":"empname","type":["null","string"],"default":null,"columnName":"empname","sqlType":"12"},{"name":"empcity","type":["null","string"],"default":null,"columnName":"empcity","sqlType":"12"}],"tableName":"employee"} � ���'h�S�T���� $ Saravanan NY � ���'h�S�T����Obj avro.schema� {"type":"record","name":"employee","doc":"Sqoop import of employee","fields":[{"name":"empno","type":["null","int"],"default":null,"columnName":"empno","sqlType":"4"},{"name":"empname","type":["null","string"],"default":null,"columnName":"empname","sqlType":"12"},{"name":"empcity","type":["null","string"],"default":null,"columnName":"empcity","sqlType":"12"}],"tableName":"employee"}򐘩N�&��~�c�H�, kavitha򐘩N�&��~�c�H�,ram@hadoop:~$


$ hadoop dfs -text /user/ram/employee/part-*{"empno":{"int":1001},"empname":{"string":"Karthik"},"empcity":{"string":"Chennai"}}
{"empno":{"int":1002},"empname":{"string":"Saravanan"},"empcity":{"string":"NY"}}
{"empno":{"int":1003},"empname":{"string":"kavitha"},"empcity":null}
ram@hadoop:~$


 --as-sequencefile:SequenceFiles are a binary format that store individual records in custom record-specific data types. These data types are manifested as Java classes. Sqoop will automatically generate these data types for you

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --as-sequencefile

ram@hadoop:~$ hadoop dfs -cat /user/ram/employee/part-*SEQ !org.apache.hadoop.io.LongWritablemployee;��|ǀ�@E�A�'qf �KarthikChennaiSEQ !org.apache.hadoop.io.LongWritablemployee�
�      Saravanan NYSEQ !org.apache.hadoop.io.LongWritablemployeecb�[N���sc �<� �kavitha

further reference: reading sequencial file via java program
http://hadooptutorial.info/reading-and-writing-sequencefile-example/

Compression:By default, data is not compressed. You can compress your data by using the deflate (gzip) algorithm with the -z or --compress argument, or specify any Hadoop compression codec using the --compression-codec argument. This applies to SequenceFile, text, and Avro files.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --as-avrodatafile \
     --compressram@hadoop:~$ hadoop dfs -ls /user/ram/employee/
Found 4 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 23:11 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup        481 2015-11-14 23:11 /user/ram/employee/part-m-00000.avro
-rw-r--r--   1 ram supergroup        478 2015-11-14 23:11 /user/ram/employee/part-m-00001.avro
-rw-r--r--   1 ram supergroup        473 2015-11-14 23:11 /user/ram/employee/part-m-00002.avro
ram@hadoop:~$

ram@hadoop:~$ hadoop dfs -text /user/ram/employee/part*{"empno":{"int":1001},"empname":{"string":"Karthik"},"empcity":{"string":"Chennai"}}
{"empno":{"int":1002},"empname":{"string":"Saravanan"},"empcity":{"string":"NY"}}
{"empno":{"int":1003},"empname":{"string":"kavitha"},"empcity":null}
ram@hadoop:~$


sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --compress \
    --compression-codec org.apache.hadoop.io.compress.BZip2Codec

ram@hadoop:~$ hadoop dfs -ls /user/ram/employee/Found 4 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 23:20 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         62 2015-11-14 23:20 /user/ram/employee/part-m-00000.bz2
-rw-r--r--   1 ram supergroup         62 2015-11-14 23:20 /user/ram/employee/part-m-00001.bz2
-rw-r--r--   1 ram supergroup         59 2015-11-14 23:20 /user/ram/employee/part-m-00002.bz2
ram@hadoop:~$

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --compress \
    --compression-codec org.apache.hadoop.io.compress.SnappyCodecError: native snappy library not available: SnappyCompressor has not been loaded.
<<how to download and configure snappy:>>


--direct to Use direct connector if exists for the database

For example, MySQL provides the mysqldump tool which can export data from MySQL to other systems very quickly. By supplying the --direct argument, you are specifying that Sqoop should attempt the direct import channel. This channel may be higher performance than using JDBC.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --direct

ram@hadoop:~$ hadoop dfs -ls /user/ram/employee/Found 4 items
-rw-r--r--   1 ram supergroup          0 2015-11-14 23:32 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         21 2015-11-14 23:32 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-14 23:32 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         18 2015-11-14 23:32 /user/ram/employee/part-m-00002

ram@hadoop:~$ hadoop dfs -ls /user/ram/employee/part-m*-rw-r--r--   1 ram supergroup         21 2015-11-14 23:32 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-14 23:32 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         18 2015-11-14 23:32 /user/ram/employee/part-m-00002
ram@hadoop:~$

incremental imports

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno \
     --direct    \
     --incremental lastmodified \
     --last-value 0 \
     --check-column empnonote: --delete-target-dir can not be used with incremental imports.

sqoop import --connect jdbc:mysql://localhost/sales \
     --username root --password ram \
     --table employee \
     --split-by empno \
     --direct    \
     --check-column empno \
     --incremental append \
     --last-value 0 Error: java.lang.RuntimeException: Column type is neither timestamp nor date!


sqoop import --connect jdbc:mysql://localhost/sales \
    --username root --password ram \
    --table employee \
    --split-by empno \
    --direct\
    --check-column empno \
    --incremental append \
    --last-value 0
15/11/15 05:19:26 INFO tool.ImportTool: Incremental import complete! To run another incremental import of all data following this import, supply the following arguments:
15/11/15 05:19:26 INFO tool.ImportTool:  --incremental append
15/11/15 05:19:26 INFO tool.ImportTool:   --check-column empno
15/11/15 05:19:26 INFO tool.ImportTool:   --last-value 1003
15/11/15 05:19:26 INFO tool.ImportTool: (Consider saving this with 'sqoop job --create')



ram@hadoop:~$ hadoop fs -ls /user/ram/
Found 4 itemsdrwxr-xr-x   - ram supergroup          0 2015-11-15 05:19 /user/ram/_sqoop
drwxr-xr-x   - ram supergroup          0 2015-11-15 05:19 /user/ram/employee
drwxr-xr-x   - ram supergroup          0 2015-11-14 16:10 /user/ram/employee_result
drwxr-xr-x   - ram supergroup          0 2015-11-14 16:28 /user/ram/employee_result2

ram@hadoop:~$ hadoop fs -ls /user/ram/employeeFound 3 items
-rw-r--r--   1 ram supergroup         21 2015-11-15 05:19 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-15 05:19 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         18 2015-11-15 05:19 /user/ram/employee/part-m-00002

ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m*1001,Karthik,Chennai
1002,Saravanan,NY
1003,kavitha,NULL
ram@hadoop:~$


Insert 2 more records:

mysql> insert into employee values (1004, 'ramkumar','NJ');Query OK, 1 row affected (0.07 sec)

mysql> insert into employee values (1005, 'gopal','NJ');Query OK, 1 row affected (0.05 sec)


sqoop import --connect jdbc:mysql://localhost/sales \
    --username root --password ram \
    --table employee \
    --split-by empno \
    --direct \
    --check-column empno \
    --incremental append \
    --last-value 100315/11/15 05:27:21 INFO tool.ImportTool:  --incremental append
15/11/15 05:27:21 INFO tool.ImportTool:   --check-column empno
15/11/15 05:27:21 INFO tool.ImportTool:   --last-value 1005
15/11/15 05:27:21 INFO tool.ImportTool: (Consider saving this with 'sqoop job --create')

ram@hadoop:~$ hadoop fs -ls /user/ram/employeeFound 5 items
-rw-r--r--   1 ram supergroup         21 2015-11-15 05:19 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-15 05:19 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         18 2015-11-15 05:19 /user/ram/employee/part-m-00002
-rw-r--r--   1 ram supergroup         17 2015-11-15 05:27 /user/ram/employee/part-m-00003
-rw-r--r--   1 ram supergroup         14 2015-11-15 05:27 /user/ram/employee/part-m-00004

ram@hadoop:~$ hadoop fs -cat /user/ram/employee/part-m*1001,Karthik,Chennai
1002,Saravanan,NY
1003,kavitha,NULL
1004,ramkumar,NJ
1005,gopal,NJ
ram@hadoop:~$


Create and execute Sqoop jobs:

sqoop job --create employee_bulk_import \
     -- import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --delete-target-dir \
     --target-dir /user/ram/employee \
     --table employee \
     --split-by empno

sqoop job --list
sqoop job --show employee_bulk_import
sqoop job --exec employee_bulk_import
sqoop job --delete employee_bulk_import



Hive Import

- If you have a Hive metastore associated with your HDFS cluster, Sqoop can also import the data into Hive by generating and executing a CREATE TABLE statement to define the data’s layout in Hive. Importing data into Hive is as simple as adding the --hive-import option to your Sqoop command line.

if the Hive table already exists, you can specify the --hive-overwrite

Sqoop will generate a Hive script containing a CREATE TABLE operation defining your columns using Hive’s types, and a LOAD DATA INPATH statement to move the data files into Hive’s warehouse directory.

Even though Hive supports escaping characters, it does not handle escaping of new-line character. Also, it does not support the notion of enclosing characters that may include field delimiters in the enclosed string

t is therefore recommended that you choose unambiguous field and record-terminating delimiters without the help of escaping and enclosing characters when working with Hive; this is due to limitations of Hive’s input parsing abilities. If you do use --escaped-by, --enclosed-by, or --optionally-enclosed-by when importing data into Hive, Sqoop will print a warning message.


Hive will have problems using Sqoop-imported data if your database’s rows contain string fields that have Hive’s default row delimiters (\n and \r characters) or column delimiters (\01 characters) present in them. You can use the --hive-drop-import-delims option to drop those characters on import to give Hive-compatible text data. Alternatively, you can use the --hive-delims-replacement option to replace those characters with a user-defined string on import to give Hive-compatible text data

Sqoop will pass the field and record delimiters through to Hive. If you do not set any delimiters and do use --hive-import, the field delimiter will be set to ^A and the record delimiter will be set to \n to be consistent with Hive’s defaults.

You should append parameters --null-string and --null-non-string in case of import job or --input-null-string and --input-null-non-string in case of an export job if you wish to properly preserve NULL values. 

The table name used in Hive is, by default, the same as that of the source table. You can control the output table name with the --hive-table option.

Hive can put data into partitions for more efficient query performance. You can tell a Sqoop job to import data for Hive into a particular partition by specifying the --hive-partition-key and --hive-partition-value arguments.

You can import compressed tables into Hive using the --compress and --compression-codec options. One downside to compressing tables imported into Hive is that many codecs cannot be split for processing by parallel map tasks. The lzop codec, however, does support splitting. When importing tables with this codec, Sqoop will automatically index the files for splitting and configuring a new Hive table with the correct InputFormat.
 


Note: External tables/partitions are not supported
Note: Partitions are created under the default location
Note: Existing Hive table can’t be truncated before the import
The --delete-target-dir option is not respected.


sqoop import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --delete-target-dir \
     --table employee \
     --split-by empno \
     --direct \
     --hive-import \
     --hive-table=hive_employee

15/11/15 06:17:35 INFO mapreduce.ImportJobBase: Transferred 88 bytes in 35.049 seconds (2.5108 bytes/sec)
15/11/15 06:17:35 INFO mapreduce.ImportJobBase: Retrieved 5 records.
15/11/15 06:17:35 INFO manager.SqlManager: Executing SQL statement: SELECT t.* FROM `employee` AS t LIMIT 1
15/11/15 06:17:35 INFO hive.HiveImport: Loading uploaded data into Hive
15/11/15 06:17:54 INFO hive.HiveImport: Time taken: 2.839 seconds
15/11/15 06:17:54 INFO hive.HiveImport: Loading data to table default.hive_employee
15/11/15 06:17:55 INFO hive.HiveImport: Table default.hive_employee stats: [numFiles=5, totalSize=88]
15/11/15 06:17:55 INFO hive.HiveImport: Hive import complete.
15/11/15 06:17:55 INFO hive.HiveImport: Export directory is contains the _SUCCESS file only, removing the directory.
ram@hadoop:~$

ram@hadoop:~$ hadoop fs -ls /user/hive/warehouse/hive_employeeFound 5 items
-rwxrwxr-x   1 ram supergroup         21 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00000
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00001
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00002
-rwxrwxr-x   1 ram supergroup         17 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00003
-rwxrwxr-x   1 ram supergroup         14 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00004

ram@hadoop:~$ hadoop fs -cat /user/hive/warehouse/hive_employee/part-m*1001 Karthik Chennai
1002 Saravanan NY
1003 kavitha NULL
1004 ramkumar NJ
1005 gopal NJ
ram@hadoop:~$

ram@hadoop:~$ hive -e "select * from hive_employee"OK
1001    Karthik    Chennai
1002    Saravanan    NY
1003    kavitha    NULL
1004    ramkumar    NJ
1005    gopal    NJ
Time taken: 2.727 seconds, Fetched: 5 row(s)

ram@hadoop:~$ hive -e "select * from hive_employee where empname = 'gopal'"1005    gopal    NJ
Time taken: 3.182 seconds, Fetched: 1 row(s)
ram@hadoop:~$


What happens when you run the import again:The --delete-target-dir option is not respected.

sqoop import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --delete-target-dir \
     --table employee \
     --split-by empno \
     --direct \
     --hive-import \
     --hive-table=hive_employee


ram@hadoop:~$ hadoop fs -ls /user/hive/warehouse/hive_employeeFound 10 items
-rwxrwxr-x   1 ram supergroup         21 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00000
-rwxrwxr-x   1 ram supergroup         21 2015-11-15 06:25 /user/hive/warehouse/hive_employee/part-m-00000_copy_1
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00001
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:25 /user/hive/warehouse/hive_employee/part-m-00001_copy_1
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00002
-rwxrwxr-x   1 ram supergroup         18 2015-11-15 06:25 /user/hive/warehouse/hive_employee/part-m-00002_copy_1
-rwxrwxr-x   1 ram supergroup         17 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00003
-rwxrwxr-x   1 ram supergroup         17 2015-11-15 06:25 /user/hive/warehouse/hive_employee/part-m-00003_copy_1
-rwxrwxr-x   1 ram supergroup         14 2015-11-15 06:17 /user/hive/warehouse/hive_employee/part-m-00004
-rwxrwxr-x   1 ram supergroup         14 2015-11-15 06:25 /user/hive/warehouse/hive_employee/part-m-00004_copy_1
 
ram@hadoop:~$ hive -e "select * from hive_employee"
 OK
1001    Karthik    Chennai
1001    Karthik    Chennai
1002    Saravanan    NY
1002    Saravanan    NY
1003    kavitha    NULL
1003    kavitha    NULL
1004    ramkumar    NJ
1004    ramkumar    NJ
1005    gopal    NJ
1005    gopal    NJ
Time taken: 2.56 seconds, Fetched: 10 row(s)
ram@hadoop:~$



sqoop import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --delete-target-dir \
     --table employee \
     --split-by empno \
     --direct \
     --hive-import \
     --hive-overwrite



Note: Hive-Overwrite will delete all the files employee folder before loading the data

hive -e "drop table employee"

sqoop import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --table employee \
     --split-by empno \
     --direct \
     --hive-import \
     --hive-overwrite \
     --hive-drop-import-delims  \
     --hive-partition-key empno \
    --hive-partition-value 1002ERROR: Direct import currently do not support dropping hive delimiters, please remove parameter --hive-drop-import-delims.


hive -e "drop table employee"

sqoop import \
     --username root \
     --password ram \
     --connect jdbc:mysql://localhost/sales \
     --table employee \
     --split-by empno \
     --direct \
     --hive-import \
     --hive-overwrite \
     --hive-partition-key "empcity" \
     --hive-partition-value "NJ"

<<this creates files under /user/ram/employee. need to check>>

ram@hadoop:~$ hadoop fs -ls /user/ram/employee15/11/15 15:36:02 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Found 6 items
-rw-r--r--   1 ram supergroup          0 2015-11-15 15:34 /user/ram/employee/_SUCCESS
-rw-r--r--   1 ram supergroup         21 2015-11-15 15:34 /user/ram/employee/part-m-00000
-rw-r--r--   1 ram supergroup         18 2015-11-15 15:34 /user/ram/employee/part-m-00001
-rw-r--r--   1 ram supergroup         18 2015-11-15 15:34 /user/ram/employee/part-m-00002
-rw-r--r--   1 ram supergroup         17 2015-11-15 15:34 /user/ram/employee/part-m-00003
-rw-r--r--   1 ram supergroup         14 2015-11-15 15:34 /user/ram/employee/part-m-00004

Installing MySQL on Ubuntu 14.04

Simple Steps:

1. Install mySql by entering below command

sudo apt-get install mysql-server

Note: you have to feed the root password during the installation

2. We are done. MySQL is installed

3. Now lets login to MySQL and do some basic CURD operations

mysql -u root -p

Type 'help;' or '\h' for help. Type '\c' to clear the current input statement.

mysql> 

4.  

mysql> create table employee (empno int, empname varchar(50), empcity varchar(50));
ERROR 1046 (3D000): No database selected

mysql> create database sales
    -> ;
Query OK, 1 row affected (0.00 sec)


mysql> use sales
Database changed

mysql> create table employee (empno int, empname varchar(50), empcity varchar(50));
Query OK, 0 rows affected (0.10 sec)

mysql> insert into employee values (1001, 'Karthik', 'Chennai');
Query OK, 1 row affected (0.06 sec)

mysql> insert into employee values (1002, 'Saravanan', 'NY');
Query OK, 1 row affected (0.06 sec)

mysql> select * from employee;
+-------+-----------+---------+
| empno | empname   | empcity |
+-------+-----------+---------+
|  1001 | Karthik   | Chennai |
|  1002 | Saravanan | NY      |
+-------+-----------+---------+
2 rows in set (0.00 sec)

mysql> exit

Aborted
ram@hadoop:~$ 

How to install Sqoop 1.4.6 on Hadoop 2.7.1 in Ubuntu 14.04

Steps:
1. Download the Sqoop .tar.gz file on below path
http://sqoop.apache.org/

2. extract the .tar.gz file, you will get sqoop setup folder

3. move the sqoop setup folder to /usr/local/sqoop path
sudo mv hadoop/sqoop-1.4.6.bin__hadoop-0.23 /usr/lib/sqoop

4. add below lines in .bashrc file
export SQOOP_HOME=/usr/lib/sqoop
export PATH=$PATH:$SQOOP_HOME/bin

5. run below command to update sqoop path
source .bashrc

6. check sqoop version
sqoop version

15/11/13 22:36:32 INFO sqoop.Sqoop: Running Sqoop version: 1.4.6
Sqoop 1.4.6git commit id c0c5a81723759fa575844a0a1eae8f510fa32c25
Compiled by root on Mon Apr 27 14:20:17 CST 2015

Hadoop Admin Commands quick reference

Reference:http://www.thegeekstuff.com/
 
Hadoop filesystem commands


hadoop fs -mkdir /dir
hadoop fs -ls
hadoop fs -cat <filename>
hadoop fs -rm <<filename>>
hadoop fs -mv file:///data/datafile /user/hduser/data
hadoop fs -touchz <<filename>> -create empty file
hadoop fs -stat <filename>
hadoop fs -expunge   <<empty trash on hdfs>>
ram@ram:/etc/init.d$ hadoop fs -du /user
50270  /user/1.log
0      /user/hive

hadoop fs -copyFromLocal <source> <destination>
hadoop fs -copyToLocal <source> <destination>
hadoop fs -put <source> <destination> --copy from remote location
hadoop fs -get <source> <destination> --copy to remote location
hadoop distcp hdfs://192.168.0.8:8020/input hdfs://192.168.0.8:8020/output
-- Copy data from one cluster to another using the cluster URL
hadoop fs -setrep -w 3 file1
hadoop fs -getmerge mydir bigfile
-- Merge files in mydir directory and download it as one big file




Hadoop Job Commands

hadoop job -submit <job-file>
hadoop job -status <job-id>
hadoop job -history
hadoop job -kill-task <task-id>


ram@ram:/etc/init.d$ hadoop job -list all
DEPRECATED: Use of
this script to execute mapred command is deprecated.
Instead use the mapred command for it.

15/07/29 21:03:51 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/07/29 21:03:51 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
Total jobs:0
                  JobId         State         StartTime        UserName           Queue      Priority     UsedContainers     RsvdContainers     UsedMem RsvdMem     NeededMem       AM info


ram@ram:/etc/init.d$ hadoop job -list-active-trackers
DEPRECATED: Use of this script to execute mapred command is deprecated.
Instead use the mapred command for it.

15/07/29 21:04:24 INFO client.RMProxy: Connecting to ResourceManager at /0.0.0.0:8032
tracker_ram:49874



Hadoop Namenode commands

hadoop namenode -format
hadoop namenode -upgrade
hadoop namenode -recover -force
hadoop fsck -delete    <<delete corrupted files>>
hadoop fsck -move    <<move corrupted files to lost+found folder>

-- Recover namenode metadata after a cluster failure (may lose data)

ram@ram:/etc/init.d$ stop-dfs.sh
Stopping namenodes on [localhost]
localhost: stopping namenode
localhost: stopping datanode
Stopping secondary namenodes [0.0.0.0]
0.0.0.0: stopping secondarynamenode

ram@ram:/etc/init.d$ stop-yarn.sh
stopping yarn daemons
stopping resourcemanager
localhost: stopping nodemanager
no proxyserver to stop


ram@ram:/etc/init.d$ start-dfs.sh
Starting namenodes on [localhost]
localhost: starting namenode, logging to /usr/local/hadoop/logs/hadoop-ram-namenode-ram.out
localhost: starting datanode, logging to /usr/local/hadoop/logs/hadoop-ram-datanode-ram.out
Starting secondary namenodes [0.0.0.0]
0.0.0.0: starting secondarynamenode, logging to /usr/local/hadoop/logs/hadoop-ram-secondarynamenode-ram.out


ram@ram:/etc/init.d$ start-yarn.sh
starting yarn daemons
starting resourcemanager, logging to /usr/local/hadoop/logs/yarn-ram-resourcemanager-ram.out
localhost: starting nodemanager, logging to /usr/local/hadoop/logs/yarn-ram-nodemanager-ram.out
ram@ram:/etc/init.d$


ram@ram:/etc/init.d$ jps
6330 NodeManager
6192 ResourceManager
5827 DataNode
6649 Jps
6028 SecondaryNameNode
5664 NameNode



ram@ram:/etc/init.d$ hadoop fsck /
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

Connecting to namenode via http://localhost:50070/fsck?ugi=ram&path=%2F
FSCK started by ram (auth:SIMPLE) from /127.0.0.1 for path / at Wed Jul 29 20:56:55 IST 2015
.
/user/1.log:  Under replicated BP-393036986-127.0.1.1-1437358619878:blk_1073741825_1001. Target Replicas is 3 but found 1 replica(s).
Status: HEALTHY
 Total size:    50270 B
 Total dirs:    7
 Total files:    1
 Total symlinks:        0
 Total blocks (validated):    1 (avg. block size 50270 B)
 Minimally replicated blocks:    1 (100.0 %)
 Over-replicated blocks:    0 (0.0 %)
 Under-replicated blocks:    1 (100.0 %)
 Mis-replicated blocks:        0 (0.0 %)
 Default replication factor:    3
 Average block replication:    1.0
 Corrupt blocks:        0
 Missing replicas:        2 (66.666664 %)
 Number of data-nodes:        1
 Number of racks:        1
FSCK ended at Wed Jul 29 20:56:55 IST 2015 in 3 milliseconds


The filesystem under path '/' is HEALTHY



ram@ram:/etc/init.d$ hadoop fsck / -files -blocks -locations -racks
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.


Connecting to namenode via http://localhost:50070/fsck?ugi=ram&files=1&blocks=1&locations=1&racks=1&path=%2F
FSCK started by ram (auth:SIMPLE) from /127.0.0.1 for path / at Wed Jul 29 20:58:22 IST 2015
/ <dir>
/tmp <dir>
/tmp/hive <dir>
/tmp/hive/ram <dir>
/user <dir>
/user/1.log 50270 bytes, 1 block(s):  Under replicated BP-393036986-127.0.1.1-1437358619878:blk_1073741825_1001. Target Replicas is 3 but found 1 replica(s).
0. BP-393036986-127.0.1.1-1437358619878:blk_1073741825_1001 len=50270 repl=1 [/default-rack/127.0.0.1:50010]

/user/hive <dir>
/user/hive/warehouse <dir>
Status: HEALTHY
 Total size:    50270 B
 Total dirs:    7
 Total files:    1
 Total symlinks:        0
 Total blocks (validated):    1 (avg. block size 50270 B)
 Minimally replicated blocks:    1 (100.0 %)
 Over-replicated blocks:    0 (0.0 %)
 Under-replicated blocks:    1 (100.0 %)
 Mis-replicated blocks:        0 (0.0 %)
 Default replication factor:    3
 Average block replication:    1.0
 Corrupt blocks:        0
 Missing replicas:        2 (66.666664 %)
 Number of data-nodes:        1
 Number of racks:        1
FSCK ended at Wed Jul 29 20:58:22 IST 2015 in 3 milliseconds


The filesystem under path '/' is HEALTHY



Hadoop dfsadmin commands

ram@ram:/etc/init.d$ hadoop dfsadmin -report
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

Configured Capacity: 98496679936 (91.73 GB)
Present Capacity: 80164052992 (74.66 GB)
DFS Remaining: 80163958784 (74.66 GB)
DFS Used: 94208 (92 KB)
DFS Used%: 0.00%
Under replicated blocks: 1
Blocks with corrupt replicas: 0
Missing blocks: 0
Missing blocks (with replication factor 1): 0

-------------------------------------------------
Live datanodes (1):

Name: 127.0.0.1:50010 (localhost)
Hostname: ram
Decommission Status : Normal
Configured Capacity: 98496679936 (91.73 GB)
DFS Used: 94208 (92 KB)
Non DFS Used: 18332626944 (17.07 GB)
DFS Remaining: 80163958784 (74.66 GB)
DFS Used%: 0.00%
DFS Remaining%: 81.39%
Configured Cache Capacity: 0 (0 B)
Cache Used: 0 (0 B)
Cache Remaining: 0 (0 B)
Cache Used%: 100.00%
Cache Remaining%: 0.00%
Xceivers: 1
Last contact: Wed Jul 29 21:06:41 IST 2015



ram@ram:/etc/init.d$ hadoop dfsadmin -setQuota 10 /user
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.



ram@ram:/etc/init.d$ hadoop fs -count -q /user
          10               6            none             inf            3            1              50270 /user
ram@ram:/etc/init.d$

ram@ram:/etc/init.d$ hadoop dfsadmin -safemode enter
Safe mode is ON


ram@ram:/etc/init.d$ hadoop dfsadmin -saveNamespace
<<Backup Metadata (fsimage & edits). Put cluster in safe mode before this command.>>
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

Save namespace successful


ram@ram:/etc/init.d$ hadoop dfsadmin -safemode get
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

Safe mode is ON



ram@ram:/etc/init.d$ hadoop dfsadmin -safemode leave
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

Safe mode is OFF
ram@ram:/etc/init.d$



Hadoop yarn commands



Hadoop Balancer commands

ram@ram:/etc/init.d$ start-balancer.sh
starting balancer, logging to /usr/local/hadoop/logs/hadoop-ram-balancer-ram.out

hadoop dfsadmin -setBalancerBandwidth <bandwidthinbytes>
ram@ram:/etc/init.d$ hadoop balancer -threshold 20
DEPRECATED: Use of this script to execute hdfs command is deprecated.
Instead use the hdfs command for it.

15/07/29 21:16:22 INFO balancer.Balancer: Using a threshold of 20.0
15/07/29 21:16:22 INFO balancer.Balancer: namenodes  = [hdfs://localhost:9000]
15/07/29 21:16:22 INFO balancer.Balancer: parameters = Balancer.Parameters[BalancingPolicy.Node, threshold=20.0, max idle iteration = 5, number of nodes to be excluded = 0, number of nodes to be included = 0]
Time Stamp               Iteration#  Bytes Already Moved  Bytes Left To Move  Bytes Being Moved
15/07/29 21:16:22 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
15/07/29 21:16:24 INFO net.NetworkTopology: Adding a new node: /default-rack/127.0.0.1:50010
15/07/29 21:16:24 INFO balancer.Balancer: 0 over-utilized: []
15/07/29 21:16:24 INFO balancer.Balancer: 0 underutilized: []
The cluster is balanced. Exiting...
29 Jul, 2015 9:16:24 PM           0                  0 B                 0 B               -1 B
29 Jul, 2015 9:16:24 PM  Balancing took 2.217 seconds
ram@ram:/etc/init.d$

HDFS Architecture

Source: http://hadoop.apache.org/docs/r2.7.0/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html

 Introduction

- The Hadoop Distributed File System (HDFS) is a distributed file system designed to run on commodity hardware

- HDFS is highly fault-tolerant

Assumptions and Goals

Hardware Failure

An HDFS instance may consist of hundreds or thousands of server machines, each storing part of the file system’s data. detection of faults and quick, automatic recovery from them is a core architectural goal of HDFS

Streaming Data Access

Applications that run on HDFS need streaming access to their data sets.

HDFS is designed more for batch processing rather than interactive use by users.

The emphasis is on high throughput of data access rather than low latency of data access

Large Data Sets

Applications that run on HDFS have large data sets. A typical file in HDFS is gigabytes to terabytes in size.

Simple Coherency Model

HDFS applications need a write-once-read-many access model for files.

A file once created, written, and closed need not be changed.

Map/Reduce application or a web crawler application fits perfectly with this model. There is a plan to support appending-writes to files in the future.

“Moving Computation is Cheaper than Moving Data”

A computation requested by an application is much more efficient if it is executed near the data it operates on. HDFS provides interfaces for applications to move themselves closer to where the data is located.

Portability Across Heterogeneous Hardware and Software Platforms

HDFS has been designed to be easily portable from one platform to another

NameNode and DataNodes

HDFS has a master/slave architecture. An HDFS cluster consists of a single NameNode, a master server that manages the file system namespace and regulates access to files by clients. In addition, there are a number of DataNodes, usually one per node in the cluster, which manage storage attached to the nodes that they run on.

Internally, a file is split into one or more blocks and these blocks are stored in a set of DataNodes. The DataNodes are responsible for serving read and write requests from the file system’s clients. The DataNodes also perform block creation, deletion, and replication upon instruction from the NameNode.

Name Node and Data Nodes machines typically run a GNU/Linux operating system (OS).

HDFS is built using the Java language; any machine that supports Java can run the NameNode or the DataNode software.

A typical deployment has a dedicated machine that runs only the NameNode software.

Each of the other machines in the cluster runs one instance of the DataNode software.

The architecture does not preclude running multiple DataNodes on the same machine but in a real deployment that is rarely the case.

The existence of a single NameNode in a cluster greatly simplifies the architecture of the systemThe NameNode is the arbitrator and repository for all HDFS metadata. The system is designed in such a way that user data never flows through the NameNode.

The File System Namespace

A user or an application can create directories and store files inside these directories.

HDFS does not support hard links or soft links. However, the HDFS architecture does not preclude implementing these features.

The NameNode maintains the file system namespace. Any change to the file system namespace or its properties is recorded by the NameNode. An application can specify the number of replicas of a file that should be maintained by HDFS. The number of copies of a file is called the replication factor of that file. This information is stored by the NameNode.

Data Replication

HDFS is designed to reliably store very large files across machines in a large cluster. It stores each file as a sequence of blocks; all blocks in a file except the last block are the same size

The blocks of a file are replicated for fault tolerance. The block size and replication factor are configurable per file.

An application can specify the number of replicas of a file. The replication factor can be specified at file creation time and can be changed later. Files in HDFS are write-once and have strictly one writer at any time.

The NameNode makes all decisions regarding replication of blocks. It periodically receives a Heartbeat and a Blockreport from each of the DataNodes in the cluster. A Blockreport contains a list of all blocks on a DataNode.

Replica Placement: The First Baby Steps

Large HDFS instances run on a cluster of computers that commonly spread across many racks. Communication between two nodes in different racks has to go through switches. In most cases, network bandwidth between machines in the same rack is greater than network bandwidth between machines in different racks.

A simple but non-optimal policy is to place replicas on unique racks. This prevents losing data when an entire rack fails and allows use of bandwidth from multiple racks when reading data.

For the common case, when the replication factor is three, HDFS’s placement policy is to put one replica on one node in the local rack, another on a different node in the local rack, and the last on a different node in a different rack. This policy cuts the inter-rack write traffic which generally improves write performance. With this policy, the replicas of a file do not evenly distribute across the racks. One third of replicas are on one node, two thirds of replicas are on one rack, and the other third are evenly distributed across the remaining racks. This policy improves write performance without compromising data reliability or read performance.

Replica Selection

To minimize global bandwidth consumption and read latency, HDFS tries to satisfy a read request from a replica that is closest to the reader. If there exists a replica on the same rack as the reader node, then that replica is preferred to satisfy the read request. If angg/ HDFS cluster spans multiple data centers, then a replica that is resident in the local data center is preferred over any remote replica.

Safemode

On startup, the NameNode enters a special state called Safemode. Replication of data blocks does not occur when the NameNode is in the Safemode state. The NameNode receives Heartbeat and Blockreport messages from the DataNodes. A Blockreport contains the list of data blocks that a DataNode is hosting. Each block has a specified minimum number of replicas. A block is considered safely replicated when the minimum number of replicas of that data block has checked in with the NameNode. After a configurable percentage of safely replicated data blocks checks in with the NameNode (plus an additional 30 seconds), the NameNode exits the Safemode state. It then determines the list of data blocks (if any) that still have fewer than the specified number of replicas. The NameNode then replicates these blocks to other DataNodes.

The Persistence of File System Metadata

The HDFS namespace is stored by the NameNode. The NameNode uses a transaction log called the EditLog to persistently record every change that occurs to file system metadata. The NameNode uses a file in its local host OS file system to store the EditLog. The entire file system namespace, including the mapping of blocks to files and file system properties, is stored in a file called the FsImage. The FsImage is stored as a file in the NameNode’s local file system too.

The NameNode keeps an image of the entire file system namespace and file Blockmap in memory. NameNode with 4 GB of RAM is plenty to support a huge number of files and directories. When the NameNode starts up, it reads the FsImage and EditLog from disk, applies all the transactions from the EditLog to the in-memory representation of the FsImage, and flushes out this new version into a new FsImage on disk. It can then truncate the old EditLog because its transactions have been applied to the persistent FsImage. This process is called a checkpoint. In the current implementation, a checkpoint only occurs when the NameNode starts up. Work is in progress to support periodic checkpointing in the near future.

The DataNode stores HDFS data in files in its local file system. The DataNode has no knowledge about HDFS files. It stores each block of HDFS data in a separate file in its local file system. The DataNode does not create all files in the same directory. Instead, it uses a heuristic to determine the optimal number of files per directory and creates subdirectories appropriately. It is not optimal to create all local files in the same directory because the local file system might not be able to efficiently support a huge number of files in a single directory. When a DataNode starts up, it scans through its local file system, generates a list of all HDFS data blocks that correspond to each of these local files and sends this report to the NameNode: this is the Blockreport.

The Communication Protocols

All HDFS communication protocols are layered on top of the TCP/IP protocol. A client establishes a connection to a configurable TCP port on the NameNode machine. It talks the ClientProtocol with the NameNode. The DataNodes talk to the NameNode using the DataNode Protocol. A Remote Procedure Call (RPC) abstraction wraps both the Client Protocol and the DataNode Protocol. By design, the NameNode never initiates any RPCs. Instead, it only responds to RPC requests issued by DataNodes or clients.

Robustness

The primary objective of HDFS is to store data reliably even in the presence of failures. The three common types of failures are NameNode failures, DataNode failures and network partitions.

1. Data Disk Failure, Heartbeats and Re-Replication

Each DataNode sends a Heartbeat message to the NameNode periodically. The NameNode detects this condition by the absence of a Heartbeat message. The NameNode marks DataNodes without recent Heartbeats as dead and does not forward any new IO requests to them. Any data that was registered to a dead DataNode is not available to HDFS any more. DataNode death may cause the replication factor of some blocks to fall below their specified value. The NameNode constantly tracks which blocks need to be replicated and initiates replication whenever necessary. The necessity for re-replication may arise due to many reasons: a DataNode may become unavailable, a replica may become corrupted, a hard disk on a DataNode may fail, or the replication factor of a file may be increased.

2. Cluster Rebalancing

The HDFS architecture is compatible with data rebalancing schemes. A scheme might automatically move data from one DataNode to another if the free space on a DataNode falls below a certain threshold. In the event of a sudden high demand for a particular file, a scheme might dynamically create additional replicas and rebalance other data in the cluster. These types of data rebalancing schemes are not yet implemented.

3. Data Integrity

It is possible that a block of data fetched from a DataNode arrives corrupted. This corruption can occur because of faults in a storage device, network faults, or buggy software. The HDFS client software implements checksum checking on the contents of HDFS files. When a client creates an HDFS file, it computes a checksum of each block of the file and stores these checksums in a separate hidden file in the same HDFS namespace. When a client retrieves file contents it verifies that the data it received from each DataNode matches the checksum stored in the associated checksum file. If not, then the client can opt to retrieve that block from another DataNode that has a replica of that block.

Metadata Disk Failure

The FsImage and the EditLog are central data structures of HDFS. A corruption of these files can cause the HDFS instance to be non-functional. For this reason, the NameNode can be configured to support maintaining multiple copies of the FsImage and EditLog. Any update to either the FsImage or EditLog causes each of the FsImages and EditLogs to get updated synchronously. This synchronous updating of multiple copies of the FsImage and EditLog may degrade the rate of namespace transactions per second that a NameNode can support. However, this degradation is acceptable because even though HDFS applications are very data intensive in nature, they are not metadata intensive. When a NameNode restarts, it selects the latest consistent FsImage and EditLog to use.

The NameNode machine is a single point of failure for an HDFS cluster. If the NameNode machine fails, manual intervention is necessary. Currently, automatic restart and failover of the NameNode software to another machine is not supported.

Snapshots

Snapshots support storing a copy of data at a particular instant of time. One usage of the snapshot feature may be to roll back a corrupted HDFS instance to a previously known good point in time. HDFS does not currently support snapshots but will in a future release.

Data Organization

Data Blocks

HDFS supports write-once-read-many semantics on files. A typical block size used by HDFS is 64 MB. Thus, an HDFS file is chopped up into 64 MB chunks, and if possible, each chunk will reside on a different DataNode.

Staging

A client request to create a file does not reach the NameNode immediately. In fact, initially the HDFS client caches the file data into a temporary local file. Application writes are transparently redirected to this temporary local file. When the local file accumulates data worth over one HDFS block size, the client contacts the NameNode. The NameNode inserts the file name into the file system hierarchy and allocates a data block for it. The NameNode responds to the client request with the identity of the DataNode and the destination data block. Then the client flushes the block of data from the local temporary file to the specified DataNode. When a file is closed, the remaining un-flushed data in the temporary local file is transferred to the DataNode. The client then tells the NameNode that the file is closed. At this point, the NameNode commits the file creation operation into a persistent store. If the NameNode dies before the file is closed, the file is lost

Replication Pipelining

When a client is writing data to an HDFS file, its data is first written to a local file as explained in the previous section. Suppose the HDFS file has a replication factor of three. When the local file accumulates a full block of user data, the client retrieves a list of DataNodes from the NameNode. This list contains the DataNodes that will host a replica of that block. The client then flushes the data block to the first DataNode. The first DataNode starts receiving the data in small portions, writes each portion to its local repository and transfers that portion to the second DataNode in the list. The second DataNode, in turn starts receiving each portion of the data block, writes that portion to its repository and then flushes that portion to the third DataNode. Finally, the third DataNode writes the data to its local repository. Thus, a DataNode can be receiving data from the previous one in the pipeline and at the same time forwarding data to the next one in the pipeline. Thus, the data is pipelined from one DataNode to the next.

Accessibility

Natively, HDFS provides a FileSystem Java API for applications to use.

In addition, an HTTP browser can also be used to browse the files of an HDFS instance.

FS Shell

HDFS allows user data to be organized in the form of files and directories. It provides a commandline interface called FS shell that lets a user interact with the data in HDFS.

Browser Interface

A typical HDFS install configures a web server to expose the HDFS namespace through a configurable TCP port. This allows a user to navigate the HDFS namespace and view the contents of its files using a web browser.

Space Reclamation

File Deletes and Undeletes

When a file is deleted by a user or an application, it is not immediately removed from HDFS. Instead, HDFS first renames it to a file in the /trash directory. The file can be restored quickly as long as it remains in /trash. A file remains in /trash for a configurable amount of time. After the expiry of its life in /trash, the NameNode deletes the file from the HDFS namespace. The deletion of a file causes the blocks associated with the file to be freed. Note that there could be an appreciable time delay between the time a file is deleted by a user and the time of the corresponding increase in free space in HDFS.

A user can Undelete a file after deleting it as long as it remains in the /trash directory. If a user wants to undelete a file that he/she has deleted, he/she can navigate the /trash directory and retrieve the file. The /trash directory contains only the latest copy of the file that was deleted. The /trash directory is just like any other directory with one special feature: HDFS applies specified policies to automatically delete files from this directory. Current default trash interval is set to 0 (Deletes file without storing in trash). This value is configurable parameter stored as fs.trash.interval stored in core-site.xml.

Decrease Replication Factor

When the replication factor of a file is reduced, the NameNode selects excess replicas that can be deleted. The next Heartbeat transfers this information to the DataNode. The DataNode then removes the corresponding blocks and the corresponding free space appears in the cluster. Once again, there might be a time delay between the completion of the setReplication API call and the appearance of free space in the cluster.