案例需求:
- 假设用户需要每个1秒钟需要统计4秒钟 窗口中数据的量,然后对统计的结果值进行checkpoint处理
复制代码 数据规划
- 使用自定义算子每秒钟产生大约10000条数据
- 产生的数据为一个四元组(Long,String,String,Interger)-- (id,name,info,count)
- 数据经统计后,统计结果打印到终端输出
- 打印输出的结果为Long类型的数据
复制代码 开发自定义数据源:
代码实现:
[code]// ** 开发自定义数据源// 1、自定义样例类case class Msg(id:Long, name:String,info:String,cout:Int)// 2、自定义数据源,继承RichSourceFunctionclass MySourceFunction extends RichSourceFunction[Msg]{ var isRunning = true // 3、实现run方法,每秒向流中注入10000个样例类 override def run(ctx: SourceFunction.SourceContext[Msg]): Unit = { while (isRunning){ for(i |