Wie ändere ich die Namen von Dataframe-Spalten in pyspark?

Ich komme aus dem Pandas-Hintergrund und bin es gewohnt, Daten aus CSV-Dateien in einen Datenrahmen zu lesen und dann einfach die Spaltennamen mit dem einfachen Befehl in etwas Nützliches zu ändern:

df.columns = new_column_name_list

Dies funktioniert jedoch nicht in pyspark-Datenrahmen, die mit sqlContext erstellt wurden. Die einzige Lösung, die ich finden könnte, um dies leicht zu tun, ist die folgende:

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)

Dies bedeutet im Grunde, die Variable zweimal zu definieren und zuerst auf das Schema zu schließen, dann die Spaltennamen umzubenennen und dann den Datenrahmen erneut mit dem aktualisierten Schema zu laden.

Gibt es einen besseren und effizienteren Weg, dies zu tun, wie wir es bei Pandas tun?

Meine Funkenversion ist 1.5.0

Dafür gibt es viele Möglichkeiten:

• Option 1. Verwenden von 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. Beachten Sie bei Verwendung von withColumnRenamed , dass Sie mit dieser Methode dieselbe Spalte "überschreiben" können. Für Python3, ersetzen xrangemit 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. Mit Alias können Sie in Scala auch als verwenden .

  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. Verwenden von sqlContext.sql , mit dem Sie SQL-Abfragen für DataFramesals Tabellen registrierte Tabellen verwenden können.

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

Vorteil dieser Verwendung: Bei einer langen Liste von Spalten möchten Sie nur wenige Spaltennamen ändern. Dies kann in diesen Szenarien sehr praktisch sein. Sehr nützlich beim Verknüpfen von Tabellen mit doppelten Spaltennamen.

Wenn Sie alle Spaltennamen ändern möchten, versuchen Sie es df.toDF(*cols)

Wenn Sie eine einfache Transformation auf alle Spaltennamen anwenden möchten, führt dieser Code den folgenden Trick aus: (Ich ersetze alle Leerzeichen durch Unterstriche.)

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

df = df.toDF(*new_column_name_list)

Danke an @ user8117731 für den toDfTrick.

Wenn Sie eine einzelne Spalte umbenennen und den Rest unverändert lassen möchten:

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

Dies ist der Ansatz, den ich verwendet habe:

Pyspark-Sitzung erstellen:

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

Datenrahmen erstellen:

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

Ansicht df mit Spaltennamen:

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

Erstellen Sie eine Liste mit neuen Spaltennamen:

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

Ändern Sie die Spaltennamen des df:

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

Ansicht df mit neuen Spaltennamen:

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

Ich habe eine benutzerfreundliche Funktion zum Umbenennen mehrerer Spalten für einen pyspark-Datenrahmen erstellt, falls jemand sie verwenden möchte:

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)

Seien Sie vorsichtig, beide Listen müssen gleich lang sein.

Eine andere Möglichkeit, nur eine Spalte umzubenennen (mit import pyspark.sql.functions as F):

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

Ich benutze dieses:

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

Mit der folgenden Funktion können Sie alle Spalten Ihres Datenrahmens umbenennen.

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

Wenn Sie nur die Namen einiger Spalten aktualisieren müssen, können Sie denselben Spaltennamen in der Liste replace_with verwenden

Um alle Spalten umzubenennen

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

Um einige Spalten umzubenennen

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

Für eine einzelne Spaltenumbenennung können Sie weiterhin toDF () verwenden. Beispielsweise,

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

Wir können verschiedene Ansätze verwenden, um den Spaltennamen umzubenennen.

Erstellen Sie zunächst einen einfachen DataFrame.

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

Versuchen wir nun, col_1 in col_3 umzubenennen. PFB einige Ansätze, um das gleiche zu tun.

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

Hier ist die Ausgabe.

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

Ich hoffe das hilft.