复杂的事件处理—我可以在ApacheFlink中打印datasteam< t>的单个元素而不使用内置的print()函数吗

sbdsn5lh 于 2021-06-25 发布在 Flink

关注(0)|答案(1)|浏览(369)

我正在尝试打印在flink中检测到的警告值
//为每个匹配的警告模式生成温度警告

DataStream<TemperatureEvent> warnings = tempPatternStream.select(
        (Map<String, MonitoringEvent> pattern) -> {
            TemperatureEvent first = (TemperatureEvent) pattern.get("first");

            return new TemperatureEvent(first.getRackID(), first.getTemperature()) ;
        }
    );

    // Print the warning and alert events to stdout

    warnings.print();

我得到如下输出（根据eventsource函数的tostring）

Rack id = 99 and temprature = 76.0

有人能告诉我，有没有什么方法可以不使用print打印datastream的值？一个例子是，如果我只想打印温度，如何访问数据流中的各个元素。
提前谢谢

apache-flink complex-event-processing

来源：https://stackoverflow.com/questions/45626368/can-i-print-individual-elements-of-datasteamt-in-apache-flink-without-using-in

1条答案

按热度按时间

tkclm6bt1#

我已经找到了访问单个元素的方法，假设我们有一个数据流

HeartRate<Integer,Integer>

它有两个属性

private Integer Patient_id ;
private  Integer HR;

//使用自定义函数生成datasteam

DataStream<HREvent> hrEventDataStream = envrionment
                .addSource(new HRGenerator()).assignTimestampsAndWatermarks(new IngestionTimeExtractor<>());

假设您已经使用custom函数生成了datasteam，现在我们可以打印heartrateevent的各个元素的值，如下所示

hrEventDataStream.keyBy(new KeySelector<HREvent, Integer>() {
    @Override
    public Integer getKey(HREvent hrEvent) throws Exception {
        return hrEvent.getPatient_id();
    }
         })
        .window(TumblingEventTimeWindows.of(milliseconds(10)))
        .apply(new WindowFunction<HREvent, Object, Integer, TimeWindow>() {
            @Override
            public void apply(Integer integer, TimeWindow timeWindow, Iterable<HREvent> iterable, Collector<Object> collector) throws Exception {

                for(HREvent in : iterable){

                    System.out.println("Patient id  = " + in.getPatient_id() + " Heart Rate  = " + in.getHR());
                }//for

            }//apply
        });

希望有帮助！

赞(0）回复(0）举报 2021-06-25

我来回答

复杂的事件处理—我可以在ApacheFlink中打印datasteam< t>的单个元素而不使用内置的print()函数吗

1条答案

相关问题

热门标签

最新问答