scala - 不支持没有相等谓词的流流连接

Question

我正在使用 Spark 2.3 并尝试连接两个数据流。我的左流和右流都有一个数组。只有当右流数组是左流数组的子集时，我才想加入这两个流。

例如，我的 streaA 看起来像这样:

StreamA:
|---|------|---------------------|-----------|
|id | dept | employeesInMeetings | DateTime  |
|---|------|---------------------|-----------|
| 1 | sales| [John]              | 7/2 14:00 |
| 2 | mktg | [Adam, Mike]        | 7/2 12:30 |
| 3 | hr   | [Rick, Jill, Andy]  | 7/2 14:00 |
|---|------|---------------------|-----------|

我的 streamB 如下所示:

StreamB:
|--------------|--------------|----------|
|employees     | confRooms    | DateTime |
|--------------|--------------|----------|
| [John, Jane] |      A       | 7/2 14:00|
| [Adam, Mike] |      C       | 7/2 12:30| 
| [Jill, Andy] |      B       | 7/2 14:00|
|--------------|--------------|----------|

我只关心参加同一次 session 的同一部门的员工。因此，作为交集的结果，我的结果流需要如下所示:

|---|------|---------------------|-----------|----------|
|id | dept | employeesInMeetings | DateTime  | confRoom |
|---|------|---------------------|-----------|----------|
| 2 | mktg | [Adam, Mike]        | 7/2 12:30 |    C     |
| 3 | hr   | [Rick, Jill, Andy]  | 7/2 14:00 |    B     |
|---|------|---------------------|-----------|----------|

我创建了一个 UDF 来进行相交:

val arrayIntersect = udf((leftArr: Array[String], rightArr: Array[String]) => {
  import spark.implicits._
  if(leftArr.intersect(rightArr.toSeq).length == rightArr.size){
    true
  } else {
    false
  }
})

并尝试按如下方式使用它:

streamA.joinWith(streamB, expr("arrayIntersect(leftArr, rightArr) AND streamA.DateTime BETWEEN streamB.DateTime and streamB.DateTime + INTERVAL 12 hours"))

但是，我得到了错误:

org.apache.spark.sql.AnalysisException: Stream stream joins without equality predicate is not supported;

有人知道这里是否有解决方法吗？任何帮助将不胜感激!谢谢!

Answer 1

不幸的是，在流-流连接中没有解决这个问题的方法:(

我们确实需要一个相等谓词，因为我们使用它来执行使用流式对称散列连接算法的连接——两个流都使用公共(public) key 进行分区，以便来自两个流的相关记录最终位于同一分区中。