Spark [Scala]：不明确的 groupBy 列名

Question

所以在测试时我收到以下错误消息：

org.apache.spark.sql.AnalysisException: Reference 'from' is ambiguous, could be: from, from.;

在测试时但不是在我运行spark-shell...中的部分时？

我正在对以下数据框进行交叉连接：

scala> timeSpanDF.show
+----------+----------+
|      from|        to|
+----------+----------+
|2018-01-01|2018-02-01|
|2018-01-01|2018-03-01|
|2018-02-01|2018-03-01|
+----------+----------+


scala> df.show
+------------+----------+--------+-----+--------------------+
|pressroom_id|     month|category|event|               email|
+------------+----------+--------+-----+--------------------+
|           1|2017-01-01| contact| open|somebody@example.com|
|           1|2018-01-01| contact| open|     me1@example.com|
|           1|2018-02-01| contact| open|     me1@example.com|
|           1|2018-02-01| contact| open|     me1@example.com|
|           1|2018-01-01| contact| open|     you@example.com|
|           1|2018-03-01| contact| open|     etc@example.com|
|           1|2018-02-01| contact| open|     me2@example.com|
|           1|2018-02-01| contact| open|     me2@example.com|
|           2|2018-01-01| contact| open|     me1@example.com|
+------------+----------+--------+-----+--------------------+

所以我这样做

val joinedDF = timeSpansDF
    .crossJoin(df)
    .filter(
        df("month") >= timeSpansDF("from") 
        && df("month") < timeSpansDF("to")
    )
    .distinct

得到这个

scala> joinedDF.show
+----------+----------+------------+----------+--------+-----+---------------+
|      from|        to|pressroom_id|     month|category|event|          email|
+----------+----------+------------+----------+--------+-----+---------------+
|2018-01-01|2018-03-01|           2|2018-01-01| contact| open|me1@example.com|
|2018-02-01|2018-03-01|           1|2018-02-01| contact| open|me1@example.com|
|2018-02-01|2018-03-01|           1|2018-02-01| contact| open|me2@example.com|
|2018-01-01|2018-03-01|           1|2018-01-01| contact| open|me1@example.com|
|2018-01-01|2018-02-01|           1|2018-01-01| contact| open|me1@example.com|
|2018-01-01|2018-03-01|           1|2018-02-01| contact| open|me2@example.com|
|2018-01-01|2018-02-01|           2|2018-01-01| contact| open|me1@example.com|
|2018-01-01|2018-03-01|           1|2018-01-01| contact| open|you@example.com|
|2018-01-01|2018-03-01|           1|2018-02-01| contact| open|me1@example.com|
|2018-01-01|2018-02-01|           1|2018-01-01| contact| open|you@example.com|
+----------+----------+------------+----------+--------+-----+---------------+

然后我想像这样聚合这个表，这就是我得到奇怪消息的地方：

joinedDF.where(col("category") === lit(category) && col("event") === lit("open"))
    .groupBy("pressroom_id", "from", "to")
    .agg(count("email").cast("integer").as("something"))

指向 groupBy。奇怪的是，这在 shell 中有效，但是当这些操作被放入函数并使用 scalaTest 测试时，它们会出现错误？

医生怎么样？

Answer 1

由于我没有生成joinedDF的代码，因此我自己准备了Dataframe来生成joinedDF。我已经在ScalaTest中对其进行了测试，它工作正常。

请更新您的代码如下。

val df = Seq(("2018-01-01", "2018-03-01", 2,"contact","open","me1@example.com"),
                ("2018-02-01","2018-03-01",1, "contact","open","me1@example.com"),
                ("2018-01-01","2018-03-01",1, "contact","open","you@example.com"),
                ("2018-02-01","2018-03-01",1, "contact","open","me1@example.com"),
                ("2018-01-01","2018-02-01",1, "contact","open","me1@example.com"),
                ("2018-01-01","2018-02-01", 1, "contact","open","you@example.com")).
      toDF("from", "to", "pressroom_id","category","event","email")

df.show() 

+----------+----------+------------+--------+-----+---------------+
|      from|        to|pressroom_id|category|event|          email|
+----------+----------+------------+--------+-----+---------------+
|2018-01-01|2018-03-01|           2| contact| open|me1@example.com|
|2018-02-01|2018-03-01|           1| contact| open|me1@example.com|
|2018-01-01|2018-03-01|           1| contact| open|you@example.com|
|2018-02-01|2018-03-01|           1| contact| open|me1@example.com|
|2018-01-01|2018-02-01|           1| contact| open|me1@example.com|
|2018-01-01|2018-02-01|           1| contact| open|you@example.com|
+----------+----------+------------+--------+-----+---------------+

val df1 = df.where(col("category") === lit("contact") && col("event") === lit("open"))
      .groupBy("pressroom_id", "from", "to")
      .agg(count("email").cast("integer").as("something"))

df1.show()



   +------------+----------+----------+---------+
|pressroom_id|      from|        to|something|
+------------+----------+----------+---------+
|           2|2018-01-01|2018-03-01|        1|
|           1|2018-01-01|2018-03-01|        1|
|           1|2018-02-01|2018-03-01|        2|
|           1|2018-01-01|2018-02-01|        2|
+------------+----------+----------+---------+

我在我的代码中添加了 import 语句。

import org.apache.spark.sql.functions._

希望对您有所帮助！

Answer 2

我不是 Scala 专家，但我是一名数据库管理员。

我怀疑您的问题源于使用 SQL 保留字 from 作为列名，因为堆栈跟踪显示异常来自 Spark SQL 模块：org.apache.spark.sql.AnalysisException。

要么：

1. 尝试将列名更改为非保留字；或

2. 将列名完全限定为joinedDF.from。

NB 您的第二个代码 sn-p 指的是一个名为 timeSpanDF 的数据帧，而您的第三个代码指的是 timeSpansDF（复数）。

编辑：作为社区的新成员，我没有足够的声誉对@KZapagol 的答案发表评论，但我相信他的回答的本质是原始发布者的@987654327 中有错字@ 子句：col("category") === lit(category) => col("category") === lit("contact")。