Pyspark：基于typ的值操作

Question

我有这样一个RDD：

[('a', ('H', 1)), ('b', (('H', 41), ('S', 1)))]

因此键可以有一个tuple或一个tuple的tuples作为值。这来自于reduceByKey。 我需要执行一个简单的操作：将S的计数除以（H+S）的计数。 当S不存在时，就像第一项的情况一样，我必须返回0。 问题是将第一种情况（单个tuple）与第二种情况（两个tuples中的tuple）分开，这样我就知道如何在map中操作。你知道吗

我该怎么办？你知道吗

Answer 1

一般来说，在上游解决这个问题更有意义，但您可以尝试以下方法：

from operator import truediv

def f(vs):
    try:
        d = dict(vs)
    except ValueError:
        d = dict([vs])

    s = sum(d.values())
    return truediv(d.get("S", 0), s) if s else float('nan')

rdd = sc.parallelize([('a', ('H', 1)), ('b', (('H', 41), ('S', 1)))])
rdd.mapValues(f).collect()

## [('a', 0.0), ('b', 0.023809523809523808)]

或者，如果不介意外部依赖关系，可以尝试使用^{}：

from multipledispatch import dispatch

@dispatch(tuple, tuple)
def f(h, s):
    try:
        return truediv(s[1], h[1] + s[1])
    except ZeroDivisionError:
        return float('nan')

@dispatch(str, int)
def f(x, y):
    return 0.0

rdd.mapValues(lambda args: f(*args)).collect()
## [('a', 0.0), ('b', 0.023809523809523808)]