Comment changer les noms de colonne Dataframe dans Pyspark?

Je viens du milieu des pandas et j'ai l'habitude de lire les données des fichiers CSV dans une trame de données, puis de changer simplement les noms de colonne en quelque chose d'utile à l'aide de la commande simple:

df.columns = new_column_name_list

Cependant, la même chose ne fonctionne pas dans les cadres de données pyspark créés à l'aide de sqlContext. La seule solution que j'ai pu trouver pour le faire facilement est la suivante:

df = sqlContext.read.format("com.databricks.spark.csv").options(header='false', inferschema='true', delimiter='\t').load("data.txt")
oldSchema = df.schema
for i,k in enumerate(oldSchema.fields):
  k.name = new_column_name_list[i]
df = sqlContext.read.format("com.databricks.spark.csv").options(header='false', delimiter='\t').load("data.txt", schema=oldSchema)

Il s'agit essentiellement de définir la variable deux fois et de déduire le schéma en premier, puis de renommer les noms de colonne, puis de charger à nouveau le cadre de données avec le schéma mis à jour.

Existe-t-il un moyen meilleur et plus efficace de procéder comme nous le faisons chez les pandas?

Ma version spark est 1.5.0

Il y a plusieurs façons de procéder:

• Option 1. Utilisation de selectExpr .

  data = sqlContext.createDataFrame([("Alberto", 2), ("Dakota", 2)], 
                                    ["Name", "askdaosdka"])
  data.show()
  data.printSchema()
  
  # Output
  #+-------+----------+
  #|   Name|askdaosdka|
  #+-------+----------+
  #|Alberto|         2|
  #| Dakota|         2|
  #+-------+----------+
  
  #root
  # |-- Name: string (nullable = true)
  # |-- askdaosdka: long (nullable = true)
  
  df = data.selectExpr("Name as name", "askdaosdka as age")
  df.show()
  df.printSchema()
  
  # Output
  #+-------+---+
  #|   name|age|
  #+-------+---+
  #|Alberto|  2|
  #| Dakota|  2|
  #+-------+---+
  
  #root
  # |-- name: string (nullable = true)
  # |-- age: long (nullable = true)

• Option 2. En utilisant withColumnRenamed , notez que cette méthode vous permet de "remplacer" la même colonne. Pour Python3, remplacez xrangepar range.

  from functools import reduce
  
  oldColumns = data.schema.names
  newColumns = ["name", "age"]
  
  df = reduce(lambda data, idx: data.withColumnRenamed(oldColumns[idx], newColumns[idx]), xrange(len(oldColumns)), data)
  df.printSchema()
  df.show()

• Option 3. en utilisant l' alias , dans Scala, vous pouvez également utiliser comme .

  from pyspark.sql.functions import col
  
  data = data.select(col("Name").alias("name"), col("askdaosdka").alias("age"))
  data.show()
  
  # Output
  #+-------+---+
  #|   name|age|
  #+-------+---+
  #|Alberto|  2|
  #| Dakota|  2|
  #+-------+---+

• Option 4. Utilisation de sqlContext.sql , qui vous permet d'utiliser des requêtes SQL sur DataFramesdes tables enregistrées.

  sqlContext.registerDataFrameAsTable(data, "myTable")
  df2 = sqlContext.sql("SELECT Name AS name, askdaosdka as age from myTable")
  
  df2.show()
  
  # Output
  #+-------+---+
  #|   name|age|
  #+-------+---+
  #|Alberto|  2|
  #| Dakota|  2|
  #+-------+---+

df = df.withColumnRenamed("colName", "newColName")
       .withColumnRenamed("colName2", "newColName2")

Avantage d'utiliser de cette façon: Avec une longue liste de colonnes, vous ne souhaitez modifier que quelques noms de colonnes. Cela peut être très pratique dans ces scénarios. Très utile pour joindre des tables avec des noms de colonne en double.

Si vous souhaitez modifier tous les noms de colonnes, essayez df.toDF(*cols)

Dans le cas où vous souhaitez appliquer une transformation simple sur tous les noms de colonnes, ce code fait l'affaire: (je remplace tous les espaces par un trait de soulignement)

new_column_name_list= list(map(lambda x: x.replace(" ", "_"), df.columns))

df = df.toDF(*new_column_name_list)

Merci à @ user8117731 pour l' toDfastuce.

Si vous souhaitez renommer une seule colonne et conserver le reste tel quel:

from pyspark.sql.functions import col
new_df = old_df.select(*[col(s).alias(new_name) if s == column_to_change else s for s in old_df.columns])

df.withColumnRenamed('age', 'age2')

c'est l'approche que j'ai utilisée:

créer une session pyspark:

import pyspark
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName('changeColNames').getOrCreate()

créer un cadre de données:

df = spark.createDataFrame(data = [('Bob', 5.62,'juice'),  ('Sue',0.85,'milk')], schema = ["Name", "Amount","Item"])

afficher df avec les noms des colonnes:

df.show()
+----+------+-----+
|Name|Amount| Item|
+----+------+-----+
| Bob|  5.62|juice|
| Sue|  0.85| milk|
+----+------+-----+

créer une liste avec de nouveaux noms de colonnes:

newcolnames = ['NameNew','AmountNew','ItemNew']

changez les noms des colonnes du df:

for c,n in zip(df.columns,newcolnames):
    df=df.withColumnRenamed(c,n)

afficher df avec de nouveaux noms de colonnes:

df.show()
+-------+---------+-------+
|NameNew|AmountNew|ItemNew|
+-------+---------+-------+
|    Bob|     5.62|  juice|
|    Sue|     0.85|   milk|
+-------+---------+-------+

J'ai créé une fonction facile à utiliser pour renommer plusieurs colonnes pour une trame de données pyspark, au cas où quelqu'un voudrait l'utiliser:

def renameCols(df, old_columns, new_columns):
    for old_col,new_col in zip(old_columns,new_columns):
        df = df.withColumnRenamed(old_col,new_col)
    return df

old_columns = ['old_name1','old_name2']
new_columns = ['new_name1', 'new_name2']
df_renamed = renameCols(df, old_columns, new_columns)

Attention, les deux listes doivent avoir la même longueur.

Une autre façon de renommer une seule colonne (en utilisant import pyspark.sql.functions as F):

df = df.select( '*', F.col('count').alias('new_count') ).drop('count')

J'utilise celui-ci:

from pyspark.sql.functions import col
df.select(['vin',col('timeStamp').alias('Date')]).show()

Vous pouvez utiliser la fonction suivante pour renommer toutes les colonnes de votre trame de données.

def df_col_rename(X, to_rename, replace_with):
    """
    :param X: spark dataframe
    :param to_rename: list of original names
    :param replace_with: list of new names
    :return: dataframe with updated names
    """
    import pyspark.sql.functions as F
    mapping = dict(zip(to_rename, replace_with))
    X = X.select([F.col(c).alias(mapping.get(c, c)) for c in to_rename])
    return X

Si vous ne devez mettre à jour que les noms de quelques colonnes, vous pouvez utiliser le même nom de colonne dans la liste replace_with

Pour renommer toutes les colonnes

df_col_rename(X,['a', 'b', 'c'], ['x', 'y', 'z'])

Pour renommer certaines colonnes

df_col_rename(X,['a', 'b', 'c'], ['a', 'y', 'z'])

Pour renommer une seule colonne, vous pouvez toujours utiliser toDF (). Par exemple,

df1.selectExpr("SALARY*2").toDF("REVISED_SALARY").show()

Nous pouvons utiliser différentes approches pour renommer le nom de la colonne.

Commençons par créer un DataFrame simple.

df = spark.createDataFrame([("x", 1), ("y", 2)], 
                                  ["col_1", "col_2"])

Essayons maintenant de renommer col_1 en col_3. PFB quelques approches pour faire de même.

# Approach - 1 : using withColumnRenamed function.
df.withColumnRenamed("col_1", "col_3").show()

# Approach - 2 : using alias function.
df.select(df["col_1"].alias("col3"), "col_2").show()

# Approach - 3 : using selectExpr function.
df.selectExpr("col_1 as col_3", "col_2").show()

# Rename all columns
# Approach - 4 : using toDF function. Here you need to pass the list of all columns present in DataFrame.
df.toDF("col_3", "col_2").show()

Voici la sortie.

+-----+-----+
|col_3|col_2|
+-----+-----+
|    x|    1|
|    y|    2|
+-----+-----+

J'espère que ça aide.