仅提取 Java Spark 中的某些列

Question

我有一个包含 10 列的文件。仅提取前 3 列或特定列的最优雅方法是什么？

例如，我的文件是这样的：

john,smith,84,male,kansas
john,doe,48,male,california
tim,jones,22,male,delaware

我想提取到这个：

[john, smith, kansas]
[john, doe, california]
[tim, jones, delaware]

我拥有的是这个，但它没有专门选择我想要的列：

JavaRDD<String> peopleRDD = sc.textFile(DATA_FILE);
peopleRDD.cache().map(lines -> Arrays.asList(lines.split(",")))
                 .forEach(person -> LOG.info(person));

我阅读了 following two Stackoverflow 的帖子，但我仍然无法决定如何执行此操作。

编辑： 我最终做了以下事情：

JavaRDD<String> peopleRDD = sc.textFile(DATA_FILE);
    peopleRDD.cache().map(lines -> Arrays.asList(new String[]{lines.split(",")[0], 
                                                        lines.split(",")[1], 
                                                        lines.split(",")[3]}
                     .forEach(person -> LOG.info(person));

不是最优雅的解决方案，但如果您有更好的方法，请在此处发布。谢谢。

Answer 1

编辑：抱歉，我刚刚意识到您要求的是 Java 解决方案，但我使用的是 Scala。我的建议中只有第 3 个在 Java 中具有等价物（添加在答案的底部）……Spark 在 Scala 中确实更好:-)

一种方法是执行split，然后对结果进行模式匹配以选择您想要的列：

peopleRDD.cache().map(_.split(",") match { case Array(a,b,_,_,e) => List(a,b,e) }) 

另一个（取决于您想要的元素组合）是使用take 和drop，使用val 以避免重复拆分。

peopleRDD.cache().map{ line => 
    val parts = line.split(",") 
    parts.take(2) ++ parts.drop(4)
}

（如果您想为 RDD 中的每个结果元素添加 List 而不是 Array，则可以在 split 之后添加 toList）

事实上，同样的方法可以用来简化你原来的解决方案，例如：

peopleRDD.cache().map{ line => 
  val parts = line.split(",")
  List(parts[0], parts[1], parts[4])
}

在 Java8 中，您可能可以执行等效操作，这是一个轻微的改进，因为我们避免重复调用 split - 类似：

peopleRDD.cache().map( line -> {
  Array<String> parts = line.split(",");
  Arrays.asList(new String[]{parts[0], parts[1], parts[4]});
})