调整和过滤PySpark RDD同时进行

Question

我有一个PySpark RDD，其中每个元素的形式为(key, val)，它采用以下两种形式之一：

elm1 = ((1, 2), ((3, 4), (5, 6)))  # key = (1,2), rest is val
elm2 = ((1, 2), ((3, 4), None))

现在，我需要做以下的事情

1. 检测val的第二部分为None的元素（如elm2）并提取它们

2. 将它们展平如下，并用空字符串的元组替换None：

   elm = (1, 2, 3, 4, ('', ''))
   

要在PySpark中执行上述两个步骤，我需要：

elm = elm.filter(lambda x: detectNone(x))  # checks if x[-1][1] is None
elm = elm.map(formatElm) # where formatElm is a function that replaces None with tuple of empty strings and flattens the tuple.

实际上，testx[-1][1] == None有点复杂，并且引入了更复杂的数据结构来代替空字符串的元组

问题：有没有办法加快这些操作。我认为把两个操作合并成一个可能会有帮助，但我不知道怎么做

Answer 1

I think combining two operations into one may help,

不会的。但如果你真的坚持这样做，那么flatMap：

rdd = sc.parallelize([((1, 2), ((3, 4), (5, 6))), ((1, 2), ((3, 4), None))])


def detect_and_format(row):
    x, (y, z) = row
    return [x + y + (("", ""), )] if z is None else []

# [(1, 2, 3, 4, ('', ''))]