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/*.

中的文件

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

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

解决方案：使用 Parquet 文件

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

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());

重要

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

Answer 1

因此，在 Spark 中，您可以按照自己想要的方式保存和读取分区数据。但是，当您使用 /customer/a/date/b/product/c 保存数据时，Spark 将使用此约定 /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 使用谓词下推，因此它只会将符合条件（由某些过滤器转换指定）的数据实际加载到内存中。但是如果你真的想明确指定范围，你可以生成一个路径名列表，然后将它传递给读取函数。像这样：

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:_*)

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

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

Apache Spark: Using folder structures to reduce run-time of analyses

wildcard

hdfs

apache-spark

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

Apache Spark: Using folder structures to reduce run-time of analyses

wildcard

hdfs

apache-spark

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