JuliaStats/DataFrames.jl — pull request #344 by tanmaykm (2013-08-09 12:31:14+00:00)
Title: WIP: Distributed DataFrames and Blocks interface

This PR would add:
- An interface to the Blocks framework (https://github.com/tanmaykm/Blocks.jl) for processing chunks of data as DataFrames
- A distributed DataFrame implementation using the above

This is still WIP and some features are not implemented / optimized yet. I would work further on this if this looks fine in general.

### The Blocks interface
Blocks framework enables chunks of files, streams, HDFS file (or any other source that may provide a Blocks interface) can be processed using DataFrames. For example, blocks of an HDFS file can be processed using dataframes as:
````
b = Blocks(HdfsUrl("http://host/path/to/hdfs/file")) |> as_recordio |> as_bufferedio | as_dataframe
nrows = mapreduce(nrow, +, b)
````
Here `nrow` and `+` can be replaced with other non trivial methods to do more useful computation. And `mapreduce` can be replaced with `pmap` or `pmapreduce` to do this process in a distributed manner where processing happens on the node where the HDFS data block resides.

The `as_dataframe` API has been exported to be able to integrate with Blocks framework:
````
as_dataframe(io::BlockableIO; kwargs…)
where io is a data stream of type IOStream/AsyncStream/IOBuffer or BlockIO
kwargs are keyword arguments that are usually passed to `readtable`
````

### Distributed DataFrame (DDataFrame)
The above integration with Blocks is also leveraged to provide a distributed DataFrame type `DDataFrame`. A `DDataFrame` represents a large DataFrame distributed across processors and provides regular DataFrame operations on them as a whole.

The following APIs are available on `DDataFrame` that provide similar functionality as they do on a `DataFrame`:
````
describe, getindex, colmins, colmaxs, colprods, colsums, colmeans,
nrow, ncol, colnames, colnames!, clean_colnames!, rename, rename!, index,
head, tail, vcat, hcat, rbind, cbind, copy, deepcopy, isfinite, isnan, isna,
all, any, isequal, without, delete!, deleterows!, with, within!, complete_cases, complete_cases!,
merge, colwise, by
````

The following `DDataFrame` APIs are for reading data as a `DDataFrame` or writing data out from a `DDataFrame`
- `dreadtable`: to read data in as a distributed dataframe. Possible sources are files, io streams, HDFS or any other source that integrates with Blocks.
- `dwritetable`, `writetable`: to write data out into files
- `gather`: to gather pieces of a distributed dataframe into a single dataframe

I would create the API docs once the interface is accepted and am happy to incorporate any comments or suggestions. There are a few examples here (https://github.com/tanmaykm/HDFS.jl/blob/master/test/blocks_test.jl) and here (https://github.com/tanmaykm/DataFrames.jl/tree/ddataframe/test/test_ddataframe.jl).