Flink 1.5 DataStream：重复过滤答案

【问题标题】：Flink 1.5 DataStream: Duplicate FilteringFlink 1.5 DataStream：重复过滤
【发布时间】：2018-12-09 23:44:07
【问题描述】：

我有一个未绑定的 DataStream，它代表社交网络中的友谊。这些友谊可以是双向的，因此在流中出现两次。

数据的结构是：timestamp|user1|user2。例如：

2010-03-09T02:51:11.571+0000|143|1219
2010-03-09T06:08:51.942+0000|1242|4624
2010-03-09T08:24:03.773+0000|2191|4986
2010-03-09T09:37:09.788+0000|459|4644

我想删除双向友谊只计算一次。在实践中，我想过滤重复项。我找到了解决方案here

我的 FilterFunction 看起来像：

   def filter(ds: DataStream[String]): DataStream[(String, String, String)] = {

    val res = data.
      mapWith(line => {
        val str = line.split("\\|")
        if (str(1).toLong > str(2).toLong)
          (str(0), str(1), str(2))
        else
           (str(0), str(2), str(1))
      })
       .keyBy(tuple => (tuple._2, tuple._3))
      .flatMap(new FilterFunction())

    res
  }

我已经将我的 RichFlatMapFunction 实现为：

class FilterFunction extends RichFlatMapFunction[(String, String, String), (String, String, String)] {

  private var seen: ValueState[Boolean] = _

  override def flatMap(value: (String, String, String), out: 
Collector[(String, String, String)]): Unit = {

     if (!seen.value() || seen.value() == null) {
       seen.update(true)
       out.collect(value)
     }
   }

  override def open(parameters: Configuration): Unit = {
     seen = getRuntimeContext.getState(
       new ValueStateDescriptor("seen", classOf[Boolean])
     )
   }
}

但是，当我打印时，我得到了随机结果。我试图在 1 年的时间窗口内进行计数：

val da1 = filter(data)
  .mapWith(tuple => Parser.parseUserConnection(tuple).get)
  .assignAscendingTimestamps(connection => connection.timestamp.getMillis)
  .mapWith(connection => (connection, 1))
  .timeWindowAll(Time.days(365))
  .sum(1)
  .mapWith(tuple => tuple._2)
  .print()

我的控制台第一次打印：

1> 33735

然后：

1> 10658
2> 33735

对于随后的执行，不同的结果（只是 33735 似乎是稳定的）。我无法理解这种奇怪的行为。

【问题讨论】：

标签： scala apache-flink

【解决方案1】：

很难理解您所发现的令人惊讶的内容。但调试此类应用程序的一般技术是打印管道不同阶段的结果，以查看结果在什么时候变得奇怪。或者在 IDE 中调试作业并单步执行。

【讨论】：

是的，问题几乎肯定出在 RichFlatMapFunction 中