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

Question

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

以下是代码片段。

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 上的 mapPartition 和 map 操作都为每个文件创建 1 个任务。

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

但是当这个参数被设置时CPU被最大限度地利用并且none的阶段即使在很长时间后也不会启动

如果我不设置这个参数，那么应用程序会像上面提到的那样成功完成。

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

mapred.max.split.size选项会发生什么？

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

更新： mapred.max.split.size 选项会发生什么？

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

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

Answer 1

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

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

How to tune Spark application with hadoop custom input format

hadoop

mapreduce

apache-spark