如何从sparkContext读取特定行

Question

你好，我正在尝试使用Spark从文本文件中读取特定行。

SparkConf conf = new SparkConf().setAppName(appName).setMaster(master);
sc = new JavaSparkContext(conf);
JavaRDD<String> lines = sc.textFile("data.txt");
String firstLine = lines.first();

可以使用 .first() 命令获取 data.text 文档的第一行。如何访问文档的第 N 行？我需要 Java 解决方案。

- Miraj Hamid

Answer 1

Apache Spark RDDs不应用于查找。获取第n行最"高效"的方法是lines.take(n + 1).get(n)。每次执行此操作时，它将读取文件的前n行。您可以运行lines.cache来避免这种情况，但它仍会在非常低效的情况下将前n行移动到网络上。

如果数据可以全部适合一台机器，请一次性收集所有数据，并本地访问：List<String> local = lines.collect(); local.get(n);。

如果数据不能适合一台机器，则需要支持高效查找的分布式系统。常见的示例是HBase和Cassandra。

还有可能您的问题可以通过Spark高效地解决，但不是通过查找。如果您在单独的问题中解释了更大的问题，则可能会得到类似的解决方案。（查找在单机应用程序中非常常见，但分布式算法必须思考不同的方式。）

- Daniel Darabos

---

如果您选择使用DataFrames的take-get方法，我认为您需要使用lines.take(n).apply(n-1)。 - Josiah Yoder

谢谢，我用一种方式修复了它。假设 "nth" 是从零开始的 :). lines 应该是一个 JavaRDD ，因此 take 返回一个 Java List<String>，因此需要使用 get 而不是 apply。 - Daniel Darabos

Answer 2

我认为这是最快的速度。

def getNthLine(n: Long) = 
  lines.zipWithIndex().filter(_._2 == n).first

- abalcerek

Answer 3

正如 @Daniel Darabos 所说，RDDs 没有为行查找索引，因此另一种方法是给它一个索引：

lines.zipWithIndex.filter(_._2==n).map(_._1).first()

给它一个索引，然后再次使用Spark上下文，但是当您的RDD大小很小时，这种方法有些低效和愚蠢。 但是，当您的RDD非常大时，将其收集到主节点变得低效（可能会受到内存限制），此方法成为更好的选择。

- GameOfThrows