如何使用 hadoop 自定义输入格式调整 Spark 应用程序

Question

我的 spark 应用程序使用自定义 hadoop 输入格式处理文件（平均大小为 20 MB）并将结果存储在 HDFS 中。

下面是代码sn-p。

Configuration conf = new Configuration();


JavaPairRDD<Text, Text> baseRDD = ctx
    .newAPIHadoopFile(input, CustomInputFormat.class,Text.class, Text.class, conf);

JavaRDD<myClass> mapPartitionsRDD = baseRDD
    .mapPartitions(new FlatMapFunction<Iterator<Tuple2<Text, Text>>, myClass>() {
        //my logic goes here
    }

//few more translformations
result.saveAsTextFile(path);

此应用程序为每个文件创建 1 个任务/分区，并在 HDFS 中处理和存储相应的部分文件。

即，对于 10,000 个输入文件，会创建 10,000 个任务，并将 10,000 个部分文件存储在 HDFS 中。

baseRDD 上的 mapPartitions 和 map 操作都为每个文件创建 1 个任务。

所以问题 How to set the number of partitions for newAPIHadoopFile? 建议设置 conf.setInt("mapred.max.split.size", 4); 用于配置分区数。

但是当这个参数被设置时，CPU被最大利用，即使经过很长时间也没有一个阶段没有启动。

如果我不设置此参数，那么应用程序将如上所述成功完成。

如何用newAPIHadoopFile设置分区数，提高效率？

ma​​pred.max.split.size 选项会发生什么？

============

更新： ma​​pred.max.split.size 选项会发生什么？

在我的用例中，文件大小很小，在这里更改拆分大小选项无关紧要。

有关此 SO 的更多信息：Behavior of the parameter "mapred.min.split.size" in HDFS

Answer 1

只需使用baseRDD.repartition(<a sane amount>).mapPartitions(...)。这会将生成的操作移动到更少的分区，尤其是在您的文件很小的情况下。