我正在使用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条答案
按热度按时间s8vozzvw1#
我不认识Flink,但我猜
您需要删除第二种格式
sdf.format(x._1)
,它将引发异常。