isin在spark中的With Column函数中抛出stackoverflow错误

Question

我在scala应用程序中使用Spark2.3。我有一个从spark sql创建的dataframe，在我共享的示例代码中名为sqlDF。我有一个字符串列表，其中包含以下项

column1 | column2 | column3
1       |1        |1       
2       |-5       |1       
6       |-6       |1       
-7      |-8       |-7       

column1 | column2 | column3
1       |1        |1       
2       |-5       |1       
6       |0        |1       
0       |0        |0

为此，我在下面对SQLDF中的所有列（超过500列）进行查询。

sqlDF = sqlDF.withColumn(currColumnName, when(col(currColumnName).isin(stringList:_*), 0).otherwise(col(currColumnName)))

但是得到下面的错误，顺便说一下，如果我只选择一列进行迭代，它可以工作，但是如果我运行上面的代码进行500列迭代，它就失败了

线程“streaming-job-executor-0”java.lang.stackOverflowError在scala.collection.generic.gentraversableFactory$GenericCanBuildFrom.apply(gentraversableFactory.scala:57)在scala.collection.generic.gentraversableFactory$GenericCanBuildFrom.apply(gentraversableFactory.scala:52)在scala.collection.traversableFactory$class.builder$1(traversableLike.scala:229)在scala.collection.traversableLike$class.builder在org.apache.spark.sql.catalyst.trees.treenode$$Anonfun$4.在org.apache.spark.sql.catalyst.trees.treenode.MapProductIterator(Treenode.scala:187)申请(Treenode.scala:333)

我缺少的是什么？

Answer 1

这里有一种不同的方法，在columnx和X之间应用left antijoin，其中X是传输到DataFrame中的项列表。左反联接将返回x中不存在的所有项，结果我们通过一个外部联接将它们连接在一起（为了更好的性能，可以用左联接代替外部联接，但这将根据用monotonically_recreasing_id分配的id排除所有为零的记录，即id==3):

import org.apache.spark.sql.functions.{monotonically_increasing_id, col}

val df = Seq(
(1, 1, 1),       
(2, -5, 1),       
(6, -6, 1),       
(-7, -8, -7))
.toDF("c1", "c2", "c3")
.withColumn("id", monotonically_increasing_id())

val exdf = Seq(-9, -8, -7, -6).toDF("x")

df.columns.map{ c =>
   df.select("id", c).join(exdf, col(c) === $"x", "left_anti")
}
.reduce((df1, df2) => df1.join(df2, Seq("id"), "outer"))
.na.fill(0)
.show

产出：

+---+---+---+---+
| id| c1| c2| c3|
+---+---+---+---+
|  0|  1|  1|  1|
|  1|  2| -5|  1|
|  3|  0|  0|  0|
|  2|  6|  0|  1|
+---+---+---+---+