ARROW-1594: [Python] Multithreaded conversions to Arrow in from_pandas

[wesm]

This results in nice speedups when column conversions do not require GIL to be held:

In [5]: import numpy as np

In [6]: import pandas as pd

In [7]: import pyarrow as pa

In [8]: NROWS = 1000000

In [9]: NCOLS = 50

In [10]: arr = np.random.randn(NCOLS, NROWS).T

In [11]: arr[::5] = np.nan

In [12]: df = pd.DataFrame(arr)

In [13]: %timeit rb = pa.RecordBatch.from_pandas(df, nthreads=1)
10 loops, best of 3: 179 ms per loop

In [14]: %timeit rb = pa.RecordBatch.from_pandas(df, nthreads=4)
10 loops, best of 3: 59.7 ms per loop

This introduces a dependency on the futures Python 2.7 backport of concurrent.futures (PSF license)

[cpcloud]

Nice perf improvement! LGTM, small comments.

[cpcloud]

Is it much slower to just use this code path?

[wesm]

I doubt it, it was more the principle of starting a thread pool for no reason. I'd be fine with this being the only code path

[wesm]

The thread pool seems to have some non-trivial fixed overhead, at least 20 microseconds per task on my machine. I did some quick testing and when the number of columns is "large" relative to the number of rows, then using a thread pool is slower than single thread. This suggests we should use some heuristic to decide whether to use the thread pool to avoid bad performance in wide tables without a ton of rows.

For example

NROWS = 10000
NCOLS = 500

arr = np.random.randn(NCOLS, NROWS).T
arr[::5] = np.nan

df = pd.DataFrame(arr)

%timeit sdf = pa.serialize_pandas(df, nthreads=1)
10 loops, best of 3: 62.8 ms per loop

%timeit sdf = pa.serialize_pandas(df, nthreads=2)
10 loops, best of 3: 83.4 ms per loop

[wesm]

Added a rough heuristic to turn off the thread pool if number of rows is less than 100 times the number of columns

[cpcloud]

What does this do when running under Python 3?

[wesm]

It's basically a no-op because concurrent.futures is part of the py3 standard library

[cpcloud]

Cool, just curious.

[wesm]

I was wrong, this should not be installed in py3. Sorting out a fix

[wesm]

I threaded (... sorry) this parameter through serialize_pandas, too. I also changed deserialize_pandas to use the same default for nthreads as to_pandas for consistency

[wesm]

I can add a unit test for this tomorrow

[wesm]

+1