update 添加 env.getConfig().setAutoWatermarkInterval(1000L);
没有解决问题。
我猜这个问题在于我的代码的另一部分。因此，首先有更多背景。
该程序从单个Kafka队列中消耗了混合消息类型的JSON流。该程序最初将其转换为ObjectNode类型的流。然后，使用.split()在大约10个单独的流中将此流拆分。这些流映射到波约斯的流。
然后将这些pojo流分配给窗口，然后将其分配给窗口（每流的POJO类型1个窗口），由由自定义量的键入并在自定义量中进行汇总，然后将其汇总到另一个Kafka队列中。
扩展的代码示例
public class flinkkafka {
public static void main(String[] args) throws Exception {
//create object mapper to allow object to JSON transform
final ObjectMapper mapper = new ObjectMapper();
final String OUTPUT_QUEUE = "test";
//setup streaming environment
StreamExecutionEnvironment env =
StreamExecutionEnvironment
.getExecutionEnvironment();
//set streaming environment variables from command line
ParameterTool parameterTool = ParameterTool.fromArgs(args);
//set time characteristic to EventTime
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
//set watermark polling interval
env.getConfig().setAutoWatermarkInterval(1000L);
//Enable checkpoints to allow for graceful recovery
env.enableCheckpointing(1000);
//set parallelism
env.setParallelism(1);
//create an initial data stream of mixed messages
DataStream<ObjectNode> messageStream = env.addSource
(new FlinkKafkaConsumer09<>(
parameterTool.getRequired("topic"),
new JSONDeserializationSchema(),
parameterTool.getProperties()))
.assignTimestampsAndWatermarks(new
BoundedOutOfOrdernessTimestampExtractor<ObjectNode>
(Time.seconds(10)){
private static final long serialVersionUID = 1L;
@Override
public long extractTimestamp(ObjectNode value) {
DateFormat format = new SimpleDateFormat("yyyy-
MM-dd HH:mm:ss", Locale.ENGLISH);
long tmp = 0L;
try {
tmp =
format.parse(value.get("EventReceivedTime")
.asText()).getTime();
} catch (ParseException e) {
e.printStackTrace();
}
System.out.println("Assigning timestamp " +
tmp);
return tmp;
}
});
//split stream by message type
SplitStream<ObjectNode> split = messageStream.split(new
OutputSelector<ObjectNode>(){
private static final long serialVersionUID = 1L;
@Override
public Iterable<String> select(ObjectNode value){
List<String> output = new ArrayList<String>();
switch (value.get("name").asText()){
case "one":
switch (value.get("info").asText()){
case "two":
output.add("info");
System.out.println("Sending message to two
stream");
break;
case "three":
output.add("three");
System.out.println("Sending message to three stream");
break;
case "four":
output.add("four");
System.out.println("Sending message to four stream");
break;
case "five":
output.add("five");
System.out.println("Sending message to five stream");
break;
case "six":
output.add("six");
System.out.println("Sending message to six stream");
break;
default:
break;
}
break;
case "seven":
output.add("seven");
System.out.println("Sending message to seven stream");
break;
case "eight":
output.add("eight");
System.out.println("Sending message to eight stream");
break;
case "nine":
output.add("nine");
System.out.println("Sending message to nine stream");
break;
case "ten":
switch (value.get("info").asText()){
case "eleven":
output.add("eleven");
System.out.println("Sending message to eleven stream");
break;
case "twelve":
output.add("twelve");
System.out.println("Sending message to twelve stream");
break;
default:
break;
}
break;
default:
output.add("failed");
break;
}
return output;
}
});
//assign splits to new data streams
DataStream<ObjectNode> two = split.select("two");
//assigning more splits to streams
//convert ObjectNodes to POJO
DataStream<Two> twoStream = two.map(new MapFunction<ObjectNode, Two>(){
private static final long serialVersionUID = 1L;
@Override
public Twomap(ObjectNode value) throws Exception {
Two stream = new Two();
stream.Time = value.get("Time").asText();
stream.value = value.get("value").asLong();
return front;
}
});
DataStream<String> keyedTwo = twoStream
.keyBy("name")
.timeWindow(Time.minutes(5))
.apply(new twoSum())
.map(new MapFunction<Two, String>(){
private static final long serialVersionUID = 1L;
@Override
public String map(Two value) throws Exception {
return mapper.writeValueAsString(value);
}
});
keyedTwo.addSink(new FlinkKafkaProducer09<String>
(parameterTool.getRequired("bootstrap.servers"),
OUTPUT_QUEUE, new SimpleStringSchema()));
env.execute();
我正在尝试使用Flink聚集Kafka队列，然后将数据流推回Kafka。聚合将使用5分钟的事件时间窗口，程序进行编译和运行，但是收集的数据永远不会使窗口传递到聚合功能，因此永远不要将消息传递给Kafka。但是，如果我评论事件时间的特征，该程序将运行并产生结果。我不知道我出了什么问题。
活动时间代码
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
ParameterTool parameterTool = ParameterTool.fromArgs(args);
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
env.enableCheckpointing(1000);
DataStream<FrontEnd> frontEndStream = frontEnd.map(new
MapFunction<ObjectNode, FrontEnd>(){
private static final long serialVersionUID = 1L;
@Override
public FrontEnd map(ObjectNode value) throws Exception {
FrontEnd front = new FrontEnd();
front.eventTime = value.get("EventReceivedTime").asText();
return front;
}
}).assignTimestampsAndWatermarks(new
BoundedOutOfOrdernessTimestampExtractor<FrontEnd>(Time.seconds(10)){
private static final long serialVersionUID = 1L;
@Override
public long extractTimestamp(FrontEnd value) {
DateFormat format = new SimpleDateFormat("yyyy-MM-
ddHH:mm:ss",Locale.ENGLISH);
long tmp = 0L;
try {
tmp = format.parse(value.eventTime).getTime();
} catch (ParseException e) {
e.printStackTrace();
}
return tmp;
}
});
DataStream<String> keyedFrontEnd = frontEndStream
.keyBy("name")
.timeWindow(Time.minutes(5))
.apply(new FrontEndSum())
.map(new MapFunction<FrontEnd, String>(){
private static final long serialVersionUID = 1L;
@Override
public String map(FrontEnd value) throws Exception {
return mapper.writeValueAsString(value);
}
});
.map(new MapFunction<FrontEnd, String>(){
private static final long serialVersionUID = 1L;
@Override
public String map(FrontEnd value) throws Exception {
return mapper.writeValueAsString(value);
}
});
keyedFrontEnd.addSink(new FlinkKafkaProducer09<String>
(parameterTool.getRequired("bootstrap.servers"), OUTPUT_QUEUE, new
SimpleStringSchema()));
env.execute();
}
}
我尝试使用附加到传入流的时间印章提取器，并将其连接到每个Pojo流。同样，该代码在事件时间内运行，并产生带有预期聚合的JSON字符串流的预期结果。但是，一旦启用了事件时间，窗口就永远不会产生结果
BoundedOutOfOrdernessTimestampExtractor实现了AssignerWithPeriodicWatermarks接口，这意味着会定期查询当前水印。
您必须通过ExecutionConfig配置轮询间隔：
env.getConfig.setAutoWatermarkInterval(1000L); // poll watermark every second
我的第一个倾向总是假设一个时区问题。
您的kafka有效载荷中"EventReceivedTime"字段的时区是什么？
SimpleDateFormat将在本地JVM时区分析：
DateFormat format = new SimpleDateFormat("yyyy-MM-ddHH:mm:ss",Locale.ENGLISH);
您可以添加
format.setTimeZone(TimeZone.getTimeZone("GMT"));
例如，如果这是文本所代表的，则将您的字符串解析为GMT。您应该确保所有日期，水印等的时区/偏移匹配，并在UTC/时期时间进行比较（这是一旦提取长时间的时间）。
<</p>@jayaananthram，是的，如果setStreamTimeCharacteristic之后的setAutoWatermarkInterval起作用。原因是setStreamTimeCharacteristic将根据代码覆盖setAutoWatermarkInterval设置的值：
public void setStreamTimeCharacteristic(TimeCharacteristic characteristic) {
this.timeCharacteristic = Preconditions.checkNotNull(characteristic);
if (characteristic == TimeCharacteristic.ProcessingTime) {
getConfig().setAutoWatermarkInterval(0);
} else {
getConfig().setAutoWatermarkInterval(200);
}
}