python - 在 'pyspark.resultiterable.ResultIterable' 上调用 Distinct

我正在写一些 Spark 代码，我有一个 RDD 看起来像

[(4, <pyspark.resultiterable.ResultIterable at 0x9d32a4c>),
 (1, <pyspark.resultiterable.ResultIterable at 0x9d32cac>),
 (5, <pyspark.resultiterable.ResultIterable at 0x9d32bac>),
 (2, <pyspark.resultiterable.ResultIterable at 0x9d32acc>)]

我需要做的是在 pyspark.resultiterable.ResultIterable 上调用一个不同的

我试过这个

def distinctHost(a, b):
  p = sc.parallelize(b)
  return (a, p.distinct())

mydata.map(lambda x: distinctHost(*x))

但我收到一个错误:



错误是不言自明的，我不能使用 sc。但我需要找到一种方法来覆盖 pyspark.resultiterable 。 ResultIterable 到 RDD 以便我可以调用它的不同。

最佳答案

直接的方法是使用集合:

from numpy.random import choice, seed
seed(323)

keys = (4, 1, 5, 2)
hosts = [
    u'in24.inetnebr.com',
    u'ix-esc-ca2-07.ix.netcom.com',
    u'uplherc.upl.com',
    u'slppp6.intermind.net',
    u'piweba4y.prodigy.com'
]

pairs = sc.parallelize(zip(choice(keys, 20), choice(hosts, 20))).groupByKey()
pairs.map(lambda (k, v): (k, set(v))).take(3)

结果:

[(1, {u'ix-esc-ca2-07.ix.netcom.com', u'slppp6.intermind.net'}),
 (2,
  {u'in24.inetnebr.com',
   u'ix-esc-ca2-07.ix.netcom.com',
   u'slppp6.intermind.net',
   u'uplherc.upl.com'}),
 (4, {u'in24.inetnebr.com', u'piweba4y.prodigy.com', u'uplherc.upl.com'})]

如果使用 rdd.disinct 有特殊原因，您可以尝试以下操作:

def distinctHost(pairs, key):
    return (pairs
        .filter(lambda (k, v): k == key)
        .flatMap(lambda (k, v): v)
        .distinct())

[(key, distinctHost(pairs, key).collect()) for key in pairs.keys().collect()]

关于python - 在 'pyspark.resultiterable.ResultIterable' 上调用 Distinct，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/30967880/