Получить плоский RDD от RDD с вложенными элементами в pyspark - PullRequest
Новая
       10

Получить плоский RDD от RDD с вложенными элементами в pyspark

0 голосов
/ 14 апреля 2020

У меня есть два RDD в Pyspark с такими вложенными элементами: 

a = sc.parallelize(( (1,2), 3,(4,(6,7,(8,9,(11),10)),5,12)))

b = sc.parallelize(1,2,(3,4))

Вложенность может иметь любую глубину.

Я хочу объединить их, а затем найти максимальный элемент в любом глубины, поэтому я попытался преобразовать его в RDD, не имея таких вложенных значений (1,2,3,4,6,7,8,9,11,10,5,12,1,2,3,4) и получить максимум, используя любой из них (карта, уменьшение, фильтр, плоская карта, функция лямда). Может кто-нибудь сказать мне, как преобразовать или получить максимальный элемент.

Я пришел с решением, но оно работает только для двух уровней глубины, как 

a = sc.parallelize(( (1,2), 3,(4,5)))
b = sc.parallelize((2,(4,6,7),8))

def maxReduce(tup):
    return int(functools.reduce(lambda a,b : a if a>b else b, tup))

maxFunc = lambda x: maxReduce(x) if type(x) == tuple else x

a.union(b).map(lambda x: maxFunc(x)).reduce(lambda a,b : a if a>b else b)

Приведенный выше код работает только для глубины два, мне нужно работать на любой данной глубине (1,(2,3,(4,5,(6,(7,(8)))))).

1 Ответ

3 голосов
/ 14 апреля 2020

Хороший вариант использования рекурсивной функции: 

from collections import Iterable

a = sc.parallelize(((1, 2), 3, (4, (6, 7, (8, 9, (11), 10)), 5, 12)))
b = sc.parallelize((1, 2, (3, 4)))

def maxIterOrNum(ele):
    """
    this method finds the maximum value in an iterable otherwise return the value itself
    :param ele: An iterable of numeric values or a numeric value
    :return: a numeric value
    """
    res = -float('inf')
    if isinstance(ele, Iterable):
        for x in ele:
            res = max(res, maxIterOrNum(x))
        return res
    else:
        return ele

a.union(b).reduce(lambda x, y: max(maxIterOrNum(x), maxIterOrNum(y)))
...