【发布时间】：2017-10-01 00:29:24
【问题描述】：
我正在尝试使用主题列表中的单个 kafka 消费者组合两个 kafka 主题，进一步将流中的 json 字符串转换为 POJO。然后，通过 keyBy ( On event time field ) 加入它们并将它们合并为单个 fat json，我计划使用窗口流并在窗口流上应用窗口函数。假设 Topic-A 和 Topic-B 可以在 Event Time 上加入，并且只有一对（ Topic A ( JSON ) ， Topic B (JSON ) 将出现相同的 eventTime。因此计划使用 coutWindow(2 ) 在 eventTime 上发布 keyBy。
我有几个相同的问题；
- 这种方法适合合并主题和创建单个 JSON 吗？
- All Window 流上的窗口函数似乎无法正常工作；任何指针将不胜感激。
代码片段：
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 类型中的方法 apply(AllWindowFunction) 不适用于参数 (MyWindowFunction) 错误。
【问题讨论】：
标签： json merge apache-kafka apache-flink