仅提取Java中的某些列Spark 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)); 

我读了followingtwo堆栈溢出帖子，但我仍然无法决定如何做到这一点。

编辑： 最后我做了以下内容：

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。只有我的第三个建议有一个相当于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) 
} 

（您可以在split后添加toList如果你想有一个List而不是Array在RDD每个结果元素）

其实同样的方法可以用来简化你的原始解决方案如：

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

在Java8，你也许可以做等同，这是一种轻微的改善，因为我们避免调用split反复 - 这是LIK e：

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