创建虚拟变量框架 pyspark

Question

我有一个 spark 数据框，例如：

|---------------------|------------------------------|
|       Brand         |       Model                  |
|---------------------|------------------------------|
|       Hyundai       |  Elentra,Creta               |
|---------------------|------------------------------|
|       Hyundai       |  Creta,Grand i10,Verna       |
|---------------------|------------------------------|
|       Maruti        |  Eritga,S-cross,Vitara Brezza|
|---------------------|------------------------------|
|       Maruti        |  Celerio,Eritga,Ciaz         |
|---------------------|------------------------------|

我想要一个这样的数据框：

|---------------------|---------|--------|--------------|--------|---------|
|       Brand         | Model0  | Model1 | Model2       | Model3 | Model4  |
|---------------------|---------|--------|--------------|--------|---------|
|       Hyundai       | Elentra | Creta  | Grand i10    | Verna  |  null   |
|---------------------|---------|--------|--------------|--------|---------|
|       Maruti        | Ertiga  | S-Cross| Vitara Brezza| Celerio|  Ciaz   |
|---------------------|---------|--------|--------------|--------|---------|

我用过这段代码：

schema = StructType([
    StructField("Brand", StringType()),StructField("Model", StringType())])
tempCSV = spark.read.csv("PATH\\Cars.csv", sep='|', schema=schema)

tempDF = tempCSV.select(
        "Brand",
        f.split("Model", ",").alias("Model"),
        f.posexplode(f.split("Model", ",")).alias("pos", "val")
    )\
    .drop("val")\
    .select(
        "Brand",
        f.concat(f.lit("Model"),f.col("pos").cast("string")).alias("name"),
        f.expr("Model[pos]").alias("val")
    )\
    .groupBy("Brand").pivot("name").agg(f.first("val")).toPandas()

但我没有得到想要的结果。而不是给出第二个表格结果：

|---------------------|---------|--------|--------------|
|       Brand         | Model0  | Model1 | Model2       |
|---------------------|---------|--------|--------------|
|       Hyundai       | Elentra | Creta  | Grand i10    |
|---------------------|---------|--------|--------------|
|       Maruti        | Ertiga  | S-Cross| Vitara Brezza|
|---------------------|---------|--------|--------------|

提前致谢。

Answer 1

发生这种情况是因为您正在旋转 pos 上的数据，该数据在同一品牌组中具有重复值。 您可以使用rownumber() 并旋转您的数据以生成所需的结果。

这是您提供的数据之上的示例代码。

df = sqlContext.createDataFrame([('Hyundai',"Elentra,Creta"),("Hyundai","Creta,Grand i10,Verna"),("Maruti","Eritga,S-cross,Vitara Brezza"),("Maruti","Celerio,Eritga,Ciaz")],("Brand","Model"))

tmpDf = df.select("Brand",f.split("Model", ",").alias("Model"),f.posexplode(f.split("Model", ",")).alias("pos", "val"))

tmpDf.createOrReplaceTempView("tbl")

seqDf = sqlContext.sql("select Brand, Model, pos, val, row_number() over(partition by Brand order by pos) as rnk from tbl")

seqDf.groupBy('Brand').pivot('rnk').agg(f.first('val'))

这将产生以下结果。

+-------+-------+-------+-------+---------+-------------+----+                  
|  Brand|      1|      2|      3|        4|            5|   6|
+-------+-------+-------+-------+---------+-------------+----+
| Maruti| Eritga|Celerio|S-cross|   Eritga|Vitara Brezza|Ciaz|
|Hyundai|Elentra|  Creta|  Creta|Grand i10|        Verna|null|
+-------+-------+-------+-------+---------+-------------+----+