使用withColumn函数从PySpark中的列表动态创建新列

Question

我试图弄清楚如何通过使用与哥伦布（）函数并在pySpark中的cp_codeset列（）函数中调用udf来动态地为列表中的每个项目创建列（在这种情况下为列表）。下面是我写的代码，但它给了我一个错误。

from pyspark.sql.functions import udf, col, lit
from pyspark.sql import Row
from pyspark.sql.types import IntegerType


codeset = set(cp_codeset['CODE'])

for col_name in cp_codeset.col_names.unique():
    def flag(d):
        if (d in codeset):
            name = cp_codeset[cp_codeset['CODES']==d].col_names
            if(name==col_name):
                return 1
            else:
                return 0

    cpf_udf = udf(flag, IntegerType())

    p.withColumn(col_name, cpf_udf(p.codes)).show()

另一种选择是手动执行它，但在这种情况下，我必须编写相同的udf函数并使用养分（）函数调用它75次（这是cp_codeset["col_names"]的大小）

下面是我的两个数据帧，我正在尝试获得结果如何显示

id|codes
1|100
2|102
3|104

codes| col_names
100|a
101|b
102|c
103|d
104|e
105|f

id|codes|a|c|e
1|100   |1|0|0
2|102   |0|1|0   
3|104   |0|0|1

Answer 1

我将使用get_dummies和joinmap

m = cp_codeset.set_index('codes').col_names

P.join(pd.get_dummies(P.codes.map(m)))

   id  codes  a  c  e
0   1    100  1  0  0
1   2    102  0  1  0
2   3    104  0  0  1

Answer 2

过滤此数据后：

cp_codeset.set_index('codes').loc[p.codes]
Out[44]: 
      col_names
codes          
100           a
102           c
104           e

只需使用get_dummies：

pd.get_dummies(cp_codeset.set_index('codes').loc[p.codes])
Out[45]: 
       col_names_a  col_names_c  col_names_e
codes                                       
100              1            0            0
102              0            1            0
104              0            0            1