подсчет строк в rdd зависит от контекста линий, pyspark

Question

В настоящее время я пытаюсь понять, как работает RDD. Например, я хочу подсчитать строки на основе контекста в каком-то объекте RDD. У меня есть некоторые Experince с DataFrames и мой код DF, который имеет, например, колонн A, B и, вероятно, некоторые другие столбцы, выглядит так:

df = sqlContext.read.json("filepath") 
df2 = df.groupBy(['A', 'B']).count() 

Логическая часть этого кода ясно для меня - я groupBy операция над именем столбца в DF. В RDD у меня нет имени столбца, только похожие строки, которые могут быть кортежем или объектами Row ... Как я могу подсчитать похожие кортежи и добавить его как целое в уникальную строку? Например мой первый код:

df = sqlContext.read.json("filepath") 
rddob = df.rdd.map(lambda line:(line.A, line.B)) 

Я делаю операцию карты и создать кортеж значений из ключей A и B. У уникальной строки больше нет ключей (это самое важное отличие от DataFrame, у которого есть имя столбца). Теперь я могу произвести что-то подобное, но он вычисляет только общее количество строк в RDD.

rddcalc = rddob.distinct().count() 

То, что я хочу для моего выхода, просто:

((a1, b1), 2) 
((a2, b2), 3) 
((a2, b3), 1) 
... 

PS

Я нашел мое личное решение этого вопроса. Здесь: rdd - начальный rdd, rddlist - список всех строк, rddmod - окончательный измененный rdd и, следовательно, решение.

rddlist = rdd.map(lambda line:(line.A, line.B)).map(lambda line: (line, 1)).countByKey().items() 
rddmod = sc.parallelize(rddlist) 

Answer 1

Я верю, что вы ищете здесь reduceByKey. Это даст вам количество раз, сколько раз появляется каждая отдельная пара линий (a,b). Это будет выглядеть следующим образом:

rddob = df.rdd.map(lambda line: (line.A + line.B, 1)) 
counts_by_key = rddob.reduceByKey(lambda a,b: a+b) 

Теперь у вас есть ключ, значение пары вида: ((a,b), count-of-times-pair-appears)

Пожалуйста, обратите внимание, что это работает только, если А и В являются строками. Если они являются списками, вам необходимо создать объект «первичного ключа» для выполнения сокращения. Вы не можете выполнить команду reduceByKey, где основным ключом является некоторый сложный объект.