Flink 流处理计算每个 repo 的唯一问题

【问题标题】：Flink stream processing count unique issues per repoFlink 流处理计算每个 repo 的唯一问题
【发布时间】：2018-01-16 16:29:44
【问题描述】：

我正在使用 Scala 中的 Flink，我正在尝试计算每个 repo 的唯一问题数。我有一个带有这样的元组的数据流：（repo_name，issue_id，event_time）。如何获取每个 repo_name 的唯一 issue_id 的计数？我想我必须使用mapWithState，但我不知道如何使用它。

提前致谢。

【问题讨论】：

您是否想要在每次有新的独特问题时更新的源源不断的流，或者这是窗口化的（例如每日或每周计数）？

标签： git scala streaming apache-flink flink-streaming

【解决方案1】：

假设您希望在 7 天的滚动时间窗口内处理事件。

// eventStream: stream of case classes of type GithubEvent 
eventStream
   // only look at IssuesEvent
  .filter(e => e.`type` == "IssuesEvent")
   // key by the name of the repository
  .keyBy("repo.name")
   // tumbling time window of a week
  .timeWindow(Time.days(7))
   // apply window function
  .apply { (key, _, vals, out: Collector[(String)]) =>
    var count = 0;
    for (_ <- vals) {
      count = count + 1;
    }
    out.collect(s"Repo name: $key Unique issues: $count")
  }

要计算每个存储库的唯一问题数量，我们需要查看 IssuesEvents。我们以存储库的名称作为关键字。然后，我们应用一个窗口函数返回一个字符串，表示问题的唯一数量。

参考文献：

【讨论】：