请教一下老师，您说的【同样数据的话，水印没有推进，窗口就不会触发】是不是意思是发送相同的数据，数据本身携带的时间戳是一样的，达不到水位线触发窗口的标准呀？ 还有两个问题想请教一下各位老师： 1、事件时间窗口的闭合是取决于下一条数据所携带的时间戳嘛，只有当下一条数据携带的时间戳大于上一个窗口的endTime，窗口才会触发，如果是这个样子的话，那如果一个最后一个窗口怎么触发啊 2、我想使用stream api去打印出来窗口的起始时间以及结束时间，这个是哪一个api呀
| | 小昌同学 | | ccc0606fight...@163.com | ---- 回复的原邮件 ---- | 发件人 | lxk<lxk7...@163.com> | | 发送日期 | 2023年5月25日 10:14 | | 收件人 | <user-zh@flink.apache.org> | | 主题 | Re:回复： flink 窗口触发计算的条件 | 你好，可以先看看官方文档中关于事件时间和水印的介绍 https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/concepts/time/ 如果你发了多条数据，但是都是同样数据的话，水印没有推进，窗口就不会触发 在 2023-05-25 10:00:36，"小昌同学" <ccc0606fight...@163.com> 写道： 是的 我发送了很多数据，发现窗口还是没有触发 | | 小昌同学 | | ccc0606fight...@163.com | ---- 回复的原邮件 ---- | 发件人 | yidan zhao<hinobl...@gmail.com> | | 发送日期 | 2023年5月25日 09:59 | | 收件人 | <user-zh@flink.apache.org> | | 主题 | Re: flink 窗口触发计算的条件 | 如果你只发送了一条数据，那么watermark不会推进，就不会触发窗口计算。你需要更多数据。 小昌同学 <ccc0606fight...@163.com> 于2023年5月25日周四 09:32写道： 各位老师，请教一下关于flink 事件时间窗口的执行时间点的相关问题； 我使用的窗口是：TumblingEventTimeWindows(Time.minutes(1L)),我使用的时间定义是System.currentTimeMillis()，watermark是2秒， 但是当我发送一条数据后，过了5分钟之后，窗口都没有触发计算，想请各位老师帮忙看一下程序的问题所在： 相关代码以及样例数据如下： | package job; import bean.MidInfo3; import bean.Result; import bean2.BaseInfo2; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONObject; import config.FlinkConfig; import org.apache.flink.api.common.eventtime.WatermarkStrategy; import org.apache.flink.api.common.functions.FilterFunction; import org.apache.flink.api.common.functions.JoinFunction; import org.apache.flink.api.common.functions.MapFunction; import org.apache.flink.api.common.serialization.SimpleStringSchema; import org.apache.flink.api.common.state.StateTtlConfig; import org.apache.flink.api.common.state.ValueState; import org.apache.flink.api.common.state.ValueStateDescriptor; import org.apache.flink.api.java.functions.KeySelector; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.datastream.ConnectedStreams; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.streaming.api.functions.ProcessFunction; import org.apache.flink.streaming.api.functions.co.CoMapFunction; import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction; import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows; import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows; import org.apache.flink.streaming.api.windowing.time.Time; import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger; import org.apache.flink.streaming.api.windowing.windows.TimeWindow; import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer; import org.apache.flink.util.Collector; import utils.DateUtil; import utils.JdbcUtil; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.time.Duration; import java.util.HashMap; import java.util.Properties; public class RytLogAnly9 { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); env.disableOperatorChaining(); //1、消费Kafka中的数据 String servers = FlinkConfig.config.getProperty("dev_bootstrap.servers"); String topicName = FlinkConfig.config.getProperty("dev_topicName"); String groupId = FlinkConfig.config.getProperty("dev_groupId"); String devMode = FlinkConfig.config.getProperty("dev_mode"); Properties prop = new Properties(); prop.setProperty("bootstrap.servers", servers); prop.setProperty("group.id", groupId); prop.setProperty("auto.offset.reset", devMode); DataStreamSource<String> sourceStream = env.addSource(new FlinkKafkaConsumer<String>(topicName, new SimpleStringSchema(), prop)); sourceStream.print("最源端的数据sourceStream"); //2、对原始数据进行处理，获取到自己需要的数据，生成BaseInfo2基类数据 SingleOutputStreamOperator<BaseInfo2> baseInfoStream = sourceStream.map(new MapFunction<String, BaseInfo2>() { @Override public BaseInfo2 map(String value) throws Exception { JSONObject jsonObject = JSON.parseObject(value); //获取到不同的服务器IP String serverIp = jsonObject.getString("ip"); //获取到不同的data的数据 String datas = jsonObject.getString("data"); String[] splits = datas.split("\n"); HashMap<String, String> dataMap = new HashMap<>(); //将time填充到自定义类型中，用来判断同一个num的请求以及应答时间 String time = splits[0].substring(7, 19).replace("-", "").trim(); //将subData填充到自定义类型中，用来判断时请求还是应答 String subData = datas.substring(0, 10); for (int i = 0; i < splits.length; i++) { if (splits[i].contains("=")) { splits[i] = splits[i].replaceFirst("=", "&"); String[] temp = splits[i].split("&"); if (temp.length > 1) { dataMap.put(temp[0].toLowerCase(), temp[1]); } } } return new BaseInfo2(dataMap.get("action"), "需要去MySQL中查找对应的功能描述", serverIp, DateUtil.string2Long(time), dataMap.get("handleserialno"), subData, System.currentTimeMillis()); } }).assignTimestampsAndWatermarks(WatermarkStrategy.<BaseInfo2>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner((element, recordTimestamp) -> element.getEvenTime())); baseInfoStream.print("不加功能描述的 baseInfoStream"); //3、上述的数据流中的action仅仅是数字，需要关联一下MySQL去拿到对应的功能中文描述 SingleOutputStreamOperator<BaseInfo2> completeInfoStream = baseInfoStream.map(new MapFunction<BaseInfo2, BaseInfo2>() { @Override public BaseInfo2 map(BaseInfo2 value) throws Exception { //拿到数据中携带的数字的action String actionId = value.getFuncId(); System.out.println("数据中的action编码是： " + actionId); String actionName = null; Connection connection = null; PreparedStatement ps = null; //根据数据的action去MySQL中查找到对应的中午注释 try { String sql = "select action_name from ActionType where action = ?"; connection = JdbcUtil.getConnection(); ps = connection.prepareStatement(sql); ps.setString(1, actionId); ResultSet resultSet = ps.executeQuery(); System.out.println("resultSet是" + resultSet); if (resultSet.next()) { actionName = resultSet.getString("action_name"); } } catch (Exception e) { throw new RuntimeException(e); } finally { JdbcUtil.closeResource(connection, ps); } return new BaseInfo2(value.getFuncId(), actionName, value.getServerIp(), value.getBaseTime(), value.getHandleSerialNo(), value.getInfo(), value.getEvenTime()); } }).assignTimestampsAndWatermarks(WatermarkStrategy.<BaseInfo2>forBoundedOutOfOrderness(Duration.ofSeconds(2L)).withTimestampAssigner((element, recordTimestamp) -> element.getEvenTime())); completeInfoStream.print("加上中文描述的 completeInfoStream"); SingleOutputStreamOperator<BaseInfo2> requestDataStream = completeInfoStream.filter(new FilterFunction<BaseInfo2>() { @Override public boolean filter(BaseInfo2 baseInfo2) throws Exception { return baseInfo2.getInfo().contains("请求"); } }); SingleOutputStreamOperator<BaseInfo2> answerDataStream = completeInfoStream.filter(new FilterFunction<BaseInfo2>() { @Override public boolean filter(BaseInfo2 baseInfo2) throws Exception { return baseInfo2.getInfo().contains("应答"); } }); requestDataStream.print("请求流是 requestDataStream"); answerDataStream.print("应答流是 answerDataStream"); DataStream<MidInfo3> joinStream = requestDataStream.join(answerDataStream) .where(BaseInfo2::getHandleSerialNo) .equalTo(BaseInfo2::getHandleSerialNo) .window(TumblingEventTimeWindows.of(Time.minutes(1L))) .apply(new JoinFunction<BaseInfo2, BaseInfo2, MidInfo3>() { @Override public MidInfo3 join(BaseInfo2 first, BaseInfo2 second) throws Exception { System.out.println("以关联：" + first.getFuncId() + second.getEvenTime()); System.out.println("关联：" + first.getEvenTime() +"|" +second.getEvenTime()+"执行时间："+System.currentTimeMillis()); return new MidInfo3(first.getFuncId(), first.getFuncIdDesc(), first.getServerIp(), first.getBaseTime(), second.getBaseTime(), first.getFuncId() + first.getServerIp(), first.getEvenTime()); } }); joinStream.print("joinStream:"); System.out.println("joinTime:"+System.currentTimeMillis()); joinStream.keyBy(new KeySelector<MidInfo3, String>() { @Override public String getKey(MidInfo3 value) throws Exception { return value.getFuncId() + value.getServerIp(); } }).process(new ProcessFunction<MidInfo3, Result>() { private ValueState<Long> timeState; @Override public void open(Configuration parameters) throws Exception { System.out.println("加载的是process中的open方法"+System.currentTimeMillis()); ValueStateDescriptor<Long> timeStateDescriptor = new ValueStateDescriptor<>("timeState", Long.class); // 过期状态清除 StateTtlConfig stateTtlConfig = StateTtlConfig .newBuilder(org.apache.flink.api.common.time.Time.days(1)) .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite) .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) .build(); // 开启ttl timeStateDescriptor.enableTimeToLive(stateTtlConfig); this.timeState = getRuntimeContext().getState(timeStateDescriptor); } @Override public void processElement(MidInfo3 value, ProcessFunction<MidInfo3, Result>.Context ctx, Collector<Result> out) throws Exception { //获取到当前的状态值 if (null==timeState.value()){ timeState.update(value.getAnswerTime()-value.getRequesTime()); }else { if ((value.getAnswerTime() - value.getRequesTime()) < timeState.value()) { timeState.update((value.getAnswerTime() - value.getRequesTime())); } } out.collect(new Result(value.getFuncId(), value.getFuncIdDesc(), value.getServerIp(), timeState.value())); } }).print("结果是"); env.execute(); } } 相关的数据样例如下; {"ip":"10.125.8.20230525_0856","data":"请求: -- 14:28:05.111 -- <44.15050>1D971BEEF138370\nAction=100\nMobileCode=13304431188\nReqno=380\niPhoneKey=1681799375200\nCFrom=dbzq.android\nTFrom=newandroid\nGateWayIp=124.234.116.150\nHandleSerialNo=20230525_0856lmuqbAABLOgVTU/3lQOcAAAClAAAABQAAAP9ZAACQHAAAAAAAAAAAAACQHAAAdAAAAGJIZDhiSzVUQUFBVWVOaFNVLzNsUU5ZQUFBREhEd0FBQXdBQ0FBa0FBQUNRSEFBQUFBQUFBQUFBQUFDUUhBQUFJZ0FBQUFGSUFBQUFBQUZTQXdBQUFETTRNQUZKRFFBQUFERTJPREUzT1Rrek56VXlNREFBAA==\nGateWayPort=41912\nclientversion=1.01.110\ntztreqfrom=android.webview\nReqlinkType=2\nnewindex=1\nReqTag=96756351=9=2=0.2.134739166=1681799375201\ntztsno=b8e947dc8498edfb9c7605f290fc13ba\npartenerName=zzinfo\nuniqueid=1C0FF05B-D047-45B4-8212-6AD8627DBA4F\nEmptyFields=Token&\ntztSDKType=0\n"} {"ip":"10.125.8.20230525_0856","data":"应答: -- 14:28:05.111 -- <44.15050>1D971BEEF138370\nAction=100\nMobileCode=13304431188\nReqno=380\niPhoneKey=1681799375200\nCFrom=dbzq.android\nTFrom=newandroid\nGateWayIp=124.234.116.150\nHandleSerialNo=20230525_0856lmuqbAABLOgVTU/3lQOcAAAClAAAABQAAAP9ZAACQHAAAAAAAAAAAAACQHAAAdAAAAGJIZDhiSzVUQUFBVWVOaFNVLzNsUU5ZQUFBREhEd0FBQXdBQ0FBa0FBQUNRSEFBQUFBQUFBQUFBQUFDUUhBQUFJZ0FBQUFGSUFBQUFBQUZTQXdBQUFETTRNQUZKRFFBQUFERTJPREUzT1Rrek56VXlNREFBAA==\nGateWayPort=41912\nclientversion=1.01.110\ntztreqfrom=android.webview\nReqlinkType=2\nnewindex=1\nReqTag=96756351=9=2=0.2.134739166=1681799375201\ntztsno=b8e947dc8498edfb9c7605f290fc13ba\npartenerName=zzinfo\nuniqueid=1C0FF05B-D047-45B4-8212-6AD8627DBA4F\nEmptyFields=Token&\ntztSDKType=0\n"} |
