循环遍历文件的行并通过 Spark 中的每次迭代执行功能

Question

我在 HDFS 中有一个名为 file1 的文件，其中包含以下几行：（每一行都是一个目录路径）

this/is/path1
this/is/path2
this/is/path3
.
.
.
this/is/path1000ormore

我有一个 Scala Spark 函数如下：

val resultset=sc.hadoopFile(inputpath,classOf[TextInputFormat],classOf[LongWritable],classOf[Text]).flatMap {
case (k, v) => if (k.get == 0) Seq(v.toString) else Seq.empty[String]
}

我想在 hadoopFile 函数中传递来自“file1”的每一行来代替“inputpath”（需要是一个字符串），并获得每个迭代/循环的结果。 我该怎么做？

额外信息：

函数的实际作用： 上面的函数从指定的目录路径中获取第一个文件，而不是“inputpath”，并给出文件的第一行。我想对存储在“file1”中的所有目录路径执行此操作，因此我正在寻找有关如何在循环/迭代中执行此操作的解决方案。

更新： 我试着把它放在这样的循环中：

val lines=Source.fromFile("/path/to/file1.txt").getLines.toList
for(i<-lines){
val firstLines=sc.hadoopFile(i,classOf[TextInputFormat],classOf[LongWritable],classOf[Text]).flatMap {
case (k, v) => if (k.get == 0) Seq(v.toString) else Seq.empty[String]
}

这运行了大约 10 分钟（文件 1 包含大约 34,000 行），并且没有导致任何错误。但是当我尝试使用以下命令查看几行输出时，

firstLines.take(3)

我收到一条错误消息：

error: not found: value firstLines
          firstLines
          ^

所以我认为循环没有成功运行，因此 firstLines 从未被创建，尽管我不知道问题可能是什么。有人可以提供解决方案吗？

Answer 1

可以分两步实现：

1. 像往常一样从 HDFS 读取“file1”文本文件，获取所有元素；
2. 对于 1) 中的每个项目，应用“结果集”逻辑。

另外 2) 可以改进：1) 中的所有项目都可以用逗号连接在一个字符串中，字符串作为“inputpath”参数传递。您将拥有一个包含所有文件数据的 RDD。过滤器“k.get == 0”可以申请得到最终结果。

首先可以这样实现：

val lines = Source.fromFile("file1.txt").getLines.toSeq.view

val resultDF = lines.map(current =>
  sc.hadoopFile(current, classOf[TextInputFormat], classOf[LongWritable], classOf[Text]).filter(_._1 == 0)
).reduce(_ union _)

resultDF.take(3).foreach(println)