ENH: create out-of-core processing module

[jreback]

Conceptually create a pipeline processor that performs out-of-core computation.
This is easily parallelizable (multi-core or machines), in theory cython / ipython / joblib / hadoop could operate with this

requirements

the data set must support chunking, and the function must operate only on that chunk
Useful in cases of a large number of rows, or a problem that you want to parrallelize.

input

a chunking iterator that reads from disk (could take chunksize parameters,
a handle and just call the iterators as well)

• read_csv
• HDFStore

function

take an iterated chunk, and an axis and return another pandas object
(could be a reduction, transformation, whatever)

output

an output mechanism to take the function application, must support appending

• to_csv (with appending)
• HDFStore (table)
• another pipeline
• in memory

Map-reduce is an example of this type of pipelining.

Interesting Library
https://github.com/enthought/distarray

[gravesee]

This is the functionality that pandas currently lacks that is preventing me from ditching SAS. This is a monumental task but so worth the effort! I'm very happy to see it being discussed. Is there any possibility of collaborating with this project? http://continuum.io/blog/blaze

[wesm]

The timeline for when something production-ready is going to ship there is unclear but worth keeping an eye on for collaboration possibilities.

[jreback]

@Zelazny7 could u put up some example calculations?
obviously doesn't have to be very big and can use random data

[gravesee]

For example, finding the within group sum for a dataframe could be broken up iteratively like so. This is all done in memory, but the same concept could be applied to a file stored on disk and read-in piece-meal. The purpose of this enhancement, I hope, is to make the management of the IO and chunking transparent to the user:

def gen(df, chunksize=50000):
    nrows = len(df)
    current = 0
    while current < nrows:
        yield df[current:current+chunksize]
        current += chunksize

res = pd.concat([chunk.groupby('YearMade').sum() for chunk in gen(df)])
res.groupby(res.index).sum().head()

          SalePrice
YearMade
1000      781142325
1919        2210750
1920         493250
1937          18000
1942          51000

This is equivalent to:

df.groupby('YearMade').sum().head()
          SalePrice
YearMade
1000      781142325
1919        2210750
1920         493250
1937          18000
1942          51000

I tried using reduce and combine functions to achieve a similar result, but I don't know enough pandas/python to understand why it was failing. Interestingly, a time comparison of the two results shows the method using a generator is slightly faster:

def test():
    res = pd.concat([chunk.groupby('YearMade').sum() for chunk in gen(df)])
    res.groupby(res.index).sum()

In [1]: %timeit test()
10 loops, best of 3: 31.2 ms per loop

In [2]: %timeit df.groupby('YearMade').sum()
10 loops, best of 3: 32.8 ms per loop

There is a broad class of calculations that can be performed on chunks of the data and the results collated together. The sum example shown above is one such case. This can easily be extended to the mean, variance and standard deviation. Median's and quantiles are a bit trickier and certain kinds of regression would have to fall back on gradient descent algorithms rather than close-form solutions.

Edited to add that scikit-learn has several algorithms that implement the partial_fit method. These functions are designed to work on chunks of data like this.

BTW, I pulled the data from Kaggle: http://www.kaggle.com/c/bluebook-for-bulldozers

[jreback]

@Zelazny7

I think we could start off with an API something like this (pseudo codeish here):

Create pipes

pipe = pipeline.Pipe(handle_or_iterator_or_pandas_object, 
                                 func, axis, name, chunksize=50k, 
                                  method='single_core', location=dir_to_save_data,
                                  combine=None, partition=None))
  handle_or_iterator_or_object : would be either an opened read_csv/HDFStore iterator
            (or could open/close it for you when done), or a pandas-object (e.g. a Frame)
  func : a function taking a single argument, plus passed kwargs to evaluate
            can return a similarly sized object (e.g. a map), or a reduction
  axis : the slicing axis (of how we partition)
  name : this is necessary to save any intermediate data (in our location), suppose this could be optional
              and just save in temp dirs of the location (which itself could be a temp dir)
  chunksize : the size of a particular chunk
  partition : an optional partitioning function (in lieu of chunksize / axis)
  method : how to execute this evaluation, where by on a chunk-by-chunk basis (single_core),
                 multi-process (multi_core), or distributed (TBD), this possibily could be
                 instead set by the class (e.g. Pipe is the base, SingleCore, MultiCore, Distributed)
  location : if saving data (e.g. in a map/transform case), need to save each chunk
  combine : a combiner function (e.g. trivial in the case of sum, but could be somewhat complicated,
                   maybe do this as a separate pipe)

Eval a pipe (and return a new pipe of this computation)

returned_pipe = pipe.eval(**kwargs)

Get results

df = pipe.get_results()
pipe.write_results(output_handle)

Cleanup (delete temporary files), how to make this cleanup nicely

pipe.clean()

the reason for the eval and get_results would be to allow chaining and potentially lazy evaluation
one could call get_results I think with the same args as eval (which in effect just uses a single pipe)

I think could support methods of single_core, and probably multi_core using the multi-processing module

so I think that we could right your computation like this

pipeline.Pipe(df).get_results(lambda x: x.groupby('YearMade').sum())

if df happens to be on disk

pipeline.Pipe('cool.csv').get_results(lambda x: x.groupby('YearMade').sum())

Alternatively

2. could have separte methods map (or apply) and reduce for evaluation
3. could support specific methods e.g. groupby, sum, mean

Could do some of all three

We usually use apply for a map type operation (and reduction for that matter)
eval is currently not really used in pandas (but maybe it makes sense to introduce it
to signal that we are doing something special here

any suggestions for names, pipeline just sort of random....

[jreback]

This probably also should be done with a decorator, mainly to facilitate open/close context of the file
I think I have seen syntax similar to this in celery (distributed task queues)

http://docs.celeryproject.org/en/latest/getting-started/introduction.html

@pipeline.Pipeline(chunksize=50k, method='single_core')
 def func(x):
      return x.groupby('YearMade').sum()

Then this will work (I think)

df = func('cool.csv')

[jreback]

having read_csv provide start/stop parameters so we can read a specific chunk from disk, see #3221

Pytables evaluator, can provide simple computations thru this directly

• http://pytables.github.com/usersguide/libref/expr_class.html
• http://www.pytables.org/moin/ComputingKernel

numpy w/multi-processing

• http://folk.uio.no/sturlamo/python/multiprocessing-tutorial.pdf
• http://www.astropython.org/snippet/2010/3/Parallel-map-using-multiprocessing

things to take note:

• the case of an already in-memory Frame is different from an on-disk Frame
• single core vs multi-core
• like using numexpr, you only want to use these strategies with minimum of rows (at least 50k),
  although should prove a method (useful in testing) to force it
• dynamic vs specific slice chunking/scheduling (prob don't want to get too complicated here and its hard to know how the function is going to be used in any event)
• write_results HAS to be serialized in the parent process! though can have intermediate results written, then a combiner could deal with see
  • Pytables using mmap / Pipe of Multiprocess: https://github.com/PyTables/PyTables/blob/develop/examples/multiprocess_access_benchmarks.py
  • Pytables collate results: https://github.com/PyTables/PyTables/blob/develop/examples/multiprocess_access_queues.py
• Multiprocessing with numpy
  • http://stackoverflow.com/questions/15414027/multiprocessing-pool-makes-numpy-matrix-multiplication-slower
  • http://stackoverflow.com/questions/5033799/how-do-i-pass-large-numpy-arrays-between-python-subprocesses-without-saving-to-d

mclapply in R

• http://www.rforge.net/doc/packages/multicore/mclapply.html

Creating shared memory with multiprocessing

mp_arr = multiprocessing.Array(ctypes.c_double, 100)
arr = np.frombuffer(mp_arr.get_obj())

from Here: http://mail.scipy.org/pipermail/scipy-user/2013-April/034415.html

Another good Ref
http://stackoverflow.com/questions/15976937/making-my-numpy-array-shared-across-processes

[jreback]

answered in this question:
http://stackoverflow.com/questions/15798209/pandas-group-by-query-on-large-data-in-hdfstore

2 methods for groupby

import numpy as np
import pandas as pd
import os

fname = 'groupby.h5'

# create a frame
df = pd.DataFrame({'A': ['foo', 'foo', 'foo', 'foo',
                         'bar', 'bar', 'bar', 'bar',
                         'foo', 'foo', 'foo'],
                   'B': ['one', 'one', 'one', 'two',
                         'one', 'one', 'one', 'two',
                         'two', 'two', 'one'],
                   'C': ['dull', 'dull', 'shiny', 'dull',
                         'dull', 'shiny', 'shiny', 'dull',
                         'shiny', 'shiny', 'shiny'],
                   'D': np.random.randn(11),
                   'E': np.random.randn(11),
                   'F': np.random.randn(11)})


# create the store and append, using data_columns where I possibily
# could aggregate
df.to_hdf(fname,'df',mode='w',table=True,data_columns=['A','B','C'])

# method 1
with pd.get_store(fname) as store:

    # get the groups
    groups = store.select_column('df','A').unique()

    # iterate over the groups and apply my operations
    l = []
    for g in groups:

        # select the group
        grp = store.select('df',where = [ 'A=%s' % g ])

        # this is a regular frame, aggregate however you would like
        l.append(grp[['D','E','F']].sum())

    print "\nresult (1):\n%s" % pd.concat(l, keys = groups)

# method 2 (only going to work for sum; for mean need to track the running count as well)
with pd.get_store(fname) as store:

    l = []
    for chunk in store.select('df', chunksize=3):
        l.append(chunk.groupby('A').sum())

    print "\nresult (2):\n%s" % pd.concat(l).reset_index().groupby('A').sum().stack()

[gravesee]

Functionality like the ff package from R would go a long ways towards making pandas easier to use with large-ish data: http://cran.r-project.org/web/packages/ff/index.html

HDFStore is great, but not being able to transparently select columns and, more importantly, append new columns makes it difficult to use efficiently. ff essentially stores a dataframe on disk. Users can bring portions of the dataframe into memory using the same syntax as an in-memory dataframe. This is why I call it transparent. Furthermore, it is very easy to "write" a new column by using the same syntax to assign a new column to an in-memory dataframe.

This functionality alone - being able to access elements of an on-disk dataframe - would be a huge win for pandas and could be implemented more quickly than the sequential processes discussed above. I think this would be a good stepping stone on the way towards out-of-core computations. I think it's reasonable for users to pull subsets of their data into memory for manipulation. It becomes a pain when we have to manage the i/o operations. This would alleviate that pain-point.

[jreback]

you can now select columns easily read_column (in 0.11 - docs are not built yet)

adding columns efficiently is quite difficult, you would need to store data column wise, and then adding rows would be difficult. You have to pick. (that said, we possibily could support a columnar access, via ctable), its not that different an interface, but it suffers from exactly the same problem.

That said you can easily add columns if you disaggregate your table, e.g. store a column per group.

The functionarily your are asking for here is bascially a numpy memmap, This tricky issue is exactly how you would use this, can you give me a pseudo-code example of how you would go about accessing this?

[gravesee]

In The predictive modeling world, think FICO and other credit scores, it's pretty much always a big table of heterogenous data. I would love to use pandas like this:

# read in a csv (or other text file) using awesome pandas parser, but store on disk
df_on_disk = pd.read_csv('somefile.csv', dtypes=dtypes, mode='disk')

# select columns into memory
col_in_mem = df_on_disk['col_that_I_need']

# multiple columns
cols_in_mem = df_on_disk[['col1','col2','col8']]

# writing to disk using column in memory
modified_col_in_mem = cool_function(col_in_mem)
df_on_disk['new_col'] = modified_col_in_mem

All of the cool indexing that pandas handles in-memory could be ignored for the on-disk dataframe. As long as it has the same number of rows, a new column would be slapped onto the end.

[gravesee]

I'm starting to understand the difficulty of quickly accessing and storing columns like this. To accomplish it using pd.read_csv, one loop for every column would have to be performed. Still, it would only have to be done once. The columns would have to be written to a mmap file in sequential order. Storing meta data about the columns and types would allow the mmap to be quickly accessed and return pandas objects. I think I"ll toy around with creating something like this and see how it goes.

[jreback]

If I remember how your original question on SO was, you just need to write the table (as is)) to HDFStore, with whatever data columnsyou want (or multiple tables if its very wide), then you just write columns to another group/groups as needed. There is some bookeeping in order to track what is where, so when you do a query you are pulling in the correct groups (and not selecting on certain ones), but this should be quite fast. And every once in a while you could rebuild the store into a new one (conceputally 'packing' it).

Its kind of like a distributed file system, you write and the system stores it where convenient. It has a map to tell you where everything is (kind of like a table of contents). The user of the system doesn't care exactly where the data is, just that they can do operations on it. Appends of rows are a bit more complicated because you have to split the row and append to multipel groups. Column appends are also easy. Even deletes of columns can be handled (just remember that column is deleted).

You ideally want locality of data when you select, e.g. if you typically want a bunch of columns at the same time, then store them in a table together.

I think could but a wrapper around this and make it pretty transparent. Wrap the selection mechnasim and it you could effectively have an out-of-core dataframe.

[jreback]

@Zelazny7 I lost the question where we were talking about column-wise tables...

you might find this link interesting
https://github.com/PyTables/proposal/blob/master/column-wise-pytables.rst

I actually think could implement this directly (and not wait for pytables to do it) via ctable