Apache Spark：使用文件夹结构来减少分析的运行时间

Question

我想通过将一个巨大的 csv 文件细分为不同的分区来优化 Spark 应用程序的运行时间，具体取决于它们的特性。

例如我有一列带有客户 ID（整数，a），一列带有日期（月 + 年，例如 01.2015，b），以及带有产品 ID（整数，c）的列（以及更多带有产品特定数据的列，不需要用于分区）。

我想建立一个像/customer/a/date/b/product/c 这样的文件夹结构。当用户想了解客户 X 于 2016 年 1 月售出的产品信息时，他可以加载并分析保存在/customer/X/date/01.2016/* 中的文件。

是否可以通过通配符加载此类文件夹结构？还应该可以加载特定时间范围内的所有客户或产品，例如01.2015 至 09.2015。是否可以使用像/customer/*/date/*.2015/product/c 这样的通配符？或者这样的问题怎么解决？

我想对数据进行一次分区，然后在分析中加​​载特定文件，以减少这些作业的运行时间（忽略分区的额外工作）。

解决方案：使用 Parquet 文件

我更改了我的 Spark 应用程序以将我的数据保存到 Parquet 文件中，现在一切正常，我可以通过提供文件夹结构来预先选择数据。这是我的代码sn-p：

JavaRDD<Article> goodRdd = ...

SQLContext sqlContext = new SQLContext(sc);

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

StructType schema = DataTypes.createStructType(fields);

JavaRDD<Row> rowRDD = goodRdd.map(new Function<Article, Row>() {
    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());

重要

不要尝试只有一列的表格！当您尝试调用partitionBy 方法时，您将得到异常！

Answer 1

因此，在 Spark 中，您可以按照自己的方式保存和读取分区数据。但是，当您使用以下方式保存数据时，Spark 不会像 /customer/a/date/b/product/c 那样创建路径，而是使用此约定 /customer=a/date=b/product=c：

df.write.partitionBy("customer", "date", "product").parquet("/my/base/path/")

当需要读入数据时，需要像这样指定basepath-option：

sqlContext.read.option("basePath", "/my/base/path/").parquet("/my/base/path/customer=*/date=*.2015/product=*/")

/my/base/path/ 之后的所有内容都将被 Spark 解释为列。在此处给出的示例中，Spark 会将三列 customer、date 和 product 添加到数据帧中。请注意，您可以根据需要对任何列使用通配符。

至于在特定时间范围内读取数据，您应该知道 Spark 使用谓词下推，因此它只会将数据实际加载到符合条件的内存中（由某些过滤器转换指定）。但是如果你真的想明确指定范围，你可以生成一个路径名列表，然后将它传递给 read 函数。像这样：

val pathsInMyRange = List("/my/path/customer=*/date=01.2015/product=*", 
                          "/my/path/customer=*/date=02.2015/product=*", 
                          "/my/path/customer=*/date=03.2015/product=*"...,
                          "/my/path/customer=*/date=09.2015/product=*")

sqlContext.read.option("basePath", "/my/base/path/").parquet(pathsInMyRange:_*)

无论如何，我希望这会有所帮助:)