Добавление целого числа с помощью combByKey добавляет их в списки

Question

У меня есть функция карты, которая создает пару ключ-значение типа [Tuple, Integer], но когда я использую комбинироватьByKey для добавления целочисленных значений, она создает кортеж из этих целых чисел вместо их добавления.

Я попытался ввести целое число, но это не помогает.

def subset_from_kv(para):
    sol = []

    para_new = (i for i in (para and ff))
    # print(list(para))
    tt = (itertools.combinations(sorted(para_new), size))

    for j in tt:
        flag = True

        for i in list(itertools.combinations(list(j), size-1)):

            if (i not in (frequent_itemset_val)):
                flag = False

                break
        if flag is True:
            sol.append(j)

    return sol


rdd1_values = rdd1_original.values()

rdd_inter = rdd1_values.mapPartitions(lambda t: subset_from_kv(t)).map(lambda x: (x, 1))

new_item_rdd_size_i = rdd_inter.combineByKey(
    lambda value: (value),
    lambda x, y: (x + y),
    lambda x, y: (x, y)
)

rdd_inter.collect() дает:

[(('-050d_XIor1NpCuWkbIVaQ', '-4TMQnQJW1yd6NqGRDvAeA'), 1), (('-050d_XIor1NpCuWkbIVaQ', '-6h3K1hj0d4DRcZNUtHDuw'), 1), (('-050d_XIor1NpCuWkbIVaQ', '-6tvduBzjLI1ISfs3F_qTg'), 1), (('-050d_XIor1NpCuWkbIVaQ', '-9eNGMp8XiygI8t8QFuFWw'), 1), (('-050d_XIor1NpCuWkbIVaQ', '-9nai28tnoylwViuJVrYEQ'), 1).....

РЕДАКТИРОВАТЬ: У меня есть 3 раздела

Ожидается, что new_item_rdd_size_i даст

[(('-050d_XIor1NpCuWkbIVaQ', '-4TMQnQJW1yd6NqGRDvAeA'),(10, 1, 3), (('-050d_XIor1NpCuWkbIVaQ', '-6h3K1hj0d4DRcZNUtHDuw'), (12, 13, 5), (('-050d_XIor1NpCuWkbIVaQ', '-6tvduBzjLI1ISfs3F_qTg'), (21, 7, 33), (('-050d_XIor1NpCuWkbIVaQ', '-9eNGMp8XiygI8t8QFuFWw'), (111, 34, 14), (('-050d_XIor1NpCuWkbIVaQ', '-9nai28tnoylwViuJVrYEQ'), (41, 33, 11)...

но вывод объединяет значения 1 вместо их добавления:

[(('-050d_XIor1NpCuWkbIVaQ', '-Bdw-5H5C4AYSMGnAvmnzw'), ((1, 1), 1)), (('-050d_XIor1NpCuWkbIVaQ', '-De4AV1Fx67mDMGrFOw44Q'), ((1, 1), 1)), (('-050d_XIor1NpCuWkbIVaQ', '-FLnsWAa4AGEW4NgE8Fqew'), ((1, 1), 1)), (('-050d_XIor1NpCuWkbIVaQ', '-Ht7HiGBox8lS1Y8IPjO8g'), ((1, 1), 1)), (('-050d_XIor1NpCuWkbIVaQ', '-ZBfr1BHvArFp1d6XH8jOQ'), ((1, 1), 1))]

Answer 1

в соответствии с официальным документом ,

combineByKey(
    createCombiner, 
    mergeValue, 
    mergeCombiners, 
    numPartitions=None, 
    partitionFunc=<function portable_hash>
)

с:

• createCombiner , который превращает V в C (например, создает один элемент список)
• mergeValue , чтобы объединить V в C (например, добавить его в конец список)
• mergeCombiners , чтобы объединить два C в один (например, объединяет списки)

В вашем случае:

• createCombiner is lambda value: (value): вы создаете кортеж
• mergeValue равно lambda x, y: (x + y): вы добавляете 2 кортежа ==> 1 кортеж с 2 элементами
• mergeCombiners is lambda x, y: (x, y): вы создаете кортеж с 2 элементами ...

Следовательно, ваш окончательный результат - кортеж кортежа кортежа кортежа и т. Д.

Вы должны попробовать более простой метод, например reduceByKey: rdd.reduceByKey(add)