Как посчитать количество записей с ключом в Spark с помощью Python?

Question

Мои данные показывают пару слов и количество раз, сколько эта пара встречается.Например:

[("('best', 'it')", 3), ("('best', 'of')", 4), ("('best', 'the')", 3), ("('best', 'was')", 3), ("('it', 'of')", 11), ("('it', 'the')", 11)]

Моя цель - подсчитать за слово, сколько пар оно существует.Например, я хочу получить:

best 4
it 3

Одна хитрость заключается в том, что "это" не только происходит в

("('it', 'of')", 11), ("('it', 'the')", 11)

, но также происходит в

('best', 'it')", 3)

Таким образом, программа должна как-то идентифицировать это.

Как мне добиться этого в Spark с использованием Python?Я новичок, поэтому ценю вашу помощь!

Answer 1

Если вы работаете с RDD, вы можете использовать ReduByKey для этого случая

>>> rdd.collect()
[("('best', 'it')", 3), ("('best', 'of')", 4), ("('best', 'the')", 3), ("('best', 'was')", 3), ("('it', 'of')", 11), ("('it', 'the')", 11)]
>>> rddMap = rdd.map(lambda x: x[0][1:-1].split(',')).flatMap(lambda x: [(i.replace("'","").strip(),1) for i in x])
>>> rddMap.collect()
[('best', 1), ('it', 1), ('best', 1), ('of', 1), ('best', 1), ('the', 1), ('best', 1), ('was', 1), ('it', 1), ('of', 1), ('it', 1), ('the', 1)]
>>> rddReduce = rddMap.reduceByKey(lambda x,y: x+y).map(lambda x: x[0]+','+str(x[1]))

>>> for i in rddReduce.collect(): print(i)
... 
best,4
it,3
of,2
the,2
was,1

Answer 2

Сначала создайте фрейм данных pyspark из данных.

df = sql.createDataFrame(
 [("('best', 'it')", 3),\
  ("('best', 'of')", 4),\
  ("('best', 'the')", 3),\
  ("('best', 'was')", 3),\
  ("('it', 'of')", 11),\
  ("('it', 'the')", 11)],
  ['text', 'count'])

df.show()

+---------------+-----+
|           text|count|
+---------------+-----+
| ('best', 'it')|    3|
| ('best', 'of')|    4|
|('best', 'the')|    3|
|('best', 'was')|    3|
|   ('it', 'of')|   11|
|  ('it', 'the')|   11|
+---------------+-----+

Затем преобразуйте строки text в Array, взорвите text и groupby.

import pyspark.sql.functions as F
import ast

convert_udf = F.udf(lambda x: ast.literal_eval(x), ArrayType(StringType()) )

df = df.withColumn('text', convert_udf('text'))\
       .withColumn('text', F.explode('text'))\
       .groupby('text').count()

df.show() 

+----+-----+                                                                    
|text|count|
+----+-----+
| was|    1|
|  it|    3|
| the|    2|
|  of|    2|
|best|    4|
+----+-----+