Как извлечь вектор из строки с помощью PySpark

Question

Я пытаюсь запустить логистическую регрессию с образцами данных с помощью PySpark. Облицовка проблемы при применении «LabeledPoint» после хеширования.

Входной кадр данных:

+--+--------+ 
|C1|  C2| 
+--+--------+ 
| 0|776ce399| 
| 0|3486227d| 
| 0|e5ba7672| 
| 1|3486227d| 
| 0|e5ba7672| 
+--+--------+ 

После применения хэширования на колонке С2,

tokenizer = Tokenizer(inputCol="C2", outputCol="words") 
wordsData = tokenizer.transform(df) 
hashingTF = HashingTF(inputCol="words", outputCol="rawFeatures", numFeatures=20) 
featurizedData = hashingTF.transform(wordsData) 
idf = IDF(inputCol="rawFeatures", outputCol="features") 
idfModel = idf.fit(featurizedData) 
rescaledData = idfModel.transform(featurizedData) 



+--+--------+--------------------+---------------+--------------------+ 
|C1|  C2|    words| rawFeatures|   features| 
+--+--------+--------------------+---------------+--------------------+ 
| 0|776ce399|ArrayBuffer(776ce...|(20,[15],[1.0])|(20,[15],[2.30003...| 
| 0|3486227d|ArrayBuffer(34862...| (20,[0],[1.0])|(20,[0],[2.455603...| 
| 0|e5ba7672|ArrayBuffer(e5ba7...| (20,[9],[1.0])|(20,[9],[0.660549...| 
| 1|3486227d|ArrayBuffer(34862...| (20,[0],[1.0])|(20,[0],[2.455603...| 
| 0|e5ba7672|ArrayBuffer(e5ba7...| (20,[9],[1.0])|(20,[9],[0.660549...| 
+--+--------+--------------------+---------------+--------------------+ 

теперь применить логистическую регрессию, когда я выполнить LabeledPoint

temp = rescaledData.map(lambda line: LabeledPoint(line[0],line[4]))

получает следующее ошибка,

ValueError: setting an array element with a sequence.

Помогите.

Answer 1

Спасибо за предложение zero323.

Выполнено с использованием концепции трубопровода.

from pyspark.ml import Pipeline 
from pyspark.ml.classification import LogisticRegression 
from pyspark.ml.feature import HashingTF, Tokenizer, IDF 
from pyspark.sql import Row 
from pyspark.sql.functions import col 
from pyspark.sql.types import DoubleType 



dfWithLabel = df.withColumn("label", col("C1").cast(DoubleType())) 
tokenizer = Tokenizer(inputCol="C2", outputCol="D2") 
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="E2") 
idf = IDF(inputCol=hashingTF.getOutputCol(), outputCol="features") 
lr = LogisticRegression(maxIter=10, regParam=0.01) 
pipeline = Pipeline(stages=[tokenizer, hashingTF,idf,lr]) 

# Fit the pipeline to training documents. 
model = pipeline.fit(dfWithLabel)