Flink DataStream API deduplication in Scala

dedup.scala

val dedupColumn = "..." // column name to use as a key for deduplication
val ttl = Some(Time.minutes(60)) // state TTL

stream
  .keyBy(row => row.getField(dedupColumn))
  .flatMap(new RichFlatMapFunction[Row, Row] {
    @transient
    private var seen: ValueState[Boolean] = _

    override def open(parameters: Configuration): Unit = {
      val stateDescriptor =
        new ValueStateDescriptor[Boolean](
          "seen",
          createTypeInformation[Boolean]
        )

      ttl foreach { ttlValue =>
        val ttlConfig = StateTtlConfig
          .newBuilder(ttlValue)
          .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite)
          .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
          .build

        stateDescriptor.enableTimeToLive(ttlConfig)
      }

      seen = getRuntimeContext.getState(stateDescriptor)
    }

    override def flatMap(row: Row, out: Collector[Row]): Unit = {
      if (seen.value() == null || !seen.value()) {
        seen.update(true)
        out.collect(row)
      }
    }
  })
  .name("Dedup")
  .uid("dedup")