La construction d'une ligne à partir d'un dict en pySpark

Je suis en train de créer dynamiquement une ligne dans pySpark 1.6.1, puis de l'intégrer dans un dataframe. L'idée générale est d'étendre les résultats de describe pour inclure, par exemple, l'inclinaison et le coefficient d'aplatissement. Voici ce que j'ai pensé devrait fonctionner:

from pyspark.sql import Row

row_dict = {'C0': -1.1990072635132698,
            'C3': 0.12605772684660232,
            'C4': 0.5760856026559944,
            'C5': 0.1951877800894315,
            'C6': 24.72378589441825,
            'summary': 'kurtosis'}

new_row = Row(row_dict)

Mais cela renvoie TypeError: sequence item 0: expected string, dict found qui est assez clair erreur. Ensuite, j'ai trouvé que si je définis les champs de Ligne tout d'abord, je pourrait utiliser un dict:

r = Row('summary', 'C0', 'C3', 'C4', 'C5', 'C6')
r(row_dict)
> Row(summary={'summary': 'kurtosis', 'C3': 0.12605772684660232, 'C0': -1.1990072635132698, 'C6': 24.72378589441825, 'C5': 0.1951877800894315, 'C4': 0.5760856026559944})

Qui serait un bel étape, sauf qu'il ne semble pas comme je peux dynamiquement spécifier les champs dans Row. J'ai besoin de ce travail pour un nombre inconnu de lignes avec des noms inconnus. Selon la documentation, vous pouvez effectivement aller dans l'autre sens:

>>> Row(name="Alice", age=11).asDict() == {'name': 'Alice', 'age': 11}
True

Donc il semble que je devrais être capable de faire cela. Il apparaît aussi il peut y avoir certaines fonctionnalités dépréciées à partir d'anciennes versions qui ont permis, par exemple ici. Est-il plus équivalent actuel je suis absent?

OriginalL'auteur Jeff | 2016-07-07

23

Vous pouvez utiliser des mots clés les arguments déballage comme suit:
```
Row(**row_dict)

## Row(C0=-1.1990072635132698, C3=0.12605772684660232, C4=0.5760856026559944, 
##     C5=0.1951877800894315, C6=24.72378589441825, summary='kurtosis')
```
Il est important de noter qu'il en interne trie les données par clé à l'adresse des problèmes avec les anciennes versions de Python.

Est-ce valable à partir d'une version spécifique de Python ou est-ce une règle générale ? La raison pour laquelle je me demande est en raison de votre la dernière modification.
Depuis Étincelle va toujours de tri en interne, il n'a tout simplement pas d'importance ce que nous faisons avant que. Et compte tenu de JIRA discussion, il ne changera pas jusqu'à ce que l'Étincelle en charge de gouttes pour Python < 3.6 (et non tout moment bientôt). OrderedDict était un peu trompeur, donc je l'ai enlevé.
Ok merci ! C'était à l'origine de ma confusion.
Je crois que ma première idée était d'utiliser *args (Row implémente __new__ et prend le chemin d'accès différent avec des arguments de position) pas **kwargs constructeurs - c'est pourquoi j'ai inclus le tri, et en quelque sorte il a fini dans la réponse.
cela peut-il être utilisé avec la version de python 2.7 et imbriquées Dictionnaire? Mon but était de convertir twitter DStream à la Ligne CA alors DF comme je suis usinf Spark version 1.6.

OriginalL'auteur zero323

Dans le cas où le dict n'est pas aplatir, vous pouvez convertir dict à la Ligne de manière récursive.

def as_row(obj):
    if isinstance(obj, dict):
        dictionary = {k: as_row(v) for k, v in obj.items()}
        return Row(**dictionary)
    elif isinstance(obj, list):
        return [as_row(v) for v in obj]
    else:
        return obj

OriginalL'auteur ryan

Vous devez vous connecter pour publier un commentaire.