Convertir les lignes en Dictionnaire dans pyspark

Question

J'ai un DataFrame(df) dans pyspark, en lisant à partir d'une table de ruche :

df=spark.sql('select * from <table_name>')

+++++++++++++++++++++++++++++++++++++++++++
|  Name    |    URL visited               |
+++++++++++++++++++++++++++++++++++++++++++
|  person1 | [google,msn,yahoo]           |
|  person2 | [fb.com,airbnb,wired.com]    |
|  person3 | [fb.com,google.com]          |
+++++++++++++++++++++++++++++++++++++++++++

Quand j'ai essayé ce qui suit, j'ai eu une erreur

df_dict = dict(zip(df['name'],df['url']))
"TypeError: zip argument #1 must support iteration."

type(df.name) is of 'pyspark.sql.column.Column'

Comment puis-je créer un dictionnaire comme le suivant, qui peut être itéré par la suite ?

{'person1':'google','msn','yahoo'}
{'person2':'fb.com','airbnb','wired.com'}
{'person3':'fb.com','google.com'}

J'apprécie vos réflexions et votre aide.

Answer 1

Je pense que vous pouvez essayer row.asDict() ce code s'exécute directement sur l'exécuteur, et vous n'avez pas besoin de collecter les données sur le pilote.

Quelque chose comme :

df.rdd.map(lambda row: row.asDict())

Answer 2

Que diriez-vous d'utiliser le pyspark Row.as_Dict() méthode ? Cette méthode fait partie de l'API dataframe (qui, si j'ai bien compris, est l'API "recommandée" au moment de la rédaction du présent document) et ne nécessite pas du tout l'utilisation de l'API RDD.

df_list_of_dict = [row.asDict() for row in df.collect()]

type(df_list_of_dict), type(df_list_of_dict[0])
#(<class 'list'>, <class 'dict'>)

df_list_of_dict
#[{'person1': ['google','msn','yahoo']},
# {'person2': ['fb.com','airbnb','wired.com']},
# {'person3': ['fb.com','google.com']}]

Answer 3

Si vous vouliez vos résultats dans un dictionnaire python, vous pourriez utiliser collect() 1 pour amener les données dans la mémoire locale et ensuite masser la sortie comme souhaité.

Commencez par collecter les données :

df_dict = df.collect()
#[Row(Name=u'person1', URL visited=[u'google', u'msn,yahoo']),
# Row(Name=u'person2', URL visited=[u'fb.com', u'airbnb', u'wired.com']),
# Row(Name=u'person3', URL visited=[u'fb.com', u'google.com'])]

Cela renvoie une liste de pyspark.sql.Row objets. Vous pouvez facilement convertir cela en une liste de dict s :

df_dict = [{r['Name']: r['URL visited']} for r in df_dict]
#[{u'person1': [u'google', u'msn,yahoo']},
# {u'person2': [u'fb.com', u'airbnb', u'wired.com']},
# {u'person3': [u'fb.com', u'google.com']}]

1 Sachez que pour les grands ensembles de données, cette opération peut être lente et potentiellement échouer avec une erreur de mémoire insuffisante. Vous devez d'abord vous demander si c'est vraiment ce que vous voulez faire car vous perdrez les avantages de la parallélisation de spark en amenant les données en mémoire locale.

Answer 4

Étant donné :

+++++++++++++++++++++++++++++++++++++++++++
|  Name    |    URL visited               |
+++++++++++++++++++++++++++++++++++++++++++
|  person1 | [google,msn,yahoo]           |
|  person2 | [fb.com,airbnb,wired.com]    |
|  person3 | [fb.com,google.com]          |
+++++++++++++++++++++++++++++++++++++++++++

Cela devrait fonctionner :

df_dict = df \
    .rdd \
    .map(lambda row: {row[0]: row[1]}) \
    .collect()

df_dict

#[{'person1': ['google','msn','yahoo']},
# {'person2': ['fb.com','airbnb','wired.com']},
# {'person3': ['fb.com','google.com']}]

De cette façon, vous ne faites que collecter après le traitement.

S'il vous plaît, faites-moi savoir si cela fonctionne pour vous :)