java - 在 Apache Spark 中按键对 csv 文件进行排序

Question

我有一个 csv 文件，其中包含以下格式的数据:

id,  name,  surname,   morecolumns
5,   John,  Lok,       more
2,   John2, Lok2,      more
1,   John3, Lok3,      more
etc..

我想使用 id 作为键对 csv 文件进行排序，并将排序结果存储在另一个文件中。

到目前为止我已经做了什么来创建(id，rest_of_line)的JavaPairs。

    SparkConf conf = new SparkConf().setAppName.....;

    JavaSparkContext sc = new JavaSparkContext(conf);

    JavaRDD<String> file = sc.textFile("inputfile.csv");

    // extract the header
    JavaRDD<String> lines = file.filter(s -> !s.equals(header));

    // create JavaPairs
    JavaPairRDD<Integer, String> pairRdd = lines.mapToPair(
      new PairFunction<String, Integer, String>() {
        public Tuple2<Integer, String> call(final String line) {


            String str = line.split(",", 2)[0];
            String str2 = line.split(",", 2)[1];
            int id = Integer.parseInt(str);

            return new Tuple2(id, str2);
        }
    });

    // sort and save the output
    pairRdd.sortByKey(true, 1);
    pairRdd.coalesce(1).saveAsTextFile("sorted.csv");

这适用于我有小文件的情况。但是，当我使用较大的文件时，输出未正确排序。我认为发生这种情况是因为排序过程发生在不同的节点上，因此来自所有节点的所有过程的合并不会给出预期的输出。

所以，问题是如何使用 id 作为键对 csv 文件进行排序并将排序结果存储在另一个文件中。

Answer 1

方法coalesce可能是罪魁祸首，因为它显然没有在契约(Contract)上保证排序或生成的 RDD(参见 Which operations preserve RDD order? )。所以如果你避免这样的coalesce ，生成的输出文件将被排序。如您想要独一无二csv文件，您可以从您正在使用的任何文件系统中获取结果，但要注意它们的实际顺序，然后合并它们。例如，如果您使用 HDFS (如@PinoSan所述)这可以使用命令 hdfs dfs -getmerge <hdfs-output-dir> <local-file.csv> 来完成.