уменьшить операцию в словаре с искровым возвратом вместо фрейма данных

Question

Я пишу операцию reduce, ожидаю фрейм данных вместо словаря. Согласно следующему коду, он выдаст словарь

def funReduce(a, b):
    result = {}
    # first element
    if type(a) is tuple:
        result = a[1]
    else:
        result = a
    if b is not None:
        for key in list(b[1].keys()):
            if key not in result:
                result[key] = 1
            else:
                result[key] = result[key] + 1
    return result

d = sc.parallelize([(1305670057984, {(1000001256903, 1000001120912): 1, (1000001423245, 1000001120913): 1}), (1000001256903, {(1000001256903, 1000001120912): 1})])

s = d.reduce(funReduce)

У меня есть фрейм данных, похожий на d, кортеж с идентификатором транзакции и его приобретенный продукт (транскрипция A-> B) со счетчиком. Поэтому моя цель сейчас состоит в том, чтобы создать фрейм данных о количестве продуктов (A-> B), объединив все детали транзакции примерно так:

{(1000001423245, 1000001120913): 1, (1000001256903, 1000001120912): 2}

с приведенным выше кодом, я могу это сделать, но в результате получается словарь. Мне нужен фрейм данных, чтобы продолжить. потому что, если он преобразован в словарь, нет смысла писать это в Spark.

Answer 1

Это просто количество слов, поэтому либо reduceByKey, чтобы получить RDD:

d.values().flatMap(lambda d: ((x, 1) for x in d.keys())).reduceByKey(lambda x, y: x + y).collect()
# [((1000001423245, 1000001120913), 1), ((1000001256903, 1000001120912), 2)]

или explode и agg:

from pyspark.sql.functions import explode

spark.createDataFrame(d).select(explode("_2")).groupBy("key").count().show(truncate=False)
# +------------------------------+-----+
# |key                           |count|
# +------------------------------+-----+
# |[1000001423245, 1000001120913]|1    |
# |[1000001256903, 1000001120912]|2    |
# +------------------------------+-----+

чтобы получить DataFrame.