在Spark Dataframe中动态创建多列

Creating multiple columns in spark Dataframe dynamically(在Spark Dataframe中动态创建多列)

问题描述

我有包含如下信息的词典

dict_segs = {'key1' : {'a' : {'col1' : 'value1', 'col2' : 'value2', 'col3': 'value3'}, 
                'b' : {'col2' : 'value2', 'col3' : 'value3'}, 
                'c' : {'col1' : 'value1'}},
        'key2' : {'d' : {'col3' : 'value3', 'col2' : 'value2'},
                'f' : {'col1' : 'value1', 'col4' : 'value4'}}}

待办事项：

键基本上是‘段’，对于这些段，关键字1的底层词典，即a、b、c是‘子段’。每个子段的过滤条件都在子段的底层字典中提供，即a、b、c、d、f。此外，子段字典键的过滤条件也是pyspark dataframe的列名。

我想为每个段在pyspark dataframe中一次性创建子段列，当满足过滤条件时，每个子段列的值将为1，否则为0，类似于

for item in dict_segs:
    pyspark_dataframe.withColumn(*dict_segs[item].keys(), when(meeting filter criteria with respect to each key), 1).otherwise(0))

在做研究时，我能够在Scala中找到类似的东西，但那里的列过滤条件是静态的，但上面的逻辑是动态的。请参见下面的Scala逻辑

Spark/Scala repeated calls to withColumn() using the same function on multiple columns

需要支持才能根据上面的伪代码为每个段派生上述逻辑。

谢谢。

推荐答案

您正在寻找select语句：

让我们创建一个示例数据帧：

df = spark.createDataFrame(
    sc.parallelize([["value" + str(i) for i in range(1, 5)], ["value" + str(i) for i in range(5, 9)]]), 
    ["col" + str(i) for i in range(1, 5)]
)

+------+------+------+------+
|  col1|  col2|  col3|  col4|
+------+------+------+------+
|value1|value2|value3|value4|
|value5|value6|value7|value8|
+------+------+------+------+

现在针对词典中的所有keys、dict_seg[key]中的所有subkeys和dict_seg[key][subkey]中的所有columns：

import pyspark.sql.functions as psf
df.select(
    ["*"] +
    [
        eval('&'.join([
            '(df["' + c + '"] == "' + dict_segs[k][sk][c] + '")' for c in dict_segs[k][sk].keys()
        ])).cast("int").alias(sk) 
        for k in dict_segs.keys() for sk in dict_segs[k].keys()
    ]
).show()

+------+------+------+------+---+---+---+---+---+
|  col1|  col2|  col3|  col4|  a|  b|  c|  d|  f|
+------+------+------+------+---+---+---+---+---+
|value1|value2|value3|value4|  1|  1|  1|  1|  1|
|value5|value6|value7|value8|  0|  0|  0|  0|  0|
+------+------+------+------+---+---+---+---+---+

• "*"允许您保留以前存在的所有列，可以替换为df.columns。
• alias(sk)允许您为新列sk命名
• cast("int")将类型boolean更改为int

我真的不明白为什么你们有深度3词典，但看起来key1, key2并不是很有用。

这篇关于在Spark Dataframe中动态创建多列的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持编程学习网！