Parquet and predicate pushdown and dictionary compression

2
votes

I use the following code sequence

val obs = spark.read.parquet("s3://xxxx/parquettests/by_pat/")
val single = obs.filter($"pat" === "abcd")
single.explain

and I get back an execution plan with predicate pushdown

+- Filter (isnotnull(pat#930) && (pat#930 = abcd))
   +- FileScan parquet PartitionFilters: [], 
                       PushedFilters: [IsNotNull(pat), EqualTo(pat,abcd)]

The execution time of this (200 files with 3MB each, hash partitioned by "pat") is 32 secs.

Given that pat=abcd is contained in a single file only, that is 32 secs of reading a 3MB file plus skipping all others. Sounds a lot.

I have checked with the parquet-tools on of those files and it says PLAIN_DICTIONARY for the column, two pages both of "RLE:BIT_PACKED VLE:PLAIN [more] VC:21400".

Hence I suspect the dictionary compression to fall back to PLAIN but I am not sure and all my tries to change that did not work.

Thoughts?

1
apache-spark

1 Answers

0
votes

Predicate Push Down for dictionary encoded columns is currently not supported in Spark 2.2 and Parquet 1.8.
On top of that, your parquet files a just too small, you should merge them to 1 or 2 files.