Hadoop application packaging

Job jar must be packaged as below;

job.jar
|--META-INF
|----MANIFEST.INF
|------Main-Class: x.y.z.Main
|--lib
|---- commons-lang.jar Note: Place your dependent jars inside lib directory
|--org.zero
|---- application classes here

Archiving large number of small files into small number of large files

A small file is one which is significantly smaller than the HDFS block size (default 64MB).

We have a lot of data feeds in the range of 2MB per day, storing each as a separate file is non-optimal.

The problem is that HDFS can't handle lots of files, because, every file, directory and block in HDFS is represented as an object in the namenode's memory, each of which occupies 150 bytes. So for 10 million files, each using a block, would use about 3 gigabytes of memory. Scaling up much beyond this level is a problem with current hardware. Certainly a billion files is not feasible.

Furthermore, HDFS is not geared up to efficiently accessing small files: it is primarily designed for streaming access of large files. Reading through small files normally causes lots of seeks and lots of hopping from datanode to datanode to retrieve each small file, all of which is an inefficient data access pattern.

Also, HDFS does not supports appends (follow http://www.cloudera.com/blog/2009/07/file-appends-in-hdfs/).

Known options are;

1. Load data to Hbase table and periodically export them to files for long term storage. Some thing like we have product log for a particular date/timestamp against the content of the files stored as plain text in Hbase table.
2. Alternatively, we can treat these files as pieces of the larger logical file and incrementally consolidate additions to a newer file. That is, file x was archived on day zero, the next day new records are available to be archived. We will rename the existing file to let's say x.bkp and then execute a mapreduce job to read the content from the exiting file and the new file to the file x.
3. Apache Chukwa solves the similar problem of distributed data collection and archival for log processing. We can also take inspiration from their and provide our custom solution to suit our requirements, if needed.

Cascading: How does cascading decides which fields should go to a column family?

I was playing with Cascading code sample as given here.

Problem Statement: let's say we have three fields in a tuple for e.g.

line_num, lower, upper, double
1, a, A, AA

and I wish to add double to its own column family or lets say club it with an existing column family 'right' How do I do that.

Solution:

String tableName = "DataLoadTable";
        Fields keyFields = new Fields("line_num");

// add a new family name
        String[] familyNames = new String[] { "left", "right", "double" };

// group your fields together in the order in which you would like them to be 

// added to column families
        Fields[] valueFields = new Fields[] { new Fields("lower"),
                new Fields("upper"), new Fields("double") };
        HBaseScheme hbaseScheme = new HBaseScheme(keyFields, familyNames,
                valueFields);
        Tap sink = new HBaseTap(tableName, hbaseScheme, SinkMode.REPLACE);

// describe your tuple entry: add the new field
        Fields fieldDeclaration = new Fields("line_num", "lower", "upper",
                "double");
        Function function = new RegexSplitter(fieldDeclaration, ", ");

The remaining of the code remains the same as given in the example.

Either, the above was too obvious that the authors didn't talked about it in the user guide or I do not know how to describe the problem and hence was not able to find them.

Let me know if I'm wrong.

How does the data flows when a job is submitted to Hadoop?

Based on the discussion here, typically the data flow is like this:

1. Client submits a job description to the JobTracker. 
2. JobTracker figures out block locations for the input file(s) by talking to HDFS NameNode. 
3. JobTracker creates a job description file in HDFS which will be read by the nodes to copy over the job's code etc. 
4. JobTracker starts map tasks on the slaves (TaskTrackers) with the appropriate data blocks. 
5. After running, maps create intermediate output files on those slaves. These are not in HDFS, they're in some temporary storage used by MapReduce. 
6. JobTracker starts reduces on a series of slaves, which copy over the appropriate map outputs, apply the reduce function, and write the outputs to HDFS (one output file per reducer). 
7. Some logs for the job may also be put into HDFS by the JobTracker.

However, there is a big caveat, which is that the map and reduce tasks run arbitrary code. It is not unusual to have a map that opens a second HDFS file to read some information (e.g. for doing a join of a small table against a big file). If you use Hadoop Streaming or Pipes to write a job in Python, Ruby, C, etc, then you are launching arbitrary processes which may also access external resources in this manner. Some people also read/write to DBs (e.g. MySQL) from their tasks.