java - 使用 Spark 和 Java 8 获取并过滤多个列

Question

嗯，我有一个 tsv 文件，有 4 列:id、用户、数据(字符串)和日期。我想做的是从最短的数据中获取用户、数据和日期。所以我尝试这样做。

        JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

         JavaRDD<String> lines = sparkContext
            .textFile(args[0]);

        JavaRDD<String> messages = lines
            .map(line -> line.split("\t+")[2]);
        JavaPairRDD<String, Integer> ones = messages
            .mapToPair(string -> new Tuple2<>(string,string.length()));
        JavaPairRDD<Integer, String> reverse = ones
            .mapToPair(tuple -> new Tuple2<>(tuple._2, tuple._1));
        JavaPairRDD<Integer, String> sorted = reverse
            .sortByKey(true)
            .cache();

        List<Tuple2<Integer, String>> output = sorted
            .take(1);

这样，我就有了最短的数据和他的长度，但现在我如何将其与用户和日期联系起来？我不知道该怎么做...有什么想法吗？

Answer 1

您可以在使用 Tuple2 创建对 rdd 时将整个记录存储为第二个字段。在 Pair Rdd 中，将键设置为列 data 的长度，值将是整个记录。然后执行sortByKey后，rdd对将根据key(即列数据的长度)进行排序。并且在值中存在整个记录，您可以使用 tuple._2 访问它，然后用分隔符分割并使用您需要的任何值。如下所示:

JavaSparkContext sparkContext = new JavaSparkContext(sparkConf);

 JavaRDD<String> lines = sparkContext
            .textFile(args[0]);

JavaPairRDD<Integer,String> pairedRdd = lines.mapToPair(lines -> new Tuple2((lines.split("\t+")[2]).length(), lines));

        JavaPairRDD<Integer, String> sortedRdd = pairedRdd.sortByKey(true);

        List<Tuple2<Integer, String>> output = sortedRdd.take(1);

        System.out.println(output.get(0)._1+" "+output.get(0)._2);

output.get(0)._2 具有完整记录，用 \t+ 分割。然后从数组中获取您需要的元素。