¿Cómo cambiar los nombres de columna del marco de datos en pyspark?

Vengo de antecedentes de pandas y estoy acostumbrado a leer datos de archivos CSV en un marco de datos y luego simplemente cambiar los nombres de columna a algo útil usando el comando simple:

df.columns = new_column_name_list

Sin embargo, lo mismo no funciona en los marcos de datos pyspark creados con sqlContext. La única solución que podría resolver para hacer esto fácilmente es la siguiente:

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)

Básicamente, esto es definir la variable dos veces e inferir el esquema primero, luego renombrar los nombres de columna y luego cargar el marco de datos nuevamente con el esquema actualizado.

¿Hay una manera mejor y más eficiente de hacer esto como lo hacemos en los pandas?

Mi versión de chispa es 1.5.0

Hay muchas formas de hacerlo:

• Opción 1. Usando 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)
  

• Opción 2. Usando withColumnRenamed , observe que este método le permite "sobrescribir" la misma columna. Para Python3, reemplace xrangecon 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()
  

• Opción 3. usando alias , en Scala también puede usar como .

  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|
  #+-------+---+
  

• Opción 4. Uso de sqlContext.sql , que le permite usar consultas SQL en DataFramestablas registradas.

  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")

Ventaja de usar de esta manera: con una larga lista de columnas, le gustaría cambiar solo unos pocos nombres de columna. Esto puede ser muy conveniente en estos escenarios. Muy útil al unir tablas con nombres de columna duplicados.

Si desea cambiar todos los nombres de columnas, intente df.toDF(*cols)

En caso de que desee aplicar una transformación simple en todos los nombres de columna, este código funciona: (Estoy reemplazando todos los espacios con guiones bajos)

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

df = df.toDF(*new_column_name_list)

Gracias a @ user8117731 por el toDftruco.

Si desea cambiar el nombre de una sola columna y mantener el resto como está:

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')

Este es el enfoque que utilicé:

crear sesión de pyspark:

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

crear marco de datos:

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

ver df con nombres de columna:

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

crear una lista con nuevos nombres de columna:

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

cambiar los nombres de columna de la df:

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

ver df con nuevos nombres de columna:

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

Hice una función fácil de usar para cambiar el nombre de varias columnas para un marco de datos pyspark, en caso de que alguien quiera usarlo:

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)

Tenga cuidado, ambas listas deben tener la misma longitud.

Otra forma de renombrar solo una columna (usando import pyspark.sql.functions as F):

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

Yo uso este:

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

Puede usar la siguiente función para cambiar el nombre de todas las columnas de su marco de datos.

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

En caso de que necesite actualizar solo los nombres de algunas columnas, puede usar el mismo nombre de columna en la lista replace_with

Para renombrar todas las columnas

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

Para renombrar algunas columnas

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

Para cambiar el nombre de una sola columna, aún puede usar toDF (). Por ejemplo,

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

Podemos usar varios enfoques para cambiar el nombre de la columna.

Primero, dejemos crear un DataFrame simple.

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

Ahora intentemos renombrar col_1 a col_3. PFB algunos enfoques para hacer lo mismo.

# 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()

Aquí está la salida.

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

Espero que esto ayude.