为啥在 pyspark 中加入两个临时视图后删除列不起作用，但它适用于数据框连接？

Posted 2023-04-17

【中文标题】为啥在 pyspark 中加入两个临时视图后删除列不起作用，但它适用于数据框连接？

【英文标题】：Why dropping columns after joining two temp views in pyspark isn't working, which however works with data frame joins?为什么在 pyspark 中加入两个临时视图后删除列不起作用，但它适用于数据框连接？

【发布时间】：2020-08-01 15:12:43

【问题描述】：

我需要从数据框创建临时视图，然后我需要加入这些临时视图并删除重复的列。所以我写了如下代码：

 Data1 = [  ("xx1",34,60),
           ("xx2",33,80),
           ("xx3",37,50) ]

dataSchema1 = StructType([
    StructField("Name",StringType(),True),
    StructField("Age",IntegerType(),True),
    StructField("Id",IntegerType(),True)
   ])

Data2 = [  (60,"M",3000.60),
           (80,"F",3300.80),
           (50,"M",5000.50) ]

dataSchema2 = StructType([
    StructField("Id",IntegerType(),True),
    StructField("Gender", StringType(), True),
    StructField("Salary", DoubleType(), True)
  ])

df1 = spark.createDataFrame(spark.sparkContext.parallelize(Data1),schema=dataSchema1)
df1.createOrReplaceTempView('view1')

df2 = spark.createDataFrame(spark.sparkContext.parallelize(Data2),schema=dataSchema2)
df2.createOrReplaceTempView('view2')

jDF=spark.sql("select * from view1 join view2 on view1.Id = view2.Id")
jDF.columns                 // ['Name', 'Age', 'Id', 'Id', 'Gender', 'Salary']
rjDF=jDF.drop('view2.ID')    //this function is not working
rjDF.columns                // ['Name', 'Age', 'Id', 'Id', 'Gender', 'Salary']

在上面的代码中，drop column 方法没有按预期工作，也没有抛出任何错误。 但是，如果我尝试使用数据框删除列（在我的用例中这显然不是我的首选），那么 drop 方法可以正常工作。

joinDF=df1.join(df2, df1.Id == df2.Id)
dropped=joinDF.drop(df2.Id)    // working absolutely fine
dropped.columns               // ['Name', 'Age', 'Id', 'Gender', 'Salary']

谁能帮助我理解第一种方法从连接的临时视图中删除一列有什么问题？

【参考方案1】：

来自数据集的drop 将column names 或column 本身丢弃。如果架构不包含 column name(s)，则这是 no-op 操作。

drop 内部使用analyzer.resolver 检查提供的字符串是否存在于数据框中。 请注意，您不能在 drop 中提供引用列的类似 sql 的语法如果您提供相同的 spark 将假定整个字符串作为列名。

selectExpr("..") 和 functions.expr("..") 在内部使用 sessionState.sqlParser 来解析像 <table/view name>.<column_name> 这样的引用列的类似 sql 的语法。

如果您想使用类似 sql 的语法，请尝试将其与 expr 函数一起使用-

rjDF=jDF.drop(F.expr('view2.Id'))

否则你可以使用基于drop的工作列-

dropped=joinDF.drop(df2.Id)

【讨论】：

>rjDF=jDF.drop(F.expr('view2.Id')) 这解决了我的问题。非常感谢！