flink中的流媒体实现

5sxhfpxr  于 2021-06-25  发布在  Flink
关注(0)|答案(0)|浏览(237)

我的任务是从本地驱动器读取csv文件。它包含定期更新的股票数据。接下来,对读取的数据执行一些操作,并定期写回新的csv文件。
我的代码是从维基百科分析代码中改编的,问题是执行只发生一次。初始运行后,程序结束,不再保持运行状态。它不像wikipediaanalysis那样以5秒的固定间隔运行(我已经设置了)。下面是我的代码:

StreamExecutionEnvironment see = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<String> dataStream = getCsvDataStream(see);
DataStream<Stock> edits = dataStream.flatMap(new Tokenizer());
@SuppressWarnings("serial")
KeyedStream<Stock, String> keyedEdits = edits.keyBy(new KeySelector<Stock, String>() {
            @Override
            public String getKey(Stock event) {
                java.util.Date time = new java.util.Date((long) event.getTimeStamp() * 1000);
                return time.toString();
            }
        });

@SuppressWarnings("serial")
DataStream<Tuple2<Double, Double>> result = keyedEdits.timeWindow(Time.seconds(5))
                .fold(new Tuple2<>(0.0d, 0.0d), new FoldFunction<Stock, Tuple2<Double, Double>>() {
                    @Override
                    public Tuple2<Double, Double> fold(Tuple2<Double, Double> acc, Stock event) {
                        acc.f0 = event.getClose();
                        acc.f1 = event.getOpen();                           
                        return acc;
                    }
                });

你能给我个建议吗!!

暂无答案!

目前还没有任何答案,快来回答吧!

相关问题