python - 过滤 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 个值关联的过滤键

输出 - 只有那些以 key1, key2 作为键的 (k,v) 对应该在那里，因为它们与至少 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'])]

现在如何将这个 (key, list(values)) 拆分为单个 (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]])

这行得通!!