Apache Spark: Using folder structures to reduce run-time of analyses
<p>I want to optimize the run-time of a Spark application by subdividing a huge csv file into different partitions, dependent of their characteristics.</p>

<p>E.g. I have a column with customer ids (integer, a), a column with dates (month+year, e.g. 01.2015, b), and a column with product ids (integer, c) (and more columns with product specific data, not needed for the partitioning).</p>

<p>I want to build a folder structure like <code>/customer/a/date/b/product/c</code>. When a user wants to know information about products from customer X, sold in January 2016, he could load and analyse the file saved in <code>/customer/X/date/01.2016/*</code>.</p>

<p>Is there a possibility to load such folder structures via wildcards? It should also be possible to load all customer or products of an specific time range, e.g. 01.2015 till 09.2015. Is it possible to use wildcards like <code>/customer/*/date/*.2015/product/c</code>? Or how could a problem like this be solved?</p>

<p>I want to partition the data once, and later load the specific files in the analysis, to reduce the run-time for these jobs (disregarded the additional work for the partitioning).</p>

<p><strong>SOLUTION: Working with Parquet files</strong></p>

<p>I changed my Spark Application to save my data to Parquet files, now everything works fine and I can pre-select the data by giving folder-structure. Here my code snippet:</p>

<pre><code>JavaRDD&lt;Article&gt; goodRdd = ...

SQLContext sqlContext = new SQLContext(sc);

List&lt;StructField&gt; fields = new ArrayList&lt;StructField&gt;();
fields.add(DataTypes.createStructField("keyStore", DataTypes.IntegerType, false));
fields.add(DataTypes.createStructField("textArticle", DataTypes.StringType, false));

StructType schema = DataTypes.createStructType(fields);

JavaRDD&lt;Row&gt; rowRDD = goodRdd.map(new Function&lt;Article, Row&gt;() {
    public Row call(Article article) throws Exception {
        return RowFactory.create(article.getKeyStore(), article.getTextArticle());
    }
});

DataFrame storeDataFrame = sqlContext.createDataFrame(rowRDD, schema);

// WRITE PARQUET FILES
 storeDataFrame.write().partitionBy(fields.get(0).name()).parquet("hdfs://hdfs-master:8020/user/test/parquet/");

// READ PARQUET FILES
DataFrame read = sqlContext.read().option("basePath", "hdfs://hdfs-master:8020/user/test/parquet/").parquet("hdfs://hdfs-master:8020/user/test/parquet/keyStore=1/");

System.out.println("READ : " + read.count());
</code></pre>

<p><strong>IMPORTANT</strong></p>

<p>Don't try out with a table with only one column! You will get Exceptions when you try to call the <code>partitionBy</code> method!</p>