You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user-zh@flink.apache.org by "samuel.qiu@ubtrobot.com" <sa...@ubtrobot.com> on 2020/09/03 01:23:20 UTC
回复: 回复：请指教一个关于时间窗的问题，非常感谢！
谢谢回复！
是Flink1.11.1的版本
以下是代码：
package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import org.apache.commons.collections.map.HashedMap;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.ubtechinc.dataplatform.flink.util.AES256;
import com.ubtechinc.dataplatform.flink.util.ConstantStr;
import com.ubtechinc.dataplatform.flink.util.MailUtils;
import com.ubtechinc.dataplatform.flink.util.SmsUtil;
import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import com.mysql.jdbc.Connection;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
* 使用广播实现动态的配置更新
*/
public class ExceptionAlertHour4{
private static final Logger LOG = LoggerFactory.getLogger(ExceptionAlertHour4.class);
public static void main(String[] args) throws Exception{
ParameterTool parameterTool = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameterTool);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//System.out.println(parameterTool.get("envFlag"));
Properties properties = new Properties();
String hdfs = null;
long maxTime = 0;
long maxSize = 0;
long inActive = 0;
DataStream<String> ds =null;
if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_TEST;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
properties.setProperty("bootstrap.servers", ConstantStr.KAFKA_IP_PORT_TEST);
properties.setProperty("zookeeper.connect", ConstantStr.ZOOKEEPER_IP_PORT_TEST);
properties.setProperty("group.id", "etl.exception.monitor.ExceptionAlertHour4-001");
Map<KafkaTopicPartition, Long> offsets = new HashedMap();
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0), 2800L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1), 2700L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2), 3300L);
ds = env.addSource(new FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
} else if (ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
properties.setProperty("bootstrap.servers", ConstantStr.KAFKA_IP_PORT_PRODUCT);
properties.setProperty("zookeeper.connect", ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
properties.setProperty("group.id", "etl.exception.monitor.ExceptionAlertHour-001");
properties.setProperty("auto.offset.reset", "earliest");
ds = env.addSource(new FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new SimpleStringSchema(), properties));
} else {
System.exit(-1);
}
// transform
SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS = ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>() {
@Override
public void flatMap(String value, Collector<Tuple4<String,String,String,Long>> out) {
//System.out.println("Kafka2Hdfs-in:" + value);
String newStr = value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t", "<brbr>");
//System.out.println("Kafka2Hdfs-newStr:" + newStr);
try {
// 解析JSON数据
JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
// 获取最新的字段值
JSONArray bodyDataArray = record.getJSONArray("body_data");
// 遍历，字段值的JSON数组，只有一个元素
for (int i = 0; i < bodyDataArray.size(); i++) {
// 获取到JSON数组的第i个元素
JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
if (bodyDataObj != null) {
Tuple4 log = Tuple4.of(
record.getString("HW-AppId"),
bodyDataObj.getString("HW-bugId"),
bodyDataObj.getString("HW-bugType"),
Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
);
out.collect(log);
}
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
});
singleDS.print();
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10))
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME '00:00:00')";
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
dataStream.print();
env.execute("etl.exception.monitor.ExceptionAlertHour");
}
public static class Result{
private String appid;
private String eventid;
private long cnt;
private Timestamp stime;
private Timestamp etime;
public String getAppid() {
return appid;
}
public void setAppid(String appid) {
this.appid = appid;
}
public String getEventid() {
return eventid;
}
public void setEventid(String eventid) {
this.eventid = eventid;
}
public long getCnt() {
return cnt;
}
public void setCnt(long cnt) {
this.cnt = cnt;
}
public Timestamp getStime(){
return stime;
}
public void setStime(Timestamp stime){
this.stime = stime;
}
public Timestamp getEtime(){
return etime;
}
public void setEtime(Timestamp etime){
this.etime = etime;
}
@Override
public String toString(){
return "ResultHour{" +
"appid=" + appid +
",eventid=" + eventid +
",cnt=" + cnt +
", stime=" + stime +
", etime=" + etime +
", SystemTime=" + System.currentTimeMillis() +
'}';
}
}
}
深圳市优必选科技股份有限公司 | 研发体系 | 产品研发中心 | 软件部
邱钺 Samuel Qiu
手机/微信: +0086 150 1356 8368
Email: Samuel.Qiu@ubtrobot.com
UBTECH Robotics ｜ www.ubtrobot.com
广东省深圳市南山区学苑大道1001号南山智园C1栋25楼
发件人： jacky-cui
发送时间： 2020-09-02 18:58
收件人： user-zh
主题： 回复：请指教一个关于时间窗的问题，非常感谢！
你这个flink是什么版本，能贴全代码吗
------------------ 原始邮件 ------------------
发件人: "user-zh" <samuel.qiu@ubtrobot.com>;
发送时间: 2020年9月2日(星期三) 下午3:20
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: 请指教一个关于时间窗的问题，非常感谢！
大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME '00:00:00')"; //希望整点结束时触发时间窗关闭
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
输出的结果是：
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12 期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
(400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39 等到这条数据上来后才触发
ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01 17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481 //2020/9/2 15:23:35}
请问一下哪里出了问题？万分感谢！
Re: 请指教一个关于时间窗的问题，非常感谢！
Posted by zilong xiao <ac...@gmail.com>.
可否发下是哪个配置，有相关的文档吗？
superainbower <su...@163.com> 于2020年9月4日周五 下午5:24写道：
> 1.11的版本已经加入了 新的配置项，避免了数据倾斜导致某个分区没数据 而不触发计算的问题
>
>
> | |
> superainbower
> |
> |
> superainbower@163.com
> |
> 签名由网易邮箱大师定制
>
>
> 在2020年09月4日 15:11，taochanglian<ta...@163.com> 写道：
> 确实是这样，比如有你多个partition，但是只有1个partition里面有数据，wm就不会执行计算。需要保证多个parititon数据都有数据。
>
> 举个例子，我这里在做测试的时候，1个topic10个partition，由于测试环境，按照key
> hash，数据只进入1个partition，就不会触发计算。后来让这个key轮询到10个parition都有数据，就可以触发计算了。
>
> 在 2020/9/4 13:14, Benchao Li 写道:
> 如果你有多个partition，并且有一些partition没有数据，这个时候的确是会存在这个问题。
> 要处理这种情况，可以了解下idle source[1]
>
> [1]
>
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#dealing-with-idle-sources
>
> samuel.qiu@ubtrobot.com <sa...@ubtrobot.com> 于2020年9月3日周四 下午3:41写道：
>
> 补充一下环境信息：
>
> 有点类似以下问题：
> 在1.11版本测试flink sql时发现一个问题,用streaming api
> 消费kafka,使用eventtime,再把stream转table,进行sql聚合,发现当kafka topic是多个分区时,flink webui
> watermarks 显示No Watermark,聚合计算也迟迟不触发计算,但当kafka
> topic只有一个分区时却能这个正常触发计算,watermarks也显示正常。
>
> 不确定是否是因为kafka多分区引起的？
>
>
>
> 发件人： samuel.qiu@ubtrobot.com
> 发送时间： 2020-09-03 09:23
> 收件人： user-zh
> 主题： 回复: 回复：请指教一个关于时间窗的问题，非常感谢！
> 谢谢回复！
>
> 是Flink1.11.1的版本
>
> 以下是代码：
> package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
> * Licensed to the Apache Software Foundation (ASF) under one
> * or more contributor license agreements. See the NOTICE file
> * distributed with this work for additional information
> * regarding copyright ownership. The ASF licenses this file
> * to you under the Apache License, Version 2.0 (the
> * "License"); you may not use this file except in compliance
> * with the License. You may obtain a copy of the License at
> *
> * http://www.apache.org/licenses/LICENSE-2.0
> *
> * Unless required by applicable law or agreed to in writing, software
> * distributed under the License is distributed on an "AS IS" BASIS,
> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> * See the License for the specific language governing permissions and
> * limitations under the License.
> */
>
> import org.apache.commons.collections.map.HashedMap;
> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
> import org.apache.flink.api.common.functions.FlatMapFunction;
> import org.apache.flink.api.common.serialization.SimpleStringEncoder;
> import org.apache.flink.api.common.serialization.SimpleStringSchema;
> import org.apache.flink.api.common.state.BroadcastState;
> import org.apache.flink.api.common.state.MapStateDescriptor;
> import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
> import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
> import org.apache.flink.api.common.typeinfo.TypeInformation;
> import org.apache.flink.api.java.tuple.Tuple4;
> import org.apache.flink.api.java.utils.ParameterTool;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.core.fs.Path;
> import org.apache.flink.runtime.state.StateBackend;
> import org.apache.flink.runtime.state.filesystem.FsStateBackend;
> import org.apache.flink.streaming.api.CheckpointingMode;
> import org.apache.flink.streaming.api.TimeCharacteristic;
> import org.apache.flink.streaming.api.datastream.BroadcastStream;
> 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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
> import org.apache.flink.streaming.api.functions.co
> .BroadcastProcessFunction;
> import
> org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
> import
> org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
> import
>
> org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
> import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
> import
> org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
> import org.apache.flink.table.api.Table;
> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
> import org.apache.flink.util.Collector;
> import org.slf4j.Logger;
> import org.slf4j.LoggerFactory;
>
> import com.alibaba.fastjson.JSON;
> import com.alibaba.fastjson.JSONArray;
> import com.alibaba.fastjson.JSONObject;
> import com.alibaba.fastjson.parser.Feature;
> import com.ubtechinc.dataplatform.flink.util.AES256;
> import com.ubtechinc.dataplatform.flink.util.ConstantStr;
> import com.ubtechinc.dataplatform.flink.util.MailUtils;
> import com.ubtechinc.dataplatform.flink.util.SmsUtil;
> import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
>
> import java.sql.DriverManager;
> import java.sql.PreparedStatement;
> import java.sql.ResultSet;
> import com.mysql.jdbc.Connection;
>
> import java.sql.Timestamp;
> import java.text.MessageFormat;
> import java.time.Duration;
> import java.util.ArrayList;
> import java.util.HashMap;
> import java.util.List;
> import java.util.Map;
> import java.util.Properties;
>
> /**
> * 使用广播实现动态的配置更新
> */
> public class ExceptionAlertHour4{
>
> private static final Logger LOG =
> LoggerFactory.getLogger(ExceptionAlertHour4.class);
>
> public static void main(String[] args) throws Exception{
> ParameterTool parameterTool = ParameterTool.fromArgs(args);
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.getConfig().setGlobalJobParameters(parameterTool);
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> //System.out.println(parameterTool.get("envFlag"));
> Properties properties = new Properties();
> String hdfs = null;
> long maxTime = 0;
> long maxSize = 0;
> long inActive = 0;
> DataStream<String> ds =null;
>
> if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
> hdfs = ConstantStr.HDFS_IP_PORT_TEST;
> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
> properties.setProperty("bootstrap.servers",
> ConstantStr.KAFKA_IP_PORT_TEST);
> properties.setProperty("zookeeper.connect",
> ConstantStr.ZOOKEEPER_IP_PORT_TEST);
> properties.setProperty("group.id",
> "etl.exception.monitor.ExceptionAlertHour4-001");
> Map<KafkaTopicPartition, Long> offsets = new HashedMap();
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0),
> 2800L);
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1),
> 2700L);
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2),
> 3300L);
> ds = env.addSource(new
> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
> SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
> } else if
> (ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
> hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
> properties.setProperty("bootstrap.servers",
> ConstantStr.KAFKA_IP_PORT_PRODUCT);
> properties.setProperty("zookeeper.connect",
> ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
> properties.setProperty("group.id",
> "etl.exception.monitor.ExceptionAlertHour-001");
> properties.setProperty("auto.offset.reset", "earliest");
> ds = env.addSource(new
> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
> SimpleStringSchema(), properties));
> } else {
> System.exit(-1);
> }
> // transform
> SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS =
> ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>()
> {
>
> @Override
> public void flatMap(String value,
> Collector<Tuple4<String,String,String,Long>> out) {
>
> //System.out.println("Kafka2Hdfs-in:" + value);
> String newStr =
> value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t",
> "<brbr>");
> //System.out.println("Kafka2Hdfs-newStr:" + newStr);
>
> try {
> // 解析JSON数据
> JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
>
> // 获取最新的字段值
> JSONArray bodyDataArray = record.getJSONArray("body_data");
> // 遍历，字段值的JSON数组，只有一个元素
> for (int i = 0; i < bodyDataArray.size(); i++) {
> // 获取到JSON数组的第i个元素
> JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
>
> if (bodyDataObj != null) {
> Tuple4 log = Tuple4.of(
> record.getString("HW-AppId"),
> bodyDataObj.getString("HW-bugId"),
> bodyDataObj.getString("HW-bugType"),
> Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
> );
> out.collect(log);
> }
> }
>
> } catch (Exception e) {
> System.out.println(e.getMessage());
> }
> }
>
> });
> singleDS.print();
> //指定eventtime字段及生成watermark
> DataStream<Tuple4<String,String,String,Long>>
> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
> WatermarkStrategy
>
>
> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
> .withIdleness(Duration.ofSeconds(10))
> .withTimestampAssigner((event, timestamp)->event.f3));
> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
> tenv.registerDataStream(
> "log",
> withTimestampsAndWatermarksDS,
> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>
>
> String sql = "select appid,eventid,cnt," +
> "(starttime + interval '8' hour ) as stime," +
> "(endtime + interval '8' hour ) as etime " +
> "from (select appid,eventid,count(*) as cnt," +
> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
> "from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1'
> HOUR),TIME '00:00:00')";
>
>
> Table table = tenv.sqlQuery(sql);
> DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
> dataStream.print();
>
> env.execute("etl.exception.monitor.ExceptionAlertHour");
> }
>
>
>
> public static class Result{
> private String appid;
> private String eventid;
> private long cnt;
> private Timestamp stime;
> private Timestamp etime;
> public String getAppid() {
> return appid;
> }
>
> public void setAppid(String appid) {
> this.appid = appid;
> }
>
> public String getEventid() {
> return eventid;
> }
>
> public void setEventid(String eventid) {
> this.eventid = eventid;
> }
>
> public long getCnt() {
> return cnt;
> }
>
> public void setCnt(long cnt) {
> this.cnt = cnt;
> }
>
>
> public Timestamp getStime(){
> return stime;
> }
>
> public void setStime(Timestamp stime){
> this.stime = stime;
> }
>
> public Timestamp getEtime(){
> return etime;
> }
>
> public void setEtime(Timestamp etime){
> this.etime = etime;
> }
>
> @Override
> public String toString(){
> return "ResultHour{" +
> "appid=" + appid +
> ",eventid=" + eventid +
> ",cnt=" + cnt +
> ", stime=" + stime +
> ", etime=" + etime +
> ", SystemTime=" + System.currentTimeMillis() +
> '}';
> }
> }
>
> }
>
>
> 发件人： jacky-cui
> 发送时间： 2020-09-02 18:58
> 收件人： user-zh
> 主题： 回复：请指教一个关于时间窗的问题，非常感谢！
> 你这个flink是什么版本，能贴全代码吗
>
>
> ------------------ 原始邮件 ------------------
> 发件人:
> "user-zh"
> <
> samuel.qiu@ubtrobot.com>;
> 发送时间: 2020年9月2日(星期三) 下午3:20
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: 请指教一个关于时间窗的问题，非常感谢！
>
>
>
> 大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
>
>
> //指定eventtime字段及生成watermark
> DataStream<Tuple4<String,String,String,Long>>
> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
> WatermarkStrategy
>
>
> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
> .withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
> .withTimestampAssigner((event, timestamp)->event.f3));
>
> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
> tenv.registerDataStream(
> "log",
> withTimestampsAndWatermarksDS,
> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>
> String sql = "select appid,eventid,cnt," +
>
> "(starttime + interval '8' hour ) as stime," +
>
> "(endtime + interval '8' hour ) as etime "
> +
> "from
> (select appid,eventid,count(*) as cnt," +
>
> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
>
> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
> "from
> log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME
> '00:00:00')"; //希望整点结束时触发时间窗关闭
>
> Table table = tenv.sqlQuery(sql);
> DataStream<Result> dataStream = tenv.toAppendStream(table,
> Result.class);
>
> 输出的结果是：
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> 期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
> (400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39
> 等到这条数据上来后才触发
> ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01
> 17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481
> //2020/9/2 15:23:35}
> 请问一下哪里出了问题？万分感谢！
>
>
>
回复： 请指教一个关于时间窗的问题，非常感谢！
Posted by superainbower <su...@163.com>.
1.11的版本已经加入了 新的配置项，避免了数据倾斜导致某个分区没数据 而不触发计算的问题
| |
superainbower
|
|
superainbower@163.com
|
签名由网易邮箱大师定制
在2020年09月4日 15:11，taochanglian<ta...@163.com> 写道：
确实是这样，比如有你多个partition，但是只有1个partition里面有数据，wm就不会执行计算。需要保证多个parititon数据都有数据。
举个例子，我这里在做测试的时候，1个topic10个partition，由于测试环境，按照key
hash，数据只进入1个partition，就不会触发计算。后来让这个key轮询到10个parition都有数据，就可以触发计算了。
在 2020/9/4 13:14, Benchao Li 写道:
如果你有多个partition，并且有一些partition没有数据，这个时候的确是会存在这个问题。
要处理这种情况，可以了解下idle source[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#dealing-with-idle-sources
samuel.qiu@ubtrobot.com <sa...@ubtrobot.com> 于2020年9月3日周四 下午3:41写道：
补充一下环境信息：
有点类似以下问题：
在1.11版本测试flink sql时发现一个问题,用streaming api
消费kafka,使用eventtime,再把stream转table,进行sql聚合,发现当kafka topic是多个分区时,flink webui
watermarks 显示No Watermark,聚合计算也迟迟不触发计算,但当kafka
topic只有一个分区时却能这个正常触发计算,watermarks也显示正常。
不确定是否是因为kafka多分区引起的？
发件人： samuel.qiu@ubtrobot.com
发送时间： 2020-09-03 09:23
收件人： user-zh
主题： 回复: 回复：请指教一个关于时间窗的问题，非常感谢！
谢谢回复！
是Flink1.11.1的版本
以下是代码：
package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import org.apache.commons.collections.map.HashedMap;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.functions.co
.BroadcastProcessFunction;
import
org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import
org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import
org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import
org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.ubtechinc.dataplatform.flink.util.AES256;
import com.ubtechinc.dataplatform.flink.util.ConstantStr;
import com.ubtechinc.dataplatform.flink.util.MailUtils;
import com.ubtechinc.dataplatform.flink.util.SmsUtil;
import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import com.mysql.jdbc.Connection;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
* 使用广播实现动态的配置更新
*/
public class ExceptionAlertHour4{
private static final Logger LOG =
LoggerFactory.getLogger(ExceptionAlertHour4.class);
public static void main(String[] args) throws Exception{
ParameterTool parameterTool = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameterTool);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//System.out.println(parameterTool.get("envFlag"));
Properties properties = new Properties();
String hdfs = null;
long maxTime = 0;
long maxSize = 0;
long inActive = 0;
DataStream<String> ds =null;
if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_TEST;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
properties.setProperty("bootstrap.servers",
ConstantStr.KAFKA_IP_PORT_TEST);
properties.setProperty("zookeeper.connect",
ConstantStr.ZOOKEEPER_IP_PORT_TEST);
properties.setProperty("group.id",
"etl.exception.monitor.ExceptionAlertHour4-001");
Map<KafkaTopicPartition, Long> offsets = new HashedMap();
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0),
2800L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1),
2700L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2),
3300L);
ds = env.addSource(new
FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
} else if
(ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
properties.setProperty("bootstrap.servers",
ConstantStr.KAFKA_IP_PORT_PRODUCT);
properties.setProperty("zookeeper.connect",
ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
properties.setProperty("group.id",
"etl.exception.monitor.ExceptionAlertHour-001");
properties.setProperty("auto.offset.reset", "earliest");
ds = env.addSource(new
FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
SimpleStringSchema(), properties));
} else {
System.exit(-1);
}
// transform
SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS =
ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>()
{
@Override
public void flatMap(String value,
Collector<Tuple4<String,String,String,Long>> out) {
//System.out.println("Kafka2Hdfs-in:" + value);
String newStr =
value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t",
"<brbr>");
//System.out.println("Kafka2Hdfs-newStr:" + newStr);
try {
// 解析JSON数据
JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
// 获取最新的字段值
JSONArray bodyDataArray = record.getJSONArray("body_data");
// 遍历，字段值的JSON数组，只有一个元素
for (int i = 0; i < bodyDataArray.size(); i++) {
// 获取到JSON数组的第i个元素
JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
if (bodyDataObj != null) {
Tuple4 log = Tuple4.of(
record.getString("HW-AppId"),
bodyDataObj.getString("HW-bugId"),
bodyDataObj.getString("HW-bugType"),
Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
);
out.collect(log);
}
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
});
singleDS.print();
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>>
withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10))
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1'
HOUR),TIME '00:00:00')";
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
dataStream.print();
env.execute("etl.exception.monitor.ExceptionAlertHour");
}
public static class Result{
private String appid;
private String eventid;
private long cnt;
private Timestamp stime;
private Timestamp etime;
public String getAppid() {
return appid;
}
public void setAppid(String appid) {
this.appid = appid;
}
public String getEventid() {
return eventid;
}
public void setEventid(String eventid) {
this.eventid = eventid;
}
public long getCnt() {
return cnt;
}
public void setCnt(long cnt) {
this.cnt = cnt;
}
public Timestamp getStime(){
return stime;
}
public void setStime(Timestamp stime){
this.stime = stime;
}
public Timestamp getEtime(){
return etime;
}
public void setEtime(Timestamp etime){
this.etime = etime;
}
@Override
public String toString(){
return "ResultHour{" +
"appid=" + appid +
",eventid=" + eventid +
",cnt=" + cnt +
", stime=" + stime +
", etime=" + etime +
", SystemTime=" + System.currentTimeMillis() +
'}';
}
}
}
发件人： jacky-cui
发送时间： 2020-09-02 18:58
收件人： user-zh
主题： 回复：请指教一个关于时间窗的问题，非常感谢！
你这个flink是什么版本，能贴全代码吗
------------------ 原始邮件 ------------------
发件人:
"user-zh"
<
samuel.qiu@ubtrobot.com>;
发送时间: 2020年9月2日(星期三) 下午3:20
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: 请指教一个关于时间窗的问题，非常感谢！
大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>>
withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime "
+
"from
(select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from
log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME
'00:00:00')"; //希望整点结束时触发时间窗关闭
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table,
Result.class);
输出的结果是：
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
(400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39
等到这条数据上来后才触发
ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01
17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481
//2020/9/2 15:23:35}
请问一下哪里出了问题？万分感谢！
Re: 回复：请指教一个关于时间窗的问题，非常感谢！
Posted by taochanglian <ta...@163.com>.
确实是这样，比如有你多个partition，但是只有1个partition里面有数据，wm就不会执行计算。需要保证多个parititon数据都有数据。
举个例子，我这里在做测试的时候，1个topic10个partition，由于测试环境，按照key
hash，数据只进入1个partition，就不会触发计算。后来让这个key轮询到10个parition都有数据，就可以触发计算了。
在 2020/9/4 13:14, Benchao Li 写道:
> 如果你有多个partition，并且有一些partition没有数据，这个时候的确是会存在这个问题。
> 要处理这种情况，可以了解下idle source[1]
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#dealing-with-idle-sources
>
> samuel.qiu@ubtrobot.com <sa...@ubtrobot.com> 于2020年9月3日周四 下午3:41写道：
>
>> 补充一下环境信息：
>>
>> 有点类似以下问题：
>> 在1.11版本测试flink sql时发现一个问题,用streaming api
>> 消费kafka,使用eventtime,再把stream转table,进行sql聚合,发现当kafka topic是多个分区时,flink webui
>> watermarks 显示No Watermark,聚合计算也迟迟不触发计算,但当kafka
>> topic只有一个分区时却能这个正常触发计算,watermarks也显示正常。
>>
>> 不确定是否是因为kafka多分区引起的？
>>
>>
>>
>> 发件人： samuel.qiu@ubtrobot.com
>> 发送时间： 2020-09-03 09:23
>> 收件人： user-zh
>> 主题： 回复: 回复：请指教一个关于时间窗的问题，非常感谢！
>> 谢谢回复！
>>
>> 是Flink1.11.1的版本
>>
>> 以下是代码：
>> package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
>> * Licensed to the Apache Software Foundation (ASF) under one
>> * or more contributor license agreements. See the NOTICE file
>> * distributed with this work for additional information
>> * regarding copyright ownership. The ASF licenses this file
>> * to you under the Apache License, Version 2.0 (the
>> * "License"); you may not use this file except in compliance
>> * with the License. You may obtain a copy of the License at
>> *
>> * http://www.apache.org/licenses/LICENSE-2.0
>> *
>> * Unless required by applicable law or agreed to in writing, software
>> * distributed under the License is distributed on an "AS IS" BASIS,
>> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
>> * See the License for the specific language governing permissions and
>> * limitations under the License.
>> */
>>
>> import org.apache.commons.collections.map.HashedMap;
>> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
>> import org.apache.flink.api.common.functions.FlatMapFunction;
>> import org.apache.flink.api.common.serialization.SimpleStringEncoder;
>> import org.apache.flink.api.common.serialization.SimpleStringSchema;
>> import org.apache.flink.api.common.state.BroadcastState;
>> import org.apache.flink.api.common.state.MapStateDescriptor;
>> import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
>> import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
>> import org.apache.flink.api.common.typeinfo.TypeInformation;
>> import org.apache.flink.api.java.tuple.Tuple4;
>> import org.apache.flink.api.java.utils.ParameterTool;
>> import org.apache.flink.configuration.Configuration;
>> import org.apache.flink.core.fs.Path;
>> import org.apache.flink.runtime.state.StateBackend;
>> import org.apache.flink.runtime.state.filesystem.FsStateBackend;
>> import org.apache.flink.streaming.api.CheckpointingMode;
>> import org.apache.flink.streaming.api.TimeCharacteristic;
>> import org.apache.flink.streaming.api.datastream.BroadcastStream;
>> 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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
>> import org.apache.flink.streaming.api.functions.co
>> .BroadcastProcessFunction;
>> import
>> org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
>> import
>> org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
>> import
>> org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
>> import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
>> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
>> import
>> org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
>> import org.apache.flink.table.api.Table;
>> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
>> import org.apache.flink.util.Collector;
>> import org.slf4j.Logger;
>> import org.slf4j.LoggerFactory;
>>
>> import com.alibaba.fastjson.JSON;
>> import com.alibaba.fastjson.JSONArray;
>> import com.alibaba.fastjson.JSONObject;
>> import com.alibaba.fastjson.parser.Feature;
>> import com.ubtechinc.dataplatform.flink.util.AES256;
>> import com.ubtechinc.dataplatform.flink.util.ConstantStr;
>> import com.ubtechinc.dataplatform.flink.util.MailUtils;
>> import com.ubtechinc.dataplatform.flink.util.SmsUtil;
>> import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
>>
>> import java.sql.DriverManager;
>> import java.sql.PreparedStatement;
>> import java.sql.ResultSet;
>> import com.mysql.jdbc.Connection;
>>
>> import java.sql.Timestamp;
>> import java.text.MessageFormat;
>> import java.time.Duration;
>> import java.util.ArrayList;
>> import java.util.HashMap;
>> import java.util.List;
>> import java.util.Map;
>> import java.util.Properties;
>>
>> /**
>> * 使用广播实现动态的配置更新
>> */
>> public class ExceptionAlertHour4{
>>
>> private static final Logger LOG =
>> LoggerFactory.getLogger(ExceptionAlertHour4.class);
>>
>> public static void main(String[] args) throws Exception{
>> ParameterTool parameterTool = ParameterTool.fromArgs(args);
>> final StreamExecutionEnvironment env =
>> StreamExecutionEnvironment.getExecutionEnvironment();
>> env.getConfig().setGlobalJobParameters(parameterTool);
>> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
>> //System.out.println(parameterTool.get("envFlag"));
>> Properties properties = new Properties();
>> String hdfs = null;
>> long maxTime = 0;
>> long maxSize = 0;
>> long inActive = 0;
>> DataStream<String> ds =null;
>>
>> if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
>> hdfs = ConstantStr.HDFS_IP_PORT_TEST;
>> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
>> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
>> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
>> properties.setProperty("bootstrap.servers",
>> ConstantStr.KAFKA_IP_PORT_TEST);
>> properties.setProperty("zookeeper.connect",
>> ConstantStr.ZOOKEEPER_IP_PORT_TEST);
>> properties.setProperty("group.id",
>> "etl.exception.monitor.ExceptionAlertHour4-001");
>> Map<KafkaTopicPartition, Long> offsets = new HashedMap();
>> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0),
>> 2800L);
>> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1),
>> 2700L);
>> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2),
>> 3300L);
>> ds = env.addSource(new
>> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
>> SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
>> } else if
>> (ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
>> hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
>> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
>> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
>> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
>> properties.setProperty("bootstrap.servers",
>> ConstantStr.KAFKA_IP_PORT_PRODUCT);
>> properties.setProperty("zookeeper.connect",
>> ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
>> properties.setProperty("group.id",
>> "etl.exception.monitor.ExceptionAlertHour-001");
>> properties.setProperty("auto.offset.reset", "earliest");
>> ds = env.addSource(new
>> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
>> SimpleStringSchema(), properties));
>> } else {
>> System.exit(-1);
>> }
>> // transform
>> SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS =
>> ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>()
>> {
>>
>> @Override
>> public void flatMap(String value,
>> Collector<Tuple4<String,String,String,Long>> out) {
>>
>> //System.out.println("Kafka2Hdfs-in:" + value);
>> String newStr =
>> value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t",
>> "<brbr>");
>> //System.out.println("Kafka2Hdfs-newStr:" + newStr);
>>
>> try {
>> // 解析JSON数据
>> JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
>>
>> // 获取最新的字段值
>> JSONArray bodyDataArray = record.getJSONArray("body_data");
>> // 遍历，字段值的JSON数组，只有一个元素
>> for (int i = 0; i < bodyDataArray.size(); i++) {
>> // 获取到JSON数组的第i个元素
>> JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
>>
>> if (bodyDataObj != null) {
>> Tuple4 log = Tuple4.of(
>> record.getString("HW-AppId"),
>> bodyDataObj.getString("HW-bugId"),
>> bodyDataObj.getString("HW-bugType"),
>> Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
>> );
>> out.collect(log);
>> }
>> }
>>
>> } catch (Exception e) {
>> System.out.println(e.getMessage());
>> }
>> }
>>
>> });
>> singleDS.print();
>> //指定eventtime字段及生成watermark
>> DataStream<Tuple4<String,String,String,Long>>
>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
>> WatermarkStrategy
>>
>> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
>> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
>> .withIdleness(Duration.ofSeconds(10))
>> .withTimestampAssigner((event, timestamp)->event.f3));
>> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
>> tenv.registerDataStream(
>> "log",
>> withTimestampsAndWatermarksDS,
>> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>>
>>
>> String sql = "select appid,eventid,cnt," +
>> "(starttime + interval '8' hour ) as stime," +
>> "(endtime + interval '8' hour ) as etime " +
>> "from (select appid,eventid,count(*) as cnt," +
>> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
>> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
>> "from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1'
>> HOUR),TIME '00:00:00')";
>>
>>
>> Table table = tenv.sqlQuery(sql);
>> DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
>> dataStream.print();
>>
>> env.execute("etl.exception.monitor.ExceptionAlertHour");
>> }
>>
>>
>>
>> public static class Result{
>> private String appid;
>> private String eventid;
>> private long cnt;
>> private Timestamp stime;
>> private Timestamp etime;
>> public String getAppid() {
>> return appid;
>> }
>>
>> public void setAppid(String appid) {
>> this.appid = appid;
>> }
>>
>> public String getEventid() {
>> return eventid;
>> }
>>
>> public void setEventid(String eventid) {
>> this.eventid = eventid;
>> }
>>
>> public long getCnt() {
>> return cnt;
>> }
>>
>> public void setCnt(long cnt) {
>> this.cnt = cnt;
>> }
>>
>>
>> public Timestamp getStime(){
>> return stime;
>> }
>>
>> public void setStime(Timestamp stime){
>> this.stime = stime;
>> }
>>
>> public Timestamp getEtime(){
>> return etime;
>> }
>>
>> public void setEtime(Timestamp etime){
>> this.etime = etime;
>> }
>>
>> @Override
>> public String toString(){
>> return "ResultHour{" +
>> "appid=" + appid +
>> ",eventid=" + eventid +
>> ",cnt=" + cnt +
>> ", stime=" + stime +
>> ", etime=" + etime +
>> ", SystemTime=" + System.currentTimeMillis() +
>> '}';
>> }
>> }
>>
>> }
>>
>>
>> 发件人： jacky-cui
>> 发送时间： 2020-09-02 18:58
>> 收件人： user-zh
>> 主题： 回复：请指教一个关于时间窗的问题，非常感谢！
>> 你这个flink是什么版本，能贴全代码吗
>>
>>
>> ------------------ 原始邮件 ------------------
>> 发件人:
>> "user-zh"
>> <
>> samuel.qiu@ubtrobot.com>;
>> 发送时间: 2020年9月2日(星期三) 下午3:20
>> 收件人: "user-zh"<user-zh@flink.apache.org>;
>>
>> 主题: 请指教一个关于时间窗的问题，非常感谢！
>>
>>
>>
>> 大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
>>
>> //指定eventtime字段及生成watermark
>> DataStream<Tuple4<String,String,String,Long>>
>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
>> WatermarkStrategy
>>
>> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
>> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
>> .withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
>> .withTimestampAssigner((event, timestamp)->event.f3));
>>
>> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
>> tenv.registerDataStream(
>> "log",
>> withTimestampsAndWatermarksDS,
>> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>>
>> String sql = "select appid,eventid,cnt," +
>>
>> "(starttime + interval '8' hour ) as stime," +
>>
>> "(endtime + interval '8' hour ) as etime "
>> +
>> "from
>> (select appid,eventid,count(*) as cnt," +
>>
>> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
>>
>> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
>> "from
>> log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME
>> '00:00:00')"; //希望整点结束时触发时间窗关闭
>>
>> Table table = tenv.sqlQuery(sql);
>> DataStream<Result> dataStream = tenv.toAppendStream(table,
>> Result.class);
>>
>> 输出的结果是：
>> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
>> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
>> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
>> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
>> 期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
>> (400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39
>> 等到这条数据上来后才触发
>> ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01
>> 17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481
>> //2020/9/2 15:23:35}
>> 请问一下哪里出了问题？万分感谢！
>>
>
Re: 回复：请指教一个关于时间窗的问题，非常感谢！
Posted by Benchao Li <li...@apache.org>.
如果你有多个partition，并且有一些partition没有数据，这个时候的确是会存在这个问题。
要处理这种情况，可以了解下idle source[1]
[1]
https://ci.apache.org/projects/flink/flink-docs-master/dev/event_timestamps_watermarks.html#dealing-with-idle-sources
samuel.qiu@ubtrobot.com <sa...@ubtrobot.com> 于2020年9月3日周四 下午3:41写道：
> 补充一下环境信息：
>
> 有点类似以下问题：
> 在1.11版本测试flink sql时发现一个问题,用streaming api
> 消费kafka,使用eventtime,再把stream转table,进行sql聚合,发现当kafka topic是多个分区时,flink webui
> watermarks 显示No Watermark,聚合计算也迟迟不触发计算,但当kafka
> topic只有一个分区时却能这个正常触发计算,watermarks也显示正常。
>
> 不确定是否是因为kafka多分区引起的？
>
>
>
> 发件人： samuel.qiu@ubtrobot.com
> 发送时间： 2020-09-03 09:23
> 收件人： user-zh
> 主题： 回复: 回复：请指教一个关于时间窗的问题，非常感谢！
> 谢谢回复！
>
> 是Flink1.11.1的版本
>
> 以下是代码：
> package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
> * Licensed to the Apache Software Foundation (ASF) under one
> * or more contributor license agreements. See the NOTICE file
> * distributed with this work for additional information
> * regarding copyright ownership. The ASF licenses this file
> * to you under the Apache License, Version 2.0 (the
> * "License"); you may not use this file except in compliance
> * with the License. You may obtain a copy of the License at
> *
> * http://www.apache.org/licenses/LICENSE-2.0
> *
> * Unless required by applicable law or agreed to in writing, software
> * distributed under the License is distributed on an "AS IS" BASIS,
> * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
> * See the License for the specific language governing permissions and
> * limitations under the License.
> */
>
> import org.apache.commons.collections.map.HashedMap;
> import org.apache.flink.api.common.eventtime.WatermarkStrategy;
> import org.apache.flink.api.common.functions.FlatMapFunction;
> import org.apache.flink.api.common.serialization.SimpleStringEncoder;
> import org.apache.flink.api.common.serialization.SimpleStringSchema;
> import org.apache.flink.api.common.state.BroadcastState;
> import org.apache.flink.api.common.state.MapStateDescriptor;
> import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
> import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
> import org.apache.flink.api.common.typeinfo.TypeInformation;
> import org.apache.flink.api.java.tuple.Tuple4;
> import org.apache.flink.api.java.utils.ParameterTool;
> import org.apache.flink.configuration.Configuration;
> import org.apache.flink.core.fs.Path;
> import org.apache.flink.runtime.state.StateBackend;
> import org.apache.flink.runtime.state.filesystem.FsStateBackend;
> import org.apache.flink.streaming.api.CheckpointingMode;
> import org.apache.flink.streaming.api.TimeCharacteristic;
> import org.apache.flink.streaming.api.datastream.BroadcastStream;
> 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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
> import org.apache.flink.streaming.api.functions.co
> .BroadcastProcessFunction;
> import
> org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
> import
> org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
> import
> org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
> import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
> import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
> import
> org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
> import org.apache.flink.table.api.Table;
> import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
> import org.apache.flink.util.Collector;
> import org.slf4j.Logger;
> import org.slf4j.LoggerFactory;
>
> import com.alibaba.fastjson.JSON;
> import com.alibaba.fastjson.JSONArray;
> import com.alibaba.fastjson.JSONObject;
> import com.alibaba.fastjson.parser.Feature;
> import com.ubtechinc.dataplatform.flink.util.AES256;
> import com.ubtechinc.dataplatform.flink.util.ConstantStr;
> import com.ubtechinc.dataplatform.flink.util.MailUtils;
> import com.ubtechinc.dataplatform.flink.util.SmsUtil;
> import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
>
> import java.sql.DriverManager;
> import java.sql.PreparedStatement;
> import java.sql.ResultSet;
> import com.mysql.jdbc.Connection;
>
> import java.sql.Timestamp;
> import java.text.MessageFormat;
> import java.time.Duration;
> import java.util.ArrayList;
> import java.util.HashMap;
> import java.util.List;
> import java.util.Map;
> import java.util.Properties;
>
> /**
> * 使用广播实现动态的配置更新
> */
> public class ExceptionAlertHour4{
>
> private static final Logger LOG =
> LoggerFactory.getLogger(ExceptionAlertHour4.class);
>
> public static void main(String[] args) throws Exception{
> ParameterTool parameterTool = ParameterTool.fromArgs(args);
> final StreamExecutionEnvironment env =
> StreamExecutionEnvironment.getExecutionEnvironment();
> env.getConfig().setGlobalJobParameters(parameterTool);
> env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
> //System.out.println(parameterTool.get("envFlag"));
> Properties properties = new Properties();
> String hdfs = null;
> long maxTime = 0;
> long maxSize = 0;
> long inActive = 0;
> DataStream<String> ds =null;
>
> if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
> hdfs = ConstantStr.HDFS_IP_PORT_TEST;
> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
> properties.setProperty("bootstrap.servers",
> ConstantStr.KAFKA_IP_PORT_TEST);
> properties.setProperty("zookeeper.connect",
> ConstantStr.ZOOKEEPER_IP_PORT_TEST);
> properties.setProperty("group.id",
> "etl.exception.monitor.ExceptionAlertHour4-001");
> Map<KafkaTopicPartition, Long> offsets = new HashedMap();
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0),
> 2800L);
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1),
> 2700L);
> offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2),
> 3300L);
> ds = env.addSource(new
> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
> SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
> } else if
> (ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
> hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
> maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
> maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
> inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
> properties.setProperty("bootstrap.servers",
> ConstantStr.KAFKA_IP_PORT_PRODUCT);
> properties.setProperty("zookeeper.connect",
> ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
> properties.setProperty("group.id",
> "etl.exception.monitor.ExceptionAlertHour-001");
> properties.setProperty("auto.offset.reset", "earliest");
> ds = env.addSource(new
> FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new
> SimpleStringSchema(), properties));
> } else {
> System.exit(-1);
> }
> // transform
> SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS =
> ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>()
> {
>
> @Override
> public void flatMap(String value,
> Collector<Tuple4<String,String,String,Long>> out) {
>
> //System.out.println("Kafka2Hdfs-in:" + value);
> String newStr =
> value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t",
> "<brbr>");
> //System.out.println("Kafka2Hdfs-newStr:" + newStr);
>
> try {
> // 解析JSON数据
> JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
>
> // 获取最新的字段值
> JSONArray bodyDataArray = record.getJSONArray("body_data");
> // 遍历，字段值的JSON数组，只有一个元素
> for (int i = 0; i < bodyDataArray.size(); i++) {
> // 获取到JSON数组的第i个元素
> JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
>
> if (bodyDataObj != null) {
> Tuple4 log = Tuple4.of(
> record.getString("HW-AppId"),
> bodyDataObj.getString("HW-bugId"),
> bodyDataObj.getString("HW-bugType"),
> Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
> );
> out.collect(log);
> }
> }
>
> } catch (Exception e) {
> System.out.println(e.getMessage());
> }
> }
>
> });
> singleDS.print();
> //指定eventtime字段及生成watermark
> DataStream<Tuple4<String,String,String,Long>>
> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
> WatermarkStrategy
>
> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
> .withIdleness(Duration.ofSeconds(10))
> .withTimestampAssigner((event, timestamp)->event.f3));
> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
> tenv.registerDataStream(
> "log",
> withTimestampsAndWatermarksDS,
> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>
>
> String sql = "select appid,eventid,cnt," +
> "(starttime + interval '8' hour ) as stime," +
> "(endtime + interval '8' hour ) as etime " +
> "from (select appid,eventid,count(*) as cnt," +
> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
> "from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1'
> HOUR),TIME '00:00:00')";
>
>
> Table table = tenv.sqlQuery(sql);
> DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
> dataStream.print();
>
> env.execute("etl.exception.monitor.ExceptionAlertHour");
> }
>
>
>
> public static class Result{
> private String appid;
> private String eventid;
> private long cnt;
> private Timestamp stime;
> private Timestamp etime;
> public String getAppid() {
> return appid;
> }
>
> public void setAppid(String appid) {
> this.appid = appid;
> }
>
> public String getEventid() {
> return eventid;
> }
>
> public void setEventid(String eventid) {
> this.eventid = eventid;
> }
>
> public long getCnt() {
> return cnt;
> }
>
> public void setCnt(long cnt) {
> this.cnt = cnt;
> }
>
>
> public Timestamp getStime(){
> return stime;
> }
>
> public void setStime(Timestamp stime){
> this.stime = stime;
> }
>
> public Timestamp getEtime(){
> return etime;
> }
>
> public void setEtime(Timestamp etime){
> this.etime = etime;
> }
>
> @Override
> public String toString(){
> return "ResultHour{" +
> "appid=" + appid +
> ",eventid=" + eventid +
> ",cnt=" + cnt +
> ", stime=" + stime +
> ", etime=" + etime +
> ", SystemTime=" + System.currentTimeMillis() +
> '}';
> }
> }
>
> }
>
>
> 发件人： jacky-cui
> 发送时间： 2020-09-02 18:58
> 收件人： user-zh
> 主题： 回复：请指教一个关于时间窗的问题，非常感谢！
> 你这个flink是什么版本，能贴全代码吗
>
>
> ------------------ 原始邮件 ------------------
> 发件人:
> "user-zh"
> <
> samuel.qiu@ubtrobot.com>;
> 发送时间: 2020年9月2日(星期三) 下午3:20
> 收件人: "user-zh"<user-zh@flink.apache.org>;
>
> 主题: 请指教一个关于时间窗的问题，非常感谢！
>
>
>
> 大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
>
> //指定eventtime字段及生成watermark
> DataStream<Tuple4<String,String,String,Long>>
> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
> WatermarkStrategy
>
> .<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
> //.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
> .withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
> .withTimestampAssigner((event, timestamp)->event.f3));
>
> StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
> tenv.registerDataStream(
> "log",
> withTimestampsAndWatermarksDS,
> "appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
>
> String sql = "select appid,eventid,cnt," +
>
> "(starttime + interval '8' hour ) as stime," +
>
> "(endtime + interval '8' hour ) as etime "
> +
> "from
> (select appid,eventid,count(*) as cnt," +
>
> "TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
>
> "TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
> "from
> log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME
> '00:00:00')"; //希望整点结束时触发时间窗关闭
>
> Table table = tenv.sqlQuery(sql);
> DataStream<Result> dataStream = tenv.toAppendStream(table,
> Result.class);
>
> 输出的结果是：
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> (400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
> 期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
> (400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39
> 等到这条数据上来后才触发
> ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01
> 17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481
> //2020/9/2 15:23:35}
> 请问一下哪里出了问题？万分感谢！
>
--
Best,
Benchao Li
回复: 回复：请指教一个关于时间窗的问题，非常感谢！
Posted by "samuel.qiu@ubtrobot.com" <sa...@ubtrobot.com>.
补充一下环境信息：
有点类似以下问题：
在1.11版本测试flink sql时发现一个问题,用streaming api 消费kafka,使用eventtime,再把stream转table,进行sql聚合,发现当kafka topic是多个分区时,flink webui watermarks 显示No Watermark,聚合计算也迟迟不触发计算,但当kafka topic只有一个分区时却能这个正常触发计算,watermarks也显示正常。
不确定是否是因为kafka多分区引起的？
发件人： samuel.qiu@ubtrobot.com
发送时间： 2020-09-03 09:23
收件人： user-zh
主题： 回复: 回复：请指教一个关于时间窗的问题，非常感谢！
谢谢回复！
是Flink1.11.1的版本
以下是代码：
package com.ubtechinc.dataplatform.flink.etl.exception.monitor;/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
import org.apache.commons.collections.map.HashedMap;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringEncoder;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
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.environment.CheckpointConfig.ExternalizedCheckpointCleanup;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.sink.filesystem.RollingPolicy;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.ubtechinc.dataplatform.flink.util.AES256;
import com.ubtechinc.dataplatform.flink.util.ConstantStr;
import com.ubtechinc.dataplatform.flink.util.MailUtils;
import com.ubtechinc.dataplatform.flink.util.SmsUtil;
import com.ubtechinc.dataplatform.flink.util.YearMonthDayBucketAssigner;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import com.mysql.jdbc.Connection;
import java.sql.Timestamp;
import java.text.MessageFormat;
import java.time.Duration;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
/**
* 使用广播实现动态的配置更新
*/
public class ExceptionAlertHour4{
private static final Logger LOG = LoggerFactory.getLogger(ExceptionAlertHour4.class);
public static void main(String[] args) throws Exception{
ParameterTool parameterTool = ParameterTool.fromArgs(args);
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.getConfig().setGlobalJobParameters(parameterTool);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//System.out.println(parameterTool.get("envFlag"));
Properties properties = new Properties();
String hdfs = null;
long maxTime = 0;
long maxSize = 0;
long inActive = 0;
DataStream<String> ds =null;
if (ConstantStr.ENV_FLAG_TEST.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_TEST;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_TEST;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_TEST;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_TEST;
properties.setProperty("bootstrap.servers", ConstantStr.KAFKA_IP_PORT_TEST);
properties.setProperty("zookeeper.connect", ConstantStr.ZOOKEEPER_IP_PORT_TEST);
properties.setProperty("group.id", "etl.exception.monitor.ExceptionAlertHour4-001");
Map<KafkaTopicPartition, Long> offsets = new HashedMap();
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 0), 2800L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 1), 2700L);
offsets.put(new KafkaTopicPartition(ConstantStr.KAFKA_TOPIC_EXCEPTION, 2), 3300L);
ds = env.addSource(new FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new SimpleStringSchema(), properties).setStartFromSpecificOffsets(offsets));
} else if (ConstantStr.ENV_FLAG_PRODUCT.equals(parameterTool.get("envFlag"))) {
hdfs = ConstantStr.HDFS_IP_PORT_PRODUCT;
maxTime = ConstantStr.ROLLINGPOLICY_MAXTIME_PRODUCT;
maxSize = ConstantStr.ROLLINGPOLICY_MAXSIZE_PRODUCT;
inActive = ConstantStr.ROLLINGPOLICY_INACTIVE_PRODUCT;
properties.setProperty("bootstrap.servers", ConstantStr.KAFKA_IP_PORT_PRODUCT);
properties.setProperty("zookeeper.connect", ConstantStr.ZOOKEEPER_IP_PORT_PRODUCT);
properties.setProperty("group.id", "etl.exception.monitor.ExceptionAlertHour-001");
properties.setProperty("auto.offset.reset", "earliest");
ds = env.addSource(new FlinkKafkaConsumer<>(ConstantStr.KAFKA_TOPIC_EXCEPTION, new SimpleStringSchema(), properties));
} else {
System.exit(-1);
}
// transform
SingleOutputStreamOperator<Tuple4<String,String,String,Long>> singleDS = ds.flatMap(new FlatMapFunction<String, Tuple4<String,String,String,Long>>() {
@Override
public void flatMap(String value, Collector<Tuple4<String,String,String,Long>> out) {
//System.out.println("Kafka2Hdfs-in:" + value);
String newStr = value.replaceAll("\\\\*\\r|\\\\+r|\\\\*\\n|\\\\+n|\\\\*\\t|\\\\+t", "<brbr>");
//System.out.println("Kafka2Hdfs-newStr:" + newStr);
try {
// 解析JSON数据
JSONObject record = JSON.parseObject(newStr, Feature.OrderedField);
// 获取最新的字段值
JSONArray bodyDataArray = record.getJSONArray("body_data");
// 遍历，字段值的JSON数组，只有一个元素
for (int i = 0; i < bodyDataArray.size(); i++) {
// 获取到JSON数组的第i个元素
JSONObject bodyDataObj = bodyDataArray.getJSONObject(i);
if (bodyDataObj != null) {
Tuple4 log = Tuple4.of(
record.getString("HW-AppId"),
bodyDataObj.getString("HW-bugId"),
bodyDataObj.getString("HW-bugType"),
Long.valueOf(bodyDataObj.getString("HW-happenedAt"))
);
out.collect(log);
}
}
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
});
singleDS.print();
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10))
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME '00:00:00')";
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
dataStream.print();
env.execute("etl.exception.monitor.ExceptionAlertHour");
}
public static class Result{
private String appid;
private String eventid;
private long cnt;
private Timestamp stime;
private Timestamp etime;
public String getAppid() {
return appid;
}
public void setAppid(String appid) {
this.appid = appid;
}
public String getEventid() {
return eventid;
}
public void setEventid(String eventid) {
this.eventid = eventid;
}
public long getCnt() {
return cnt;
}
public void setCnt(long cnt) {
this.cnt = cnt;
}
public Timestamp getStime(){
return stime;
}
public void setStime(Timestamp stime){
this.stime = stime;
}
public Timestamp getEtime(){
return etime;
}
public void setEtime(Timestamp etime){
this.etime = etime;
}
@Override
public String toString(){
return "ResultHour{" +
"appid=" + appid +
",eventid=" + eventid +
",cnt=" + cnt +
", stime=" + stime +
", etime=" + etime +
", SystemTime=" + System.currentTimeMillis() +
'}';
}
}
}
发件人： jacky-cui
发送时间： 2020-09-02 18:58
收件人： user-zh
主题： 回复：请指教一个关于时间窗的问题，非常感谢！
你这个flink是什么版本，能贴全代码吗
------------------ 原始邮件 ------------------
发件人: "user-zh" <samuel.qiu@ubtrobot.com>;
发送时间: 2020年9月2日(星期三) 下午3:20
收件人: "user-zh"<user-zh@flink.apache.org>;
主题: 请指教一个关于时间窗的问题，非常感谢！
大牛好：使用flink SQL,希望可以通过tumble window（每小时）来计算，现在遇到问题是到整点时，是没有触发计算的，请帮忙看看！
//指定eventtime字段及生成watermark
DataStream<Tuple4<String,String,String,Long>> withTimestampsAndWatermarksDS = singleDS.assignTimestampsAndWatermarks(
WatermarkStrategy
.<Tuple4<String,String,String,Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
//.<Tuple4<String,String,String,Long>>forMonotonousTimestamps()
.withIdleness(Duration.ofSeconds(10)) //即时没数据时，也生成watermark
.withTimestampAssigner((event, timestamp)->event.f3));
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
tenv.registerDataStream(
"log",
withTimestampsAndWatermarksDS,
"appid,bugid,eventid,rowtime.rowtime,proctime.proctime");
String sql = "select appid,eventid,cnt," +
"(starttime + interval '8' hour ) as stime," +
"(endtime + interval '8' hour ) as etime " +
"from (select appid,eventid,count(*) as cnt," +
"TUMBLE_START(rowtime,INTERVAL '1' HOUR) as starttime," +
"TUMBLE_END(rowtime,INTERVAL '1' HOUR) as endtime " +
"from log group by appid,eventid,TUMBLE(rowtime,INTERVAL '1' HOUR),TIME '00:00:00')"; //希望整点结束时触发时间窗关闭
Table table = tenv.sqlQuery(sql);
DataStream<Result> dataStream = tenv.toAppendStream(table, Result.class);
输出的结果是：
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12
(400030024,123123123123,others,1598951712000) //2020/9/1 17:15:12 期待的是2020-09-01 18:00:00.0结束时触发关闭窗口，结果是没有的。
(400030024,123123123123,others,1599030999000) //2020/9/2 15:16:39 等到这条数据上来后才触发
ResultHour{appid=400030024,eventid=others,cnt=4, stime=2020-09-01 17:00:00.0, etime=2020-09-01 18:00:00.0, SystemTime=1599031415481 //2020/9/2 15:23:35}
请问一下哪里出了问题？万分感谢！