在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
并不是很有用。
相关文章