python - 根据row_number过滤RDD

Question

sc.textFile(path) 允许读取 HDFS 文件，但它不接受参数(比如跳过一些行，has_headers，...)。

《Learning Spark》O'Reilly 电子书建议使用如下函数读取CSV(例5-12.Python加载CSV示例)

import csv
import StringIO

def loadRecord(line):
"""Parse a CSV line"""
input = StringIO.StringIO(line)
reader = csv.DictReader(input, fieldnames=["name", "favouriteAnimal"])
return reader.next()
input = sc.textFile(inputFile).map(loadRecord)

我的问题是关于如何选择“已采取”的行:

1. 如何避免加载第一行(标题)
2. 如何删除特定行(例如，第 5 行)

我在这里看到了一些不错的解决方案:select range of elements但我想看看是否有更简单的方法。

谢谢!

Answer 1

不用担心加载不需要的行/行。当你这样做时:

input = sc.textFile(inputFile)

您没有加载文件。您只是获得一个允许您对文件进行操作的对象。所以为了提高效率，最好从只得到你想要的东西的角度来思考。例如:

header = input.take(1)[0]
rows = input.filter(lambda line: line != header)

请注意，这里我没有使用索引来引用我想要删除的行，而是它的值。这有一个副作用，即其他具有此值的行也将被忽略，但更符合 Spark 的精神，因为 Spark 会将您的文本文件分布在节点的不同部分，并且行号的概念在每个分区中都会丢失。这也是为什么这在 Spark(Hadoop) 中不容易做到的原因，因为每个分区都应该被认为是独立的，而全局行号会打破这个假设。

如果您确实需要使用行号，我建议您将它们添加到 Spark 外部的文件中(请参阅 here )，然后在 Spark 内部按此列进行过滤。

编辑:按照@Daniel Darabos 的建议添加了zipWithIndex 解决方案。

sc.textFile('test.txt')\
  .zipWithIndex()\            # [(u'First', 0), (u'Second', 1), ...
  .filter(lambda x: x[1]!=5)\ # select columns
  .map(lambda x: x[0])\       # [u'First', u'Second'
  .collect()