RevolutionAnalytics/RHadoop — Issue #62: possible performance problem with make.input.format()-generated parsers
Author: jeffreybreen
Created: 2012-03-06 21:06:01+00:00
State: open

I just wanted to capture my observations for future research.

I really, really like the new I/O stuff -- especially the common use of read.table() and the ability to use its features like col.names, etc.

So, building on @jseidman's "hadoop-r" examples, I wrote this parser to replace his csvtextinputformat():

```
col.names=c('Year','Month','DayofMonth','DayOfWeek','DepTime','CRSDepTime',
'ArrTime','CRSArrTime','UniqueCarrier','FlightNum','TailNum',
'ActualElapsedTime','CRSElapsedTime','AirTime','ArrDelay',
'DepDelay','Origin','Dest','Distance','TaxiIn','TaxiOut',
'Cancelled','CancellationCode','Diverted','CarrierDelay',
'WeatherDelay','NASDelay','SecurityDelay','LateAircraftDelay')

asa.csv.input.format = make.input.format('csv', 'text', sep=',',
col.names=col.names,
stringsAsFactors=F)
```

Naming the output columns allows for much more readable mapper code:

```
mapper.year.market.enroute_time = function(k, fields) {
# the make.input.format() reader treats the first field as the key
# whether you want that or not:
fields$Year = k

# Skip header lines, cancellations, and diversions:
if (!(identical(fields$Year, "Year")) &
fields$Cancelled==0 & fields$Diverted==0 ) {

# We don't care about direction of travel, so convert LAX-JFK to JFK-LAX
# sort alphabetically)
if (fields$Origin < fields$Dest)
market = paste(fields$Origin, fields$Dest, sep='-')
else
market = paste(fields$Dest, fields$Origin, sep='-')

# key consists of year, market
k = c(fields$Year, market)

keyval(k, fields$ActualElapsedTime)
}
}

```

The only problem is... when I ran this test on three of the ASA data files (1988.csv.bz2, 1998.csv.bz2, and 2008.csv.bz2), it took about 3 hours to get the first 33% status update and another hour to get to 100%:

```
12/03/05 14:23:54 INFO streaming.StreamJob: map 0% reduce 0%
12/03/05 17:29:23 INFO streaming.StreamJob: map 33% reduce 0%
12/03/05 17:29:42 INFO streaming.StreamJob: map 33% reduce 11%
12/03/05 17:38:19 INFO streaming.StreamJob: map 67% reduce 11%
12/03/05 17:38:31 INFO streaming.StreamJob: map 67% reduce 22%
12/03/05 18:35:10 INFO streaming.StreamJob: map 100% reduce 22%
```

This performance is in stark contrast with Jonathan's simple-but-elegant original function, appropriately wrapped for rmr 1.2:

```
csvtextinputformat = list(mode = 'text', format = function(line) {
keyval(NULL, unlist(strsplit(line, "\\,")))
}, streaming.format=NULL)

```

For the price of keeping track that fields[1] is the Year, etc., mapping hits the 33% mark in about half an hour and is 100% shortly thereafter:

```
12/03/05 22:31:29 INFO streaming.StreamJob: map 0% reduce 0%
12/03/05 23:00:50 INFO streaming.StreamJob: map 33% reduce 0%
12/03/05 23:01:10 INFO streaming.StreamJob: map 33% reduce 11%
12/03/05 23:01:43 INFO streaming.StreamJob: map 67% reduce 11%
12/03/05 23:02:04 INFO streaming.StreamJob: map 67% reduce 22%
12/03/05 23:10:29 INFO streaming.StreamJob: map 100% reduce 22%
```

These tests were all run on the same EC2 cluster with more m1.large nodes than I needed (since each mapper ripping through its own file on its own node.) Also -- I did not decompress the input files, so there was the bzip2 overhead in there too.

Thanks,
Jeffrey