代码之家 › 专栏 › 技术社区 › S. N

scala流处理,计算每个日期的发生次数

stream-processing apache-flink scala

0

S. N · 技术社区 · 6 年前

我正在使用scala流处理,我有一个事件列表,我试图以这种格式计算每个日期的发生次数( dd-mm-yyyy )

这是我对案例类的定义:

case class Event(prId: Int, author: String, event: String, timestamp: Date)

我的解决方案如下

def CountOccur(events: DataStreaming[Event]): DataStreaming[(String, Int)] = {
    events.map(c => (c.timestamp, 1)).keyBy(x => x._1).timeWindow(Time.seconds(5)).sum(1)
  }

这将返回如下答案:

2010-09-25,10
2010-09-27,7
.
.
.

我现在的问题是 timestamp 是的 date format 但是我需要 string .

我怎样才能把它转换成字符串格式的日期呢?

固定的:

以下是解决此问题的方法:

events.map(c => (sdf.format(c.timestamp), 1)).keyBy(x => sdf.format(x._1)).window(EventTimeSessionWindows.withGap(Time.seconds(10))

但问题是现在它什么都不返回!!

1 回复 | 直到 6 年前

1

-1

Sagi 6 年前

我不认识弗林克,但我猜

def CountOccur(events: DataStreaming[Event]): DataStreaming[(String, Int)] = {
    val sdf = new SimpleDateFormat("dd-mm-yyyy")

    events.map(c => (sdf.format(c.timestamp), 1)).keyBy(x => x._1).timeWindow(Time.seconds(5)).sum(1)
}

您需要删除第二种格式 sdf.format(x._1) ,它将引发一个例外。