将字符串类型的列转换为结构，并使用PySpark解压该列

Question

输入自由度：

+------+-----------------------------------------------------+
|rowNum|infoCol                                              |
+------+-----------------------------------------------------+
|100   |[('john', 'customer'), ('abc, mno, xyz', 'purchase')]|
|200   |[('doe', 'customer')]                                |
+------+-----------------------------------------------------+

root
 |-- rowNum: string (nullable = false)
 |-- infoCol: string (nullable = false)

（预期）输出DF：

+------+--------+-----------------+
|rowNum|customer|         purchase|
+------+--------+-----------------+
|   100|['john']|['abc, mno, xyz']|
|   100| ['doe']|             null|
+------+--------+-----------------+

我已经尝试使用< code>split函数，但它并不完全符合我的需求。

inputdf = spark.createDataFrame(
    [
        ("100", "[('john', 'customer'), ('abc, mno, xyz', 'purchase')]"),
        ("200", "[('doe', 'customer')]"),
    ],
    ['rowNum', 'infoCol'] 
)

from pyspark.sql.functions import col, regexp_replace, split
outputdf = inputdf.withColumn("newcol", split(col("infoCol"), ","))

Answer 1

这是我的尝试，这可以用于许多列，不仅用于客户，购买，而且如果列名称在最后。

import pyspark.sql.functions as f

df = inputdf \
  .withColumn('infoCol', f.regexp_replace('infoCol', '[\[\]]', '')) \
  .withColumn('infoCol', f.regexp_replace('infoCol', '(\),)', ') ,')) \
  .withColumn('infoCol', f.explode(f.split('infoCol', ' , '))) \
  .withColumn('infoCol', f.regexp_replace('infoCol', '[\(\)]', '')) \
  .withColumn('infoCol', f.regexp_replace('infoCol', '(\',)', '\' ,')) \
  .withColumn('cols', f.split('infoCol', ' , ')[1]) \
  .withColumn('cols', f.regexp_replace('cols', '\'', '')) \
  .withColumn('infoCol', f.split('infoCol', ' , ')[0]) \
  .withColumn('infoCol', f.concat(f.lit('['), f.col('infoCol'), f.lit(']'))) \

values = df.select('cols').distinct().rdd.map(lambda x: x.cols).collect()

df.groupBy('rowNum') \
  .pivot('cols', values) \
  .agg(f.first('infoCol')) \
  .show(10, False)

+------+--------+-----------------+
|rowNum|customer|purchase         |
+------+--------+-----------------+
|200   |['doe'] |null             |
|100   |['john']|['abc, mno, xyz']|
+------+--------+-----------------+

Answer 2

这是我对火花内置函数的尝试。

这里的想法是首先创建2列与客户，购买作为值和其他值在另一列，以获得这些列我使用拆分然后爆炸。

一旦我们得到客户，购买值然后分组通过PIVot来透视数据，最终拆分列以获得数组。

示例：

inputdf = spark.createDataFrame(
    [
        ("100", "[('john', 'customer'), ('abc, mno, xyz', 'purchase')]"),
        ("200", "[('doe', 'customer')]"),
    ],
    ['rowNum', 'infoCol'] 
)

from pyspark.sql.functions import *

inputdf.withColumn("newcol", split(col("infoCol"), "\),")).\
selectExpr("explode(newcol)","rowNum").\
withColumn("newCol1",split(regexp_replace(col("col"),"[\[|\]|\(|\)]",""),"',")).\
withColumn("new1",regexp_replace(trim(element_at(col("newCol1"),1)),"[']","")).\
withColumn("new2",regexp_replace(trim(element_at(col("newCol1"),2)),"[']","")).\
groupby("rowNum").\
pivot("new2").\
agg(first(col("new1"))).\
withColumn("customer",split(col("customer"),",")).\
withColumn("purchase",split(col("purchase"),",")).\
show()

#+------+--------+-----------------+
#|rowNum|customer|         purchase|
#+------+--------+-----------------+
#|   200|   [doe]|             null|
#|   100|  [john]|[abc,  mno,  xyz]|
#+------+--------+-----------------+

更新：

inputdf = spark.createDataFrame(
    [
        ("100", "[('john', 'customer'), ('abc, mno, xyz', 'purchase')]"),
        ("200", "[('doe', 'customer')]"),
    ],
    ['rowNum', 'infoCol'] 
)

from pyspark.sql.functions import *

inputdf.withColumn("newcol", split(col("infoCol"), "\),")).\
selectExpr("explode(newcol)","rowNum").\
withColumn("newCol1",split(regexp_replace(col("col"),"[\[|\]|\(|\)]",""),"',")).\
withColumn("new1",regexp_replace(trim(element_at(col("newCol1"),1)),"[']","")).\
withColumn("new2",regexp_replace(trim(element_at(col("newCol1"),2)),"[']","")).\
groupby("rowNum").\
pivot("new2").\
agg(first(col("new1"))).\
withColumn("customer",col("customer")).\
withColumn("purchase",col("purchase")).\
show()

#+------+--------+-------------+
#|rowNum|customer|     purchase|
#+------+--------+-------------+
#|   200|     doe|         null|
#|   100|    john|abc, mno, xyz|
#+------+--------+-------------+

更新2：

inputdf = spark.createDataFrame(
    [
        ("100", "[('john', 'customer'), ('abc, mno, xyz', 'purchase'), ('abc123', 'purchase')]"),
        ("200", "[('doe', 'customer')]"),
    ],
    ['rowNum', 'infoCol'] 
)

from pyspark.sql.functions import *


inputdf.withColumn("newcol", split(col("infoCol"), "\),")).\
selectExpr("explode(newcol)","rowNum").\
withColumn("newCol1",expr("""transform(split(regexp_replace(col,"[\[|\]|\(|\)]",""),"',"),x -> regexp_replace(trim(x),"[']",""))""")).\
withColumn("new1",regexp_replace(element_at(col("newCol1"),-1),"[\]]","")).\
withColumn("new2",array_except(col("newCol1"),array(lit('purchase'),lit('customer'),lit('purchase]'),lit('customer]')))).\
withColumn("new2",expr("""transform(new2,x -> concat("'",regexp_replace(x,"[\\\\[]",""),"'"))""")).\
drop(*['col','newCol1']).\
groupby("new1","rowNum").agg(flatten(collect_list(col("new2"))).alias("new2")).\
groupby("rowNum").pivot("new1").agg(first(col("new2"))).\
show(10,False)

#+------+--------+---------------------------+
#|rowNum|customer|purchase                   |
#+------+--------+---------------------------+
#|200   |['doe'] |null                       |
#|100   |['john']|['abc, mno, xyz', 'abc123']|
#+------+--------+---------------------------+