Spark Dataframe 中的重复列

Question

我在 hadoop 集群中有一个 10GB 的 csv 文件，其中包含重复的列。我尝试在 SparkR 中分析它，所以我使用 spark-csv 包将其解析为 DataFrame:

  df <- read.df(
    sqlContext,
    FILE_PATH,
    source = "com.databricks.spark.csv",
    header = "true",
    mode = "DROPMALFORMED"
  )

但是由于 df 有重复的 Email 列，如果我想 select 这一列，它会出错：

select(df, 'Email')

15/11/19 15:41:58 ERROR RBackendHandler: select on 1422 failed
Error in invokeJava(isStatic = FALSE, objId$id, methodName, ...) : 
  org.apache.spark.sql.AnalysisException: Reference 'Email' is ambiguous, could be: Email#350, Email#361.;
    at org.apache.spark.sql.catalyst.plans.logical.LogicalPlan.resolve(LogicalPlan.scala:278)
...

我想保留第一次出现的 Email 列并删除后者，我该怎么做？

Answer 1

尝试重命名列。

您可以 select 按位置而不是 select 调用。

colnames(df)[column number of interest] <- 'deleteme'

或者您可以直接删除该列

 newdf <- df[,-x]

其中 x 是您不需要的列号。

更新：

如果上述方法不起作用，您可以将 header 设置为 false，然后使用第一行重命名列：

  df <- read.df(
    sqlContext,
    FILE_PATH,
    source = "com.databricks.spark.csv",
    header = "FALSE",
    mode = "DROPMALFORMED"
  )

#get first row to use as column names
mycolnames <- df[1,]

#edit the dup column *in situ*
mycolnames[x] <- 'IamNotADup'
colnames(df) <- df[1,]

# drop the first row:
df <- df[-1,]

Answer 2

最好的方法是更改上游的列名 ;)

然而，这似乎是不可能的，所以有几个选择：

如果列的大小写不同（"email" vs "Email"）您可以打开区分大小写：
```
     sql(sqlContext, "set spark.sql.caseSensitive=true")
```

如果列名完全相同，您将需要手动指定模式并跳过第一行以避免 headers:

customSchema <- structType(
structField("year", "integer"), 
structField("make", "string"),
structField("model", "string"),
structField("comment", "string"),
structField("blank", "string"))

df <- read.df(sqlContext, "cars.csv", source = "com.databricks.spark.csv", header="true", schema = customSchema)

Answer 3

您还可以使用 toDF 创建新数据框。

对于 pyspark，这是同样的事情：Selecting or removing duplicate columns from spark dataframe

Answer 4

你可以在启动spark session时添加一行，成功创建spark session后添加这一行来设置spark config...

spark.conf.set("spark.sql.caseSensitive", "true")

Spark Dataframe 中的重复列

Duplicate columns in Spark Dataframe

csv

hadoop

r

apache-spark

sparkr