hadoop - 我需要使用pyspark从RDD过滤掉一些内容

Question

我需要从存储在HDFS中的选项卡已分发文件中提取几列。

我能够从HDFS读取到RDD并将行拆分为列表，但是我不知道如何获取我关心的列。

我的代码:

raw_file = sc.textFile("hdfs.......tsv")
rdd = raw_file.map(lambda line: line.split('\t'))

newfile中的每一列都包含相同类型的内容，我想提取列26、80、109、452并将它们放入列表中。

我试过了:

filtered = rdd.filter(lambda line: append(line[26]), append(line[80]), append(line[109], append(line[452])).collect()

但显然没有添加方法。那我该怎么办？

Answer 1

过滤器用于根据某些条件返回或省略行。

rdd = sc.range(10)
even = rdd.filter(lambda x: x % 2 == 0)
even.collect()
# Out: [0, 2, 4, 6, 8]

您正在寻找的是 map 。 map 用于转换(或提取部分)行。让我们以字母列表的rdd为例。如果只需要第1，第3和第5个元素(列)，则可以使用 map 提取它们。

rdd = sc.range(2).map(lambda i: ["a", "b", "c", "d", "e", "f"])
print rdd.collect()
# [['a', 'b', 'c', 'd', 'e', 'f'], ['a', 'b', 'c', 'd', 'e', 'f']]

mapped = rdd.map(lambda r: [r[1], r[3], r[5]])
print mapped.collect()
# [['b', 'd', 'f'], ['b', 'd', 'f']]

过滤器作用于行，映射作用于行中的数据