[Python] Support reading Parquet datasets with standard partition directory schemes

[asfimport]

Currently, we only support multi-file directories with a flat structure (non-partitioned).

Reporter: Wes McKinney / @wesm
Assignee: Wes McKinney / @wesm

Related issues:

• Arrow 0.3 release (blocks)

Original Issue Attachments:

• partitioned_parquet.tar.gz

_{Note: This issue was originally created as ARROW-539. Please see the migration documentation for further details.}

[asfimport]

Miki Tebeka / @tebeka:
@wesm Do you have an example of such directory with files? If not - easy way to generate one?

[asfimport]

Wes McKinney / @wesm:
I recommend either using Spark or Impala + Ibis to generate one, here's a docker image you can pull to run Impala:

https://github.com/cloudera/ibis/blob/master/circle.yml#L43

Here's some examples of creating partitioned tables in Impala+HDFS with Ibis:

https://github.com/cloudera/ibis/blob/master/ibis/impala/tests/test_partition.py#L58

Let me generate a quick example tarball to attach to this JIRA

[asfimport]

Wes McKinney / @wesm:
Making partitioned tables with Hive or Impala is pretty difficult, here was the code I used to make one

import ibis
import pandas as pd
import hdfs
hdfs = ibis.hdfs_connect('localhost', port=5070)
con = ibis.impala.connect('localhost', port=21050, hdfs_client=hdfs)

df = pd.DataFrame({'year': [2009, 2009, 2009, 2010, 2010, 2010],
                   'month': ['1', '2', '3', '1', '2', '3'],
                   'value': [1, 2, 3, 4, 5, 6]})
df = pd.concat([df] * 10, ignore_index=True)

con.create_database('temp_partition', path='/tmp/my_db')
con.create_table('unpartitioned', df, database='temp_partition')

db = con.database('temp_partition')
unpart_t = db.table('unpartitioned')
part_keys = ['year', 'month']
unique_keys = df[part_keys].drop_duplicates()

con.create_table('partitioned', schema=unpart_t.schema(), 
                 database='temp_partition', partition=part_keys)
part_t = db.table('partitioned')

for i, (year, month) in enumerate(unique_keys.itertuples(index=False)):
    select_stmt = unpart_t[(unpart_t.year == year) &
                           (unpart_t.month == month)]

    part = {'year': year, 'month': month}
    part_t.insert(select_stmt, partition=part)

Now we have

>>> hdfs.ls('/tmp/my_db/partitioned')
['_impala_insert_staging', 'year=2009', 'year=2010']

>>> hdfs.ls('/tmp/my_db/partitioned/year=2009')
['month=1', 'month=2', 'month=3']

Finally I ran

hdfs.get('/tmp/my_db/partitioned', 'partitioned_parquet')

to download from HDFS. see attached tarball

[asfimport]

Miki Tebeka / @tebeka:
Thanks @wesm

[asfimport]

Miki Tebeka / @tebeka:
We can either do it in the arrow level and return a table with extra fields generated from the directory structure or we can do it in the Pandas level, read only the values from the parquet files and then generate columns for the DataFrame from the directory structure.

Both cases we'll need to guess the type of the fields from the directory structure. Unless there are metadata files.

Which is better?

[asfimport]

Jeff Knupp / @jeffknupp:
Wouldn't we need to do this at the Arrow level so all non-Pandas consumers of pyarrow can still read these files?

[asfimport]

Wes McKinney / @wesm:
Yes – I think the most performant / robust option will be to generate DictionaryArray fields from the partition keys.

For example, if we have 3 partitions with the keys "a", "b", and "c", then we will a pyarrow.Table from each file and add DictionaryArray columns for the partition keys. We have to determine all the partition keys up front so that we can produce correct dictionary metadata, so it might be that

a -> 0
b -> 1
c -> 2

So in the first table for partition "a", the dictionary indices are all 0. But we can concatenate and then convert to pandas.Categorical at the end

[asfimport]

Miki Tebeka / @tebeka:
Moving an email conversation to here (where it belongs).

@tebeka said:

I'm working on ARROW-539 (see https://github.com/tebeka/arrow/compare/master...ARROW-539)

The code is almost done (sans testing). The issue I'm facing is in _add_parts (line 246). The first for loop (line 251) adds Column while the 2nd for loop (line 255) adds Array. This causes Table.from_arrays to complain. I've looks for a way to convert an Array to Column (or the other way around) and didn't find anything obvious. I can of course convert to Python list and back but this is wasteful.

Any pointers on how to do this?

(Later I'll convert the StringArray to a Dictionary as you suggested, starting simple ...)

And @wesm answered

Looks like _schema_from_arrays must be refactored to have the type
check inside the for loop like

https://github.com/apache/arrow/blob/master/python/pyarrow/table.pyx#L684

Unfortunately, I don't think this is going to work (for performance reasons):

arrays.append(from_pylist([parts[name]] * size))

https://github.com/tebeka/arrow/compare/master...ARROW-539#diff-211255c3faff2b62de57e1dca9f60e13R256

I think you're going to need to create a function (in C++) like

type = int32()
arr = array_from_constant(type, 0)

You can then use these integers to make a DictionaryArray using
something like DictionaryArray.from_arrays
https://github.com/apache/arrow/blob/master/python/pyarrow/array.pyx#L406

Doesn't look like that handles Arrow arrays as inputs. Probably this
deserves a new API that looks like
DictionaryArray.from_indices(indices, type).

New JIRA for this

https://issues.apache.org/jira/browse/ARROW-643

[asfimport]

Wes McKinney / @wesm:
I'm marking this for 0.3 and will try to get a patch up this week

[asfimport]

Wes McKinney / @wesm:
PR: #529

[asfimport]

Wes McKinney / @wesm:
Issue resolved by pull request 529
#529