Pyspark에서 데이터 프레임 열 이름을 변경하는 방법

Pyspark에서 데이터 프레임 열 이름을 변경하는 방법

나는 판다 출신이고 CSV 파일의 데이터를 데이터 프레임으로 읽고, 간단한 명령을 사용하여 칼럼 이름을 유용한 것으로 바꾸는 것에 익숙하다.

df.columns = new_column_name_list

그러나 sqlContext를 사용하여 만든 pyspark 데이터프레임에서는 같은 기능이 작동하지 않는다.내가 이것을 쉽게 할 수 있는 유일한 해결책은 다음과 같다.

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)

이것은 기본적으로 변수를 두 번 정의하고 먼저 스키마를 유추한 다음 열 이름을 바꾼 다음 업데이트된 스키마로 데이터 프레임을 다시 로드하는 것이다.

우리가 팬더에서 하는 것처럼 이것을 하는 더 좋고 더 효율적인 방법이 있을까?

나의 스파크 버전은 1.5.0이다.

이를 위한 여러 가지 방법이 있다.

• 옵션 1.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)
  

• 옵션 2.ColumnRenamed와 함께 사용하면 이 방법을 사용하여 동일한 열을 "덮어쓸" 수 있다는 점에 유의하십시오.Python3의 경우 교체xrange와 함께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()
  

• 옵션 3. 별칭을 사용하여 스칼라에서 다음과 같이 사용할 수 있다.

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

• 옵션 4.sqlContext 사용.SQL(SQL 쿼리를 사용할 수 있음DataFrames테이블로 등록되어 있다.

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

이 방법의 장점: 긴 열 목록을 사용하여 몇 개의 열 이름만 변경하고자 할 경우이것은 이러한 시나리오에서 매우 편리할 수 있다.중복된 열 이름으로 테이블을 조인할 때 매우 유용하다.

모든 열 이름을 변경하려면df.toDF(*cols)

모든 열 이름에 단순 변환을 적용하려는 경우 이 코드는 다음과 같은 트릭을 실행한다: (모든 공백을 밑줄로 대체함)

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

df = df.toDF(*new_column_name_list)

@user8117731 덕분에toDf속임수를 쓰다

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

단일 열의 이름을 변경하고 나머지는 그대로 유지하려면:

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

이것이 내가 사용한 접근방식이다:

Pyspark 세션 생성:

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

데이터 프레임 생성:

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

열 이름이 있는 보기 df:

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

새 열 이름으로 목록 만들기:

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

df의 열 이름 변경:

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

새 열 이름을 사용하여 df 보기:

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

나는 누구나 사용하기를 원할 경우에 대비하여 피스파크 데이터프레임에 대해 여러 열의 이름을 바꾸는 기능을 쉽게 만들었다.

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)

조심해, 두 명단의 길이가 같아야 해.

한 열만 바꾸는 다른 방법(사용)import pyspark.sql.functions as F):

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

방법 1:

df = df.withColumnRenamed("new_column_name", "old_column_name")

방법 2: 계산을 수행하고 새 값의 이름을 변경하려면

df = df.withColumn("old_column_name", F.when(F.col("old_column_name") > 1, F.lit(1)).otherwise(F.col("old_column_name"))
df = df.drop("new_column_name", "old_column_name")

다음 함수를 사용하여 데이터 프레임의 모든 열의 이름을 바꿀 수 있다.

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

몇 개의 열 이름만 업데이트해야 하는 경우 replace_with 목록에서 동일한 열 이름을 사용할 수 있음

모든 열의 이름을 변경하려면 다음과 같이 하십시오.

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

일부 열의 이름을 변경하려면 다음과 같이 하십시오.

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

난 이걸 사용한다:

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

우리는 다양한 접근 방식을 사용하여 칼럼 이름을 바꿀 수 있다.

먼저 간단한 DataFrame을 생성하십시오.

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

이제 col_1의 이름을 col_3으로 바꾸도록 합시다.PFB는 동일한 작업을 하기 위해 몇 가지 접근한다.

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

여기 산출물이 있다.

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

이게 도움이 되었으면 좋겠어.

루프를 위해 삽입할 수 있으며, zip을 사용하여 각 열 이름을 두 개의 배열로 쌍으로 구성할 수 있다.

new_name = ["id", "sepal_length_cm", "sepal_width_cm", "petal_length_cm", "petal_width_cm", "species"]

new_df = df
for old, new in zip(df.columns, new_name):
    new_df = new_df.withColumnRenamed(old, new)

나는 df의 이름을 바꾸기 위해 받아쓰기를 좋아한다.

rename = {'old1': 'new1', 'old2': 'new2'}
for col in df.schema.names:
    df = df.withColumnRenamed(col, rename[col])

'alias'를 사용하여 열 이름을 변경하는 방법:

col('my_column').alias('new_name')

'alias'를 사용할 수 있는 또 다른 방법(아마도 언급되지 않았을 것이다):

df.my_column.alias('new_name')

단일 열 이름 변경의 경우에도 toDF()를 사용할 수 있다.예를 들어,

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

다음과 같은 여러 가지 방법을 사용할 수 있다.

1. df1=df.withColumn("new_column","old_column").drop(col("old_column"))

2. df1=df.withColumn("new_column","old_column")

3. df1=df.select("old_column".alias("new_column"))

from pyspark.sql.types import StructType,StructField, StringType, IntegerType

CreatingDataFrame = [("James","Sales","NY",90000,34,10000),
    ("Michael","Sales","NY",86000,56,20000),
    ("Robert","Sales","CA",81000,30,23000),
    ("Maria","Finance","CA",90000,24,23000),
    ("Raman","Finance","CA",99000,40,24000),
    ("Scott","Finance","NY",83000,36,19000),
    ("Jen","Finance","NY",79000,53,15000),
    ("Jeff","Marketing","CA",80000,25,18000),
    ("Kumar","Marketing","NY",91000,50,21000)
  ]

schema = StructType([ \
    StructField("employee_name",StringType(),True), \
    StructField("department",StringType(),True), \
    StructField("state",StringType(),True), \
    StructField("salary", IntegerType(), True), \
    StructField("age", StringType(), True), \
    StructField("bonus", IntegerType(), True) \
  ])

 
OurData = spark.createDataFrame(data=CreatingDataFrame,schema=schema)

OurData.show()

# COMMAND ----------

GrouppedBonusData=OurData.groupBy("department").sum("bonus")


# COMMAND ----------

GrouppedBonusData.show()


# COMMAND ----------

GrouppedBonusData.printSchema()

# COMMAND ----------

from pyspark.sql.functions import col

BonusColumnRenamed = GrouppedBonusData.select(col("department").alias("department"), col("sum(bonus)").alias("Total_Bonus"))
BonusColumnRenamed.show()

# COMMAND ----------

GrouppedBonusData.groupBy("department").count().show()

# COMMAND ----------

GrouppedSalaryData=OurData.groupBy("department").sum("salary")

# COMMAND ----------

GrouppedSalaryData.show()

# COMMAND ----------

from pyspark.sql.functions import col

SalaryColumnRenamed = GrouppedSalaryData.select(col("department").alias("Department"), col("sum(salary)").alias("Total_Salary"))
SalaryColumnRenamed.show()

다음 방법을 시도해 보십시오.다음 방법을 사용하여 여러 파일의 열 이름을 변경할 수 있음

참조: https://www.linkedin.com/pulse/pyspark-methods-rename-columns-kyle-gibson/

df_initial = spark.read.load('com.databricks.spark.csv')
    
    rename_dict = {
      'Alberto':'Name',
      'Dakota':'askdaosdka'
    }
    
    df_renamed = df_initial \
    .select([col(c).alias(rename_dict.get(c, c)) for c in df_initial.columns])

    
     rename_dict = {
       'FName':'FirstName',
       'LName':'LastName',
       'DOB':'BirthDate'
        }

     return df.select([col(c).alias(rename_dict.get(c, c)) for c in df.columns])


df_renamed = spark.read.load('/mnt/datalake/bronze/testData') \
.transform(renameColumns)

참조URL: https://stackoverflow.com/questions/34077353/how-to-change-dataframe-column-names-in-pyspark