Hi Michal, For (1), would it be possible to partitionBy two columns to reduce the size? Something like partitionBy("event_type", "date").
For (2), is there a way to separate the different event types upstream, like on different Kafka topics, and then process them separately? Xinh On Wed, May 4, 2016 at 7:47 AM, Michal Vince <vince.mic...@gmail.com> wrote: > Hi guys > > I`m trying to store kafka stream with ~5k events/s as efficiently as > possible in parquet format to hdfs. > > I can`t make any changes to kafka (belongs to 3rd party) > > > Events in kafka are in json format, but the problem is there are many > different event types (from different subsystems with different number of > fields, different size etc..) so it doesn`t make any sense to store them in > the same file > > > I was trying to read data to DF and then repartition it by event_type and > store > > events.write.partitionBy("event_type").format("parquet").mode(org.apache.spark.sql.SaveMode.Append).save(tmpFolder) > > which is quite fast but have 2 drawbacks that I`m aware of > > 1. output folder has only one partition which can be huge > > 2. all DFs created like this share the same schema, so even dfs with few > fields have tons of null fields > > > My second try is bit naive and really really slow (you can see why in > code) - filter DF by event type and store them temporarily as json (to get > rid of null fields) > > val event_types = events.select($"event_type").distinct().collect() // get > event_types in this batch > for (row <- event_types) { > val currDF = events.filter($"event_type" === row.get(0)) > val tmpPath = tmpFolder + row.get(0) > > currDF.write.format("json").mode(org.apache.spark.sql.SaveMode.Append).save(tmpPath) > sqlContext.read.json(tmpPath).write.format("parquet").save(basePath) > > }hdfs.delete(new Path(tmpFolder), true) > > > Do you have any suggestions for any better solution to this? > > thanks > > >