Apache Flink:所有窗口流上的窗口函数-组合Kafka主题

Question

我正在尝试使用主题列表中的单个kafka使用者组合两个kafka主题，进一步将流中的json字符串转换为POJO。然后，通过keyBy（On事件时间字段）将它们加入，并将它们合并为单个胖json，我计划使用窗口流并在窗口流上应用窗口函数。假设主题A

我有几个问题。

1. 这种方法适合合并主题并创建单个JSON吗
2. 所有窗口流上的窗口函数似乎工作不正常；任何指点都将不胜感激

代码片段:

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

logger.info("Flink Stream Window Charger has started");

Properties properties = new Properties();

properties.setProperty("bootstrap.servers", "127.0.0.1:1030");

properties.setProperty("zookeeper.connect", "127.0.0.1:2181/service-kafka");

properties.setProperty("group.id", "group-0011");

properties.setProperty("auto.offset.reset", "smallest");



List < String > names = new ArrayList < > ();



names.add("Topic-A");

names.add("Topic-B");



DataStream < String > stream = env.addSource(new FlinkKafkaConsumer08 < > (names, new SimpleStringSchema(), properties));

DataStream < TopicPojo > pojo = stream.map(new Deserializer()).keyBy((eventTime) -> TopicPojo.getEventTime());

List < String > where = new ArrayList < String > ();

AllWindowedStream < String, GlobalWindow > data_window = pojo.flatMap(new Tokenizer()).countWindowAll(2);

DataStream < String > data_charging = data_window.apply(new MyWindowFunction());

data_charging.addSink(new SinkFunction < String > () {



public void invoke(String value) throws Exception {



  // Yet to be implemented - Merge two POJO into one 

 }

});



try

{

 env.execute();

} catch (Exception e)

{

 return;

}

}

}

class Tokenizer implements FlatMapFunction < TopicPojo, String > {

 private static final long serialVersionUID = 1 L;

 @Override

 public void flatMap(TopicPojo value, Collector < String > out) throws Exception {

  ObjectMapper mapper = new ObjectMapper();

  out.collect(mapper.writeValueAsString(value));

 }

}

class MyWindowFunction implements WindowFunction < TopicPojo, String, String, GlobalWindow > {

 @Override

 public void apply(String key, GlobalWindow window, Iterable < TopicPojo > arg2, Collector < String > out)

 throws Exception {

  int count = 0;

  for (TopicPojo in : arg2) {

   count++;

  }

  // Test Result - TO be modified

  out.collect("Window: " + window + "count: " + count);



 }

}

class Deserializer implements MapFunction < String, TopicPojo > {

 private static final long serialVersionUID = 1 L;

 @Override

 public TopicPojo map(String value) throws IOException {

  // TODO Auto-generated method stub

  ObjectMapper mapper = new ObjectMapper();

  TopicPojo obj = null;

  try {



   System.out.println(value);



   obj = mapper.readValue(value, TopicPojo.class);



  } catch (JsonParseException e) {



   // TODO Auto-generated catch block



   throw new IOException("Failed to deserialize JSON object.");



  } catch (JsonMappingException e) {



   // TODO Auto-generated catch block



   throw new IOException("Failed to deserialize JSON object.");

  } catch (IOException e) {



   // TODO Auto-generated catch block



   throw new IOException("Failed to deserialize JSON object.");

  }

  return obj;

 }

} 

我得到了-

AllWindowedStream类型中的方法Application（AllWindowFunction）不适用于参数（MyWindowFunction）错误。

Answer 1

AllWindowedStream是无键流，因此AllWindowedStreams的apply方法没有键参数。由于您正在对密钥流进行窗口化，因此data_window应该是密钥流。