Fast Parquet row count in Spark

Question

The Parquet files contain a per-block row count field. Spark seems to read it at some point (SpecificParquetRecordReaderBase.java#L151).

I tried this in spark-shell:

sqlContext.read.load("x.parquet").count

And Spark ran two stages, showing various aggregation steps in the DAG. I figure this means it reads through the file normally instead of using the row counts. (I could be wrong.)

The question is: Is Spark already using the row count fields when I run count? Is there another API to use those fields? Is relying on those fields a bad idea for some reason?

Answer 1

That is correct, Spark is already using the rowcounts field when you are running count.

Diving into the details a bit, the SpecificParquetRecordReaderBase.java references the Improve Parquet scan performance when using flat schemas commit as part of [SPARK-11787] Speed up parquet reader for flat schemas. Note, this commit was included as part of the Spark 1.6 branch.

If the query is a row count, it pretty much works the way you described it (i.e. reading the metadata). If the predicates are fully satisfied by the min/max values, that should work as well though that is not as fully verified. It's not a bad idea to use those Parquet fields but as implied in the previous statement, the key issue is to ensure that the predicate filtering matches the metadata so you are doing an accurate count.

To help understand why there are two stages, here's the DAG created when running the count() statement.

When digging into the two stages, notice that the first one (Stage 25) is running the file scan while the second stage (Stage 26) runs the shuffle for the count.

Thanks to Nong Li (the author of the SpecificParquetRecordReaderBase.java commit) for validating!

 

Updated

To provide additional context on the bridge between Dataset.count and Parquet, the flow of the internal logic surrounding this is:

• Spark does not read any Parquet columns to calculate the count
• Passing of the Parquet schema to the VectorizedParquetRecordReader is actually an empty Parquet message
• Computing the count using the metadata stored in the Parquet file footers. involves the wrapping of the above within an iterator that returns an InternalRow per InternalRow.scala.

To work with the Parquet File format, internally, Apache Spark wraps the logic with an iterator that returns an InternalRow; more information can be found in InternalRow.scala. Ultimately, the count() aggregate function interacts with the underlying Parquet data source using this iterator. BTW, this is true for both vectorized and non-vectorized Parquet reader.

Therefore, to bridge the Dataset.count() with the Parquet reader, the path is:

• The Dataset.count() call is planned into an aggregate operator with a single count() aggregate function.
• Java code is generated at planning time for the aggregate operator as well as the count() aggregate function.
• The generated Java code interacts with the underlying data source ParquetFileFormat with an RecordReaderIterator, which is used internally by the Spark data source API.

For more information, please refer to Parquet Count Metadata Explanation.