Creare labeledPoints da Spark DataFrame in Python

Question

Quale funzione .map() in python devo usare per creare un set di labeledPoints da un frame di dati spark? Qual è la notazione se L'etichetta/risultato non è la prima colonna ma posso fare riferimento al suo nome di colonna, 'stato'?

ho creare la dataframe pitone con questa funzione .map():

def parsePoint(line): 
    listmp = list(line.split('\t')) 
    dataframe = pd.DataFrame(pd.get_dummies(listmp[1:]).sum()).transpose() 
    dataframe.insert(0, 'status', dataframe['accepted']) 
    if 'NULL' in dataframe.columns: 
     dataframe = dataframe.drop('NULL', axis=1) 
    if '' in dataframe.columns: 
     dataframe = dataframe.drop('', axis=1) 
    if 'rejected' in dataframe.columns: 
     dataframe = dataframe.drop('rejected', axis=1) 
    if 'accepted' in dataframe.columns: 
     dataframe = dataframe.drop('accepted', axis=1) 
    return dataframe 

ho convertirlo in un dataframe scintilla dopo che la funzione di ridurre le ha ricombinato tutti i dataframes panda.

parsedData=sqlContext.createDataFrame(parsedData) 

Ma ora come posso creare punti labled da questo in python? Presumo che potrebbe essere un'altra funzione .map()?

Answer 1

Se si dispone già di caratteristiche numeriche e che non necessitano di ulteriori trasformazioni è possibile utilizzare VectorAssembler per combinare le colonne contenenti variabili indipendenti:

from pyspark.ml.feature import VectorAssembler 

assembler = VectorAssembler(
    inputCols=["your", "independent", "variables"], 
    outputCol="features") 

transformed = assembler.transform(parsedData) 

Poi si può semplicemente mappare:

from pyspark.mllib.regression import LabeledPoint 
from pyspark.sql.functions import col 

(transformed.select(col("outcome_column").alias("label"), col("features")) 
    .rdd 
    .map(lambda row: LabeledPoint(row.label, row.features))) 

Come di Spark 2.0 ml e mllib API non sono più compatibili e il secondo sta andando verso la deprecazione e la rimozione. Se hai ancora bisogno di questo devi convertire ml.Vectors a mllib.Vectors.

from pyspark.mllib import linalg as mllib_linalg 
from pyspark.ml import linalg as ml_linalg 

def as_old(v): 
    if isinstance(v, ml_linalg.SparseVector): 
     return mllib_linalg.SparseVector(v.size, v.indices, v.values) 
    if isinstance(v, ml_linalg.DenseVector): 
     return mllib_linalg.DenseVector(v.values) 
    raise ValueError("Unsupported type {0}".format(type(v))) 

e mappa:

lambda row: LabeledPoint(row.label, as_old(row.features)))