https://flink.apache.org/ logo
Join Slack
Powered by
Hello! I am trying to process a CSV file source ro...
# troubleshooting
t
Hello! I am trying to process a CSV file source row by row and add watermarks to mimic real-time streaming use case. Unfortunately I can't get it to work properly, i.e. the watermark doesn't seem to be applied properly. Do people know if this is even possible? Here's my code:
Copy code
env.getConfig().setGlobalJobParameters(params);
        env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        env.setParallelism(1);
        String path = "/home/ubuntu/flink/flink-examples/flink-examples-streaming/target/csv";
        final CsvReaderFormat<Pojo> format = CsvReaderFormat.forPojo(Pojo.class);
        File csvfile = new File(path);
        final FileSource<Pojo> csv_source =
                FileSource.forRecordStreamFormat(format, Path.fromLocalFile(csvfile)).
                monitorContinuously(Duration.ofMillis(1)).build();

        DataStream<Pojo> source = env.fromSource(
                csv_source,
                new WatermarkStrategy<Pojo>() {
                    @Override
                    public WatermarkGenerator<Pojo> createWatermarkGenerator(
                            WatermarkGeneratorSupplier.Context context) {
                        return new MyPerElementWatermarkGenerator();
                    }
                }
                .withTimestampAssigner((event, timestamp) -> event.event_time), "csv-source");
d
And what does 
MyPerElementWatermarkGenerator
look like?
t
Copy code
@Internal
    public static class MyPerElementWatermarkGenerator implements WatermarkGenerator<Pojo> {

        private long maxTimestamp;

        public MyPerElementWatermarkGenerator() {
            maxTimestamp = Long.MIN_VALUE;
        }


        @Override
        public void onEvent(Pojo event, long eventTimestamp, WatermarkOutput output) {
//            if (eventTimestamp > maxTimestamp) {
//                maxTimestamp = eventTimestamp;
//                output.emitWatermark(new Watermark(maxTimestamp));
//            }
            if (event.event_time > maxTimestamp)
            {
                maxTimestamp = event.event_time;
            }

            LOG.debug("maxTimestamp: " + maxTimestamp);
            output.emitWatermark(new Watermark(maxTimestamp));
        }

        @Override
        public void onPeriodicEmit(WatermarkOutput output) {}

//        public WatermarkGeneratorSupplier<Pojo> getSupplier() {
//            return (ctx) -> new MyPerElementWatermarkGenerator();
//        }

    }
I want the watermark to be called OnEvent (on each row in the CSV), as periodic emitting a watermark probably doesn't make too much sense processing an offline CSV. I have attached the code if anybody wants to give it a shot
csvSessionWindowing.java
3 Views
Open in Slack
PreviousNext
Powered by