Как удалить повторяющиеся значения из RDD [PYSPARK]

У меня есть следующая таблица как RDD:

Key Value
1    y
1    y
1    y
1    n
1    n
2    y
2    n
2    n

Я хочу удалить все дубликаты из Value.

Выход должен выглядеть следующим образом:

Key Value
1    y
1    n
2    y
2    n

При работе в pyspark вывод должен появиться как список пар ключ-значение, например:

[(u'1',u'n'),(u'2',u'n')]

Я не знаю, как применять цикл for здесь. В обычной программе Python это было бы очень легко.

Интересно, есть ли какая-то функция в pyspark для того же самого.

Ответ 1

Я боюсь, что у меня нет знаний о python, поэтому все ссылки и код, которые я предоставляю в этом ответе, относятся к java. Однако преобразовать его в код python не очень сложно.

Вы должны взглянуть на следующую веб-страницу. Он перенаправляет на официальную веб-страницу Spark, которая предоставляет список всех преобразований и действий, поддерживаемых Spark.

Если я не ошибаюсь, лучшим способом (в вашем случае) было бы использовать преобразование distinct(), которое возвращает новый набор данных, содержащий отдельные элементы исходного набора данных (взятого из ссылки). В java это будет что-то вроде:

JavaPairRDD<Integer,String> myDataSet = //already obtained somewhere else
JavaPairRDD<Integer,String> distinctSet = myDataSet.distinct();

Итак, например:

Partition 1:

1-y | 1-y | 1-y | 2-y
2-y | 2-n | 1-n | 1-n

Partition 2:

2-g | 1-y | 2-y | 2-n
1-y | 2-n | 1-n | 1-n

Будет преобразован в:

Partition 1:

1-y | 2-y
1-n | 2-n 

Partition 2:

1-y | 2-g | 2-y
1-n | 2-n |

Конечно, у вас все равно будет несколько наборов данных RDD, каждый из которых содержит список отдельных элементов.

Ответ 2

Эту проблему легко решить с помощью операции distinct библиотеки pyspark из Apache Spark.

from pyspark import SparkContext, SparkConf

# Set up a SparkContext for local testing
if __name__ == "__main__":
    sc = SparkContext(appName="distinctTuples", conf=SparkConf().set("spark.driver.host", "localhost"))

# Define the dataset
dataset = [(u'1',u'y'),(u'1',u'y'),(u'1',u'y'),(u'1',u'n'),(u'1',u'n'),(u'2',u'y'),(u'2',u'n'),(u'2',u'n')]

# Parallelize and partition the dataset 
# so that the partitions can be operated
# upon via multiple worker processes.
allTuplesRdd = sc.parallelize(dataset, 4)

# Filter out duplicates
distinctTuplesRdd = allTuplesRdd.distinct() 

# Merge the results from all of the workers
# into the driver process.
distinctTuples = distinctTuplesRdd.collect()

print 'Output: %s' % distinctTuples

В результате вы получите следующее:

Output: [(u'1',u'y'),(u'1',u'n'),(u'2',u'y'),(u'2',u'n')]

Ответ 3

Если вы хотите удалить все дубликаты из определенного столбца или набора столбцов, т.е. выполнив distinct в наборе столбцов, тогда у pyspark будет функция dropDuplicates, которая будет принимать определенный набор столбцов, на которые он может быть включен.

ака

df.dropDuplicates(['value']).show()