Pyspark数据帧：对一列中的唯一值进行计数，与其他列中的值相互独立

Question

我有一个spark数据框架，包括从不同来源获得的关于两种分子、调节器和目标（它们之间没有重叠）之间相互作用的数十亿个预测。我需要添加一列 包含预测给定“调节器”和给定“目标”的至少一个交互的数字资源。你知道吗

换句话说，对于每一对“调节器”和“目标”，我试图获得包含“调节器”和“目标”值的源的数量，即使在一次交互中没有配对。你知道吗

示例：

+---------+------+------+
|Regulator|Target|Source|
+---------+------+------+
|        m|     A|     x|
|        m|     B|     x|
|        m|     C|     z|
|        n|     A|     y|
|        n|     C|     x|
|        n|     C|     z|
+---------+------+------+

我想得到的是：

+---------+------+------+----------+
|Regulator|Target|Source|No.sources|
+---------+------+------+----------+
|        m|     A|     x|         1|
|        m|     B|     x|         1|
|        m|     C|     z|         2|
|        n|     A|     y|         2|
|        n|     C|     x|         2|
|        n|     C|     z|         2|
+---------+------+------+----------+

进一步说明：

第一行(m, A, x)：

• 涉及m的相互作用由x和z源预测
• 涉及A的相互作用由源x和y预测
• 因此，它们的重叠部分是x没有来源等于1。你知道吗

第二行(m, B, x)：

• 涉及m的相互作用由x和z源预测
• 涉及B的相互作用仅由x源预测。你知道吗
• 因此，它们的重叠部分是x没有来源等于1。你知道吗

第三行(m, C, z)：

• 涉及m的相互作用由x和z预测
• 源x和z预测了与C有关的相互作用
• 因此，它们的重叠部分是x，z没有来源等于2。你知道吗

Answer 1

这里有一种解决这个问题的方法。为每行创建两个新列：

• 列'RS'：'Regulator'的源集合
• 列'TS'：'Target'的源集合

那么你想要的输出就是这些集合的交集的长度。你知道吗

考虑以下示例：

创建数据帧

from pyspark.sql Window
import pyspark.sql.functions as f
cols = ["Regulator", "Target", "Source"]
data = [
    ('m', 'A', 'x'),
    ('m', 'B', 'x'),
    ('m', 'C', 'z'),
    ('n', 'A', 'y'),
    ('n', 'C', 'x'),
    ('n', 'C', 'z')
]

df = sqlCtx.createDataFrame(data, cols)

创建新列

使用^{}和^{}计算'Source'列的不同值：

df = df.withColumn(
    'RS',
    f.collect_set(f.col('Source')).over(Window.partitionBy('Regulator'))
)

df = df.withColumn(
    'TS',
    f.collect_set(f.col('Source')).over(Window.partitionBy('Target'))
)
df.sort('Regulator', 'Target', 'Source').show()
#+    -+   +   +   +    -+
#|Regulator|Target|Source|    TS|       RS|
#+    -+   +   +   +    -+
#|        m|     A|     x|[y, x]|   [z, x]|
#|        m|     B|     x|   [x]|   [z, x]|
#|        m|     C|     z|[z, x]|   [z, x]|
#|        n|     A|     y|[y, x]|[y, z, x]|
#|        n|     C|     x|[z, x]|[y, z, x]|
#|        n|     C|     z|[z, x]|[y, z, x]|
#+    -+   +   +   +    -+

计算交叉口的长度

定义一个udf来返回两个集合交集的长度，并使用它来计算'No_sources'列。（注意，我在列名中使用了_而不是.，因为这样更容易使用select()。）

intersection_length_udf = f.udf(lambda u, v: len(set(u) & set(v)), IntegerType())

df = df.withColumn('No_sources', intersection_length_udf(f.col('TS'), f.col('RS')))

df.select('Regulator', 'Target', 'Source', 'No_sources')\
    .sort('Regulator', 'Target', 'Source')\
    .show()
#+    -+   +   +     +
#|Regulator|Target|Source|No_sources|
#+    -+   +   +     +
#|        m|     A|     x|         1|
#|        m|     B|     x|         1|
#|        m|     C|     z|         2|
#|        n|     A|     y|         2|
#|        n|     C|     x|         2|
#|        n|     C|     z|         2|
#+    -+   +   +     +