Implement parquet simple record as a source for the druid spark indexer

[Igosuki]

Probably can't be merged right away.

Works fine with the parquet file I included (mapped the csv from test resources into a parquet file). The strategy is to directly map to json without going through avro as loading generic data made it more complicated than not for me.

I'm going to run production tests soon (vs a few hundreds of gigs of data).

[drcrallen]

For future maintainability, there is already a parquet input row parser in druid: https://github.com/druid-io/druid/blob/master/extensions-contrib/parquet-extensions/src/main/java/io/druid/data/input/parquet/ParquetHadoopInputRowParser.java is there any reason that cannot be used?

It would be nice if the row parser (or factory) could be passed in and eliminate the giant switch statement which is not really sustainable.

Also regarding #10 it might make sense to have an RDD passed in the index method in the future, though I don't think such a refactoring would be in scope for a specific new format.

Getting rid of the giant switch statement should be investigated more though.

[drcrallen]

test failure looks like a guava version problem:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 6.0 failed 1 times, most recent failure: Lost task 0.0 in stage 6.0 (TID 58, localhost, executor driver): java.lang.NoSuchMethodError: com.google.common.io.Files.asByteSource(Ljava/io/File;)Lcom/google/common/io/ByteSource;

Guava version reconciliation is always a huge issue.

[Igosuki]

Looks like checks have passed.
A few things :

• This uses the simplest read api from parquet which I copied over because parquet-tools which this api belongs to imports a very old version of guava
• I didn't add push down filters as an option, which can be very effective while doing parquet
• I fixed the druid version to 0.10.1 but we can revert that to snapshot
• I am unsure how hadoopFileAPI works with Spark/Parquet over a directory, meaning if it splits the ParquetFileReader workload properly or if it's a bottleneck.

[Gauravshah]

how would we specify the nested columns ?
also would it respect partition column ?

[Igosuki]

The point is that it transforms parquet records to json, so that you can use the json spec (including flatten). In 0.10.1 the avro flattening, nesting access isn't available yet afaik which is why I went with a custom parser that can easily transform to json without too much overhead.

…

On Wed, Oct 4, 2017, 1:56 PM Gaurav M Shah ***@***.***> wrote: how would we specify the nested columns ? also would it respect partition column ? — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#111 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAYyhiLZpG5OzSorXW-Yzt0-oDQVkEnIks5so3JkgaJpZM4PsLLC> .

[Gauravshah]

I feel going down dataframe route might be better.
Current way there are two transforms needed, Parquet to Json to Map<string,Object>
With dataframe router we could do Parquet to Map<String,Object>
Would allow us to do other transforms also would make use of parquet pruning features from spark

[Igosuki]

Sure, why not, let me have a look

[xanec]

Maybe we should not commit binary file (i.e., lineitems.parquet)? Could we generate this on-the-fly instead?

[xanec]

Just putting a note here as a reminder to correct this before merge.

[xanec]

There are a lot of repeated code (i.e., for parquet case and text case). I suggest that we can have some refactoring to combine them.

[xanec]

May I ask if prettyPrint is used (i.e., in production)? Or if it is only used for testing/development? If it is the latter, then perhaps we should not include it in the class.

And also, if we are to remove it, then a lot of the methods/classes can also be removed.

[xanec]

Please correct me if I am wrong but I think that the parser need not be "simple" in that the parsing can produce a POJO for our manipulation. Hence, it is up to us to decide what is the record that will be produced and how are we going to map it into druid.

[Igosuki]

@xanec You're right, I've had other things to patch on our end on Druid, I'm getting back to this as soon as I can to refactor it.

[Igosuki]

I rebased on 0.11 to use the transform and flatten APIs, but I had to copy maps in the parser because AbstractMaps in the ObjectFlatteners API aren't directly serializable by Kryo.