在 spark python 中重命名数据框列

Posted 2023-04-15

【中文标题】在 spark python 中重命名数据框列

【英文标题】：Rename dataframe columns in spark python

【发布时间】：2020-06-24 12:15:23

【问题描述】：

我有一个带有标题的 CSV，我想将其保存为 Parquet（实际上是一个增量表）

列标题中有空格，这是 parquet 无法处理的。如何将空格改为下划线？

这是我到目前为止的内容，从其他 SO 帖子拼凑而成：

from pyspark.sql.functions import *

df = spark.read.option("header", True).option("delimiter","\u0001").option("inferSchema",True).csv("/mnt/landing/MyFile.TXT")

names = df.schema.names
for name in names:
    df2 = df.withColumnRenamed(name,regexp_replace(name, ' ', '_'))

当我运行它时，最后一行给了我这个错误：

TypeError: 列不可迭代

我认为这将是一个常见的要求，因为镶木地板无法处理空间，但很难找到任何示例。

【问题讨论】：

你可以试试 select:df.select([col(a).alias(b) for a,b in zip(df.columns,[re.sub(" ","_",i) for i in df.columns])])

【参考方案1】：

您需要使用 reduce 函数迭代地对数据框应用重命名，因为在您的代码中 df2 只会重命名最后一列...

代码如下所示（而不是 for 循环）：

df2 = reduce(lambda data, name: data.withColumnRenamed(name, name.replace('1', '2')), 
             names, df)

【讨论】：

太棒了。我只需要在顶部添加from functools import reduce。到目前为止，这正在做我想做的事——我也会看看其他一些人

在这种情况下，reduce 函数接受三个参数：lambda 函数、名称和 df？

似乎names 被传递给name 参数并且df 被传递给data 参数？试图了解这里发生了什么

我看到的所有reduce例子都带两个参数？

好的，这里有一些很好的例子展示了两种方法（减少和循环）medium.com/@mrpowers/…

【参考方案2】：

使用 .toDF（或）.select 并传递列列表来创建新的数据框。

df.show()
#+---+----+----+
#| id|id a|id b|
#+---+----+----+
#|  1|   a|   b|
#|  2|   c|   d|
#+---+----+----+
new_cols=list(map(lambda x: x.replace(" ", "_"), df.columns))

df.toDF(*new_cols).show()

df.select([col(s).alias(s.replace(' ','_')) for s in df.columns]).show()
#+---+----+----+
#| id|id_a|id_b|
#+---+----+----+
#|  1|   a|   b|
#|  2|   c|   d|
#+---+----+----+

【讨论】：

感谢您的意见。我还没有尝试过你的答案，但我相信我会回来的。

【参考方案3】：

您遇到异常是因为 - 函数 regexp_replace 返回 Column 类型但函数 withColumnRenamed 是 String 类型的异常。

def regexp_replace(e: org.apache.spark.sql.Column,pattern: String,replacement: String): org.apache.spark.sql.Column

def withColumnRenamed(existingName: String,newName: String): org.apache.spark.sql.DataFrame