pyspark piplineRDD適合Dataframe列

Question

之前我所有的人都是蟒蛇和火花世界的新人。 我從大學作業，但我卡在一個地方。 我做集羣化從我的數據，現在我有我的PipelinedRDD

集羣

aftre這樣的：現在我有cluster和我的數據幀 cluster = featurizedScaledRDD.map(lambda r: kmeansModelMllib.predict(r)) cluster = [2,1,2,0,0,0,1,2]

現在dataDf我需要適合我的cluster像一個新列dataDf

i Have:    i Need: 
    +---+---+---+  +---+---+---+-------+ 
    | x | y | z |  | x | y | z |cluster| 
    +---+---+---+  +---+---+---+-------+ 
    | 0 | 1 | 1 |  | 0 | 1 | 1 | 2 | 
    | 0 | 0 | 1 |  | 0 | 0 | 1 | 1 | 
    | 0 | 8 | 0 |  | 0 | 8 | 0 | 2 | 
    | 0 | 8 | 0 |  | 0 | 8 | 0 | 0 | 
    | 0 | 1 | 0 |  | 0 | 1 | 0 | 0 | 
    +---+---+---+  +---+---+---+-------+ 

Answer 1

您可以使用zipWithIndex,join添加索引，並將其轉換回df。

swp = lambda x: (x[1], x[0]) 

cluster.zipWithIndex().map(swp).join(dataDf.rdd.zipWithIndex().map(swp)) \ 
    .values().toDF(["cluster", "point"]) 

在某些情況下，它應該可以使用zip：

cluster.zip(dataDf.rdd).toDF(["cluster", "point"]) 

您可以.select("cluster", "point.*")遵循扁平化輸出。