Coder et de les assembler de multiples fonctions dans PySpark

J'ai un Python de la classe que j'utilise pour charger et traiter des données dans Spark. Parmi les différentes choses que je dois faire, je suis en générant une liste de variables muettes provenant de différentes colonnes dans une Étincelle dataframe. Mon problème est que je ne suis pas sûr de savoir comment bien définir une Fonction Définie par l'Utilisateur pour accomplir ce dont j'ai besoin.

Je ne ont actuellement une méthode qui, lorsqu'elles sont mappées sur la dataframe RDD, résout une partie du problème (n'oubliez pas que c'est une méthode dans un plus grand data_processor classe):

def build_feature_arr(self,table):
    # this dict has keys for all the columns for which I need dummy coding
    categories = {'gender':['1','2'], ..}

    # there are actually two differnt dataframes that I need to do this for, this just specifies which I'm looking at, and grabs the relevant features from a config file
    if table == 'users':
        iter_over = self.config.dyadic_features_to_include
    elif table == 'activty':
        iter_over = self.config.user_features_to_include

    def _build_feature_arr(row):
        result = []
        row = row.asDict()
        for col in iter_over:
            column_value = str(row[col]).lower()
            cats = categories[col]
            result += [1 if column_value and cat==column_value else 0 for cat in cats]
        return result
    return _build_feature_arr

Essentiellement ce que ce n'est, pour le dataframe, prend la variable catégorique valeurs pour les colonnes spécifiées, et renvoie une liste de valeurs de ces nouvelles variables muettes. Que signifie le code suivant:

data = data_processor(init_args)
result = data.user_data.rdd.map(self.build_feature_arr('users'))

retourne quelque chose comme:

In [39]: result.take(10)
Out[39]:
[[1, 0, 0, 0, 1, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 0],
 [1, 0, 1, 0, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [0, 1, 1, 0, 0, 0],
 [1, 0, 1, 1, 0, 0],
 [1, 0, 0, 1, 0, 0],
 [1, 0, 0, 0, 0, 1]]

C'est exactement ce que je veux en terme de génération de la liste des variables muettes que je veux, mais voici ma question: Comment puis-je soit (un) de faire un fichier UDF avec les mêmes fonctionnalités que je peux utiliser dans une Étincelle de requête SQL (ou d'une autre façon, je suppose), ou (b) de prendre les RDD résultant de la carte décrite ci-dessus et ajoutez une nouvelle colonne à la user_data dataframe?

De toute façon, ce que je dois faire est de générer un nouveau dataframe contenant les colonnes de user_data, avec une nouvelle colonne (appelons feature_array) contenant le résultat de la fonction ci-dessus (ou quelque chose d'équivalent sur le plan fonctionnel).

OriginalL'auteur moustachio | 2015-10-07

34

Spark >= 2.3

Depuis Étincelle 2.3 OneHotEncoder est dépréciée en faveur de OneHotEncoderEstimator. Si vous utilisez une version récente veuillez modifier encoder code
```
from pyspark.ml.feature import OneHotEncoderEstimator

encoder = OneHotEncoderEstimator(
    inputCols=["gender_numeric"],  
    outputCols=["gender_vector"]
)
```
Spark < 2.3

Bien, vous pouvez écrire un UDF mais pourquoi le feriez-vous? Il y a déjà un certain nombre d'outils conçus pour gérer cette catégorie de tâches:
```
from pyspark.sql import Row
from pyspark.ml.linalg import DenseVector

row = Row("gender", "foo", "bar")

df = sc.parallelize([
  row("0", 3.0, DenseVector([0, 2.1, 1.0])),
  row("1", 1.0, DenseVector([0, 1.1, 1.0])),
  row("1", -1.0, DenseVector([0, 3.4, 0.0])),
  row("0", -3.0, DenseVector([0, 4.1, 0.0]))
]).toDF()
```
Tout d'abord StringIndexer.
```
from pyspark.ml.feature import StringIndexer

indexer = StringIndexer(inputCol="gender", outputCol="gender_numeric").fit(df)
indexed_df = indexer.transform(df)
indexed_df.drop("bar").show()

## +------+----+--------------+
## |gender| foo|gender_numeric|
## +------+----+--------------+
## |     0| 3.0|           0.0|
## |     1| 1.0|           1.0|
## |     1|-1.0|           1.0|
## |     0|-3.0|           0.0|
## +------+----+--------------+
```
Prochaine OneHotEncoder:
```
from pyspark.ml.feature import OneHotEncoder

encoder = OneHotEncoder(inputCol="gender_numeric", outputCol="gender_vector")
encoded_df = encoder.transform(indexed_df)
encoded_df.drop("bar").show()

## +------+----+--------------+-------------+
## |gender| foo|gender_numeric|gender_vector|
## +------+----+--------------+-------------+
## |     0| 3.0|           0.0|(1,[0],[1.0])|
## |     1| 1.0|           1.0|    (1,[],[])|
## |     1|-1.0|           1.0|    (1,[],[])|
## |     0|-3.0|           0.0|(1,[0],[1.0])|
## +------+----+--------------+-------------+
```
VectorAssembler:
```
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(
    inputCols=["gender_vector", "bar", "foo"], outputCol="features")

encoded_df_with_indexed_bar = (vector_indexer
    .fit(encoded_df)
    .transform(encoded_df))

final_df = assembler.transform(encoded_df)
```
Si bar contenus variables catégorielles vous pouvez utiliser VectorIndexer pour définir les métadonnées requises:
```
from pyspark.ml.feature import VectorIndexer

vector_indexer = VectorIndexer(inputCol="bar", outputCol="bar_indexed")
```
mais il n'est pas le cas ici.

Enfin, vous pouvez envelopper tout cela à l'aide de pipelines:
```
from pyspark.ml import Pipeline
pipeline = Pipeline(stages=[indexer, encoder, vector_indexer, assembler])
model = pipeline.fit(df)
transformed = model.transform(df)
```
Sans doute, il est beaucoup plus robuste et propre approche de l'écriture tout à partir de zéro. Il ya quelques mises en garde en particulier quand vous en avez besoin uniforme de codage entre les différents ensembles de données. Vous pouvez en lire plus dans la documentation officielle pour StringIndexer et VectorIndexer.

Une autre façon d'obtenir une production comparable est RFormula qui:

RFormula produit d'un vecteur colonne de fonctionnalités et d'un lit double ou colonne de type chaîne de l'étiquette. Comme lorsque les formules sont utilisées dans la R pour la régression linéaire, la chaîne des colonnes d'entrée sera de chaud codé, et des colonnes numériques seront jetés à double. Si l'étiquette de la colonne est de type chaîne de caractères, il sera d'abord transformé en double avec StringIndexer. Si l'étiquette de la colonne n'existe pas dans le DataFrame, la sortie de l'étiquette de la colonne sera créé à partir de cette variable de réponse dans la formule.
```
from pyspark.ml.feature import RFormula

rf = RFormula(formula="~ gender +  bar + foo - 1")
final_df_rf = rf.fit(df).transform(df)
```
Comme vous pouvez le voir, il est beaucoup plus concis, mais plus difficile à composer ne permet pas beaucoup de personnalisation. Néanmoins, le résultat d'un simple pipeline comme celui-ci sera identique:
```
final_df_rf.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+


final_df.select("features").show(4, False)

## +----------------------+
## |features              |
## +----------------------+
## |[1.0,0.0,2.1,1.0,3.0] |
## |[0.0,0.0,1.1,1.0,1.0] |
## |(5,[2,4],[3.4,-1.0])  |
## |[1.0,0.0,4.1,0.0,-3.0]|
## +----------------------+
```
Concernant vos questions:

faire un fichier UDF avec les mêmes fonctionnalités que je peux utiliser dans une Étincelle de requête SQL (ou d'une autre façon, je suppose)

C'est juste une UDF comme les autres. Assurez-vous que vous utilisez des types pris en charge, et au-delà que tout devrait fonctionner parfaitement.

prendre la RDD résultant de la carte décrite ci-dessus et ajoutez une nouvelle colonne à la user_data dataframe?
```
from pyspark.ml.linalg import VectorUDT
from pyspark.sql.types import StructType, StructField

schema = StructType([StructField("features", VectorUDT(), True)])
row = Row("features")
result.map(lambda x: row(DenseVector(x))).toDF(schema)
```
Note:

Pour Spark 1.x remplacez pyspark.ml.linalg avec pyspark.mllib.linalg.

Dans ce contexte particulier, c'est parce que l'OP veut obtenir des variables muettes (comme model.matrix dans R). Plus susceptibles de former un certain type de modèle linéaire. Rish explication - la chaîne de l'indexeur genre de crée comme facteur de colonne à partir de chaînes, l'une chaude appels model.matrix 🙂
merci @zero323! Juste une remarque: à partir de Spark 2.0+ from pyspark.mllib.linalg import DenseVector doit être remplacé par from pyspark.ml.linalg import DenseVector, sinon, vous aurez des erreurs de types dans le VectorIndexer stade
Précisé, merci.
J'ai une question....si je lance un randomforest_Classifier ce sont des données que je voudrais obtenir la forêt aléatoire feuilles en termes de nombre de personnes(en raison de l'indexation). Comment puis-je attacher en arrière à l'original, les descriptions (j'.e texte en anglais) dans une belle façon. Par exemple, la forêt aléatoire classificateur n'a pas meatadata et il devient une tâche difficile. J'ai une vague idée que je dois utiliser quelque chose comme IndexToString() mais je ne suis pas sûr de la façon de l'utiliser
pouvez-vous expliquer comment lire les gender_vector après oneHotEncoding dans votre exemple?

OriginalL'auteur

Vous devez vous connecter pour publier un commentaire.