我在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(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
列并删除后者,我该怎么做?
最好的方法是更改上游的列名;)
但是,似乎这是不可能的,所以有几个选择:
sql(sqlContext, "set spark.sql.caseSensitive=true")
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)
尝试重命名该列。
您可以按位置而不是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,]
您还可以使用toDF
创建新的数据框。
对于pyspark来说,这是同样的事情:Selecting or removing duplicate columns from spark dataframe