会话可以使用
EventTimeSessionWindows
从事件流中提取。它将所有一起出现的事件组合成一个
会话窗口,直到它们之间的间隔大于指定的值。如果流包含许多会话(可以通过每个事件中的
sessionId
进行标识),则应首先按会话ID分组,以便为每个会话单独保留会话窗口。
在下面的代码示例中,形式为的事件
case class Event(
createdat: Timestamp,
session: String
)
被转换为
case class SessionEvent(
sessionId: String,
start: Instant,
end: Instant,
`type`: String
)
SessionEvent
是在最后20分钟内没有事件(sessionTimeout
)时发出的。
import java.util.Properties
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer09, FlinkKafkaProducer09}
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import org.apache.flink.util.Collector
object Main extends App {
val sessionTimeout = Time minutes 20
val kafkaCluster = "localhost:9092"
val inputTopic = "events"
val outputTopic = "sessions"
val env = StreamExecutionEnvironment.getExecutionEnvironment
val properties = new Properties
properties.setProperty("bootstrap.servers", kafkaCluster)
properties.setProperty("group.id", "sessions")
val consumer = new FlinkKafkaConsumer09[String](inputTopic, new SimpleStringSchema, properties)
val producer = new FlinkKafkaProducer09[String](kafkaCluster, outputTopic, new SimpleStringSchema)
val stream =
env
.addSource(consumer)
.map(Formats.readEvent _)
.keyBy(_.session)
.window(ProcessingTimeSessionWindows withGap sessionTimeout)
.apply[SessionEvent] {
(key: String, window: TimeWindow, values: Iterable[Event], out: Collector[SessionEvent]) ⇒
val session =
SessionEvent(
key,
values.head.createdat.toInstant,
values.last.createdat.toInstant,
"end"
)
out.collect(session)
}
.map(Formats.writeSessionEvent _)
.addSink(producer)
env.execute("sessions")
}