在RDD中筛选数据

Question

我已经成功地在pyspark中预处理了我的数据，得到了这样的结果

[(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key3', u'2'), (u'key4', u'1'), (u'key1', u'4'), (u'key5', u'1'), (u'key6', u'2'), (u'key7', u'4'), (u'key8', u'5'), (u'key9', u'6'), (u'key10', u'7')]

现在我需要根据这些条件进行筛选：

1）筛选与至少2个键关联的值。

输出-只有那些值为“1”、“2”和“4”的（k，v）对才应该出现，因为它们与2个以上的键关联

 [(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key3', u'2'), (u'key4', u'1'), (u'key1', u'4'), (u'key5', u'1'), (u'key6', u'2'), (u'key2', u'4')]

2）筛选与至少2个值关联的键

输出-只有那些（k，v）对具有key1，key2作为键，因为它们与至少2个值关联

[(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key1', u'4'), (u'key2', u'4')]

任何建议都会有很大帮助。

更新：我使用groupBy和一个过滤器对具有多个值的键进行分组

 [(u'key1', [u'1', u'2', u'4']), (u'key2',[u'1', u'4'])]

现在如何将这个（键，列表（值））拆分为单独的（k，v）对以应用进一步的转换？

Answer 1

my_rdd = sc.parallelize([(u'key1', u'1'), (u'key2', u'1'), (u'key1', u'2'), (u'key2', u'3'), (u'key4', u'1'), (u'key1', u'4'), (u'key4', u'1'), (u'key6', u'2'), (u'key7', u'4'), (u'key8', u'5'), (u'key9', u'6'), (u'key10', u'7')])

#filter keys which are associated to atleast 2 values

filter2_rdd = my_rdd.groupByKey() \
                    .mapValues(lambda x: list(x)) \
                    .filter(lambda x: len(x[1])>=2) \
                    .flatMap(lambda x: [(x[0],item) for item in x[1]])

#filter values associated to atleast 2 keys.
filte1_rdd = filter2_rdd.map(lambda x: (x[1],x[0])) \
                        .groupByKey().mapValues(lambda x: list(x))\
                        .filter(lambda x: len(x[1])>=2)\
                        .flatMap(lambda x: [(item,x[0]) for item in x[1]])

这会有用的！！

Answer 2

按键减少、筛选和联接：

>>> rdd.mapValues(lambda _: 1) \  # Add key of value 1
...     .reduceByKey(lambda x, y: x + y) \ # Count keys
...     .filter(lambda x: x[1] >= 2) \ # Keep only if number is >= 2
...     .join(rdd) # join with original (serves as filter)
...     .mapValues(lambda x: x[0]) # reshape