[Flink] rich function

    public static void main(String[] args) throws Exception {
        // Create an execution environment 
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // Set parallelism 
        env.setParallelism(1);

        // Read data source 
        DataStreamSource<Event> dataStream = env.fromElements(
                new Event("Bob", "./cart", 1000L),
                new Event("Bob", "./cart", 1000L),
                new Event("Bob", "./cart/asdf", 1000L));


        SingleOutputStreamOperator<Integer> map = dataStream.map(new RichMapFunction<Event, Integer>() {

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                System.out.println("open The lifecycle is called : " + getRuntimeContext().getIndexOfThisSubtask() + " Task No. 1 starts ");
            }


            @Override
            public Integer map(Event event) throws Exception {
                return event.url.length();
            }

            @Override
            public void close() throws Exception {
                super.close();
                System.out.println("close The lifecycle is called : " + getRuntimeContext().getIndexOfThisSubtask() + " End of task No ");
            }
        });

        // Print 
        map.print();

        // Start 
        env.execute();
    }