Flink-04-Flink自定义Source

Myth丶恋晨 2022-02-25 06:54 285阅读 0赞

### 1、实现并行度为1 的自定义Source ###

实现SourceFunction接口–一定要指定数据类型

package com.test.streaming.customSource;
     
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
     
    /** * @Author:renxin.tang * @Desc:自定义实现并行度为1 的source * 注意：SourceFunction和SourceContext都需要指定数据类型，如果不指定会报错； * @Date: Created in 16:07 2019/3/29 */
    public class MyNoParalleSource implements SourceFunction<Long> { 
        private boolean isRunning = true;
        private long count = 1L;
     
        /** * 主要方法;启动一个数据源 * 大部分情况下都需要在run方法中实现一个循环，这样就可以循环产生数据了 * 模拟产生从1开始的递增数字 * @param ctx * @throws Exception */
        @Override
        public void run(SourceContext<Long> ctx) throws Exception { 
     
            while(isRunning){ 
     
                ctx.collect(count);
                count++;
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
     
        /** * 取消一个cancel的时候调用的方法 */
        @Override
        public void cancel() { 
     
            isRunning=false;
        }
    }

### 2、实现并行化的自定义source ###

实现ParallelSourceFunction接口或者继承RichParallelSourceFunction类

package com.test.streaming.customSource;
     
    import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
     
    /** * @Author:renxin.tang * @Desc:自定义一个支持并行度的source * 注意：SourceFunction和SourceContext都需要指定数据类型，如果不指定会报错； * @Date: Created in 16:07 2019/3/29 */
    public class MyParalleSource implements ParallelSourceFunction<Long> { 
        private boolean isRunning = true;
        private long count = 1L;
     
        /** * 主要方法;启动一个数据源 * 大部分情况下都需要在run方法中实现一个循环，这样就可以循环产生数据了 * 模拟产生从1开始的递增数字 * @param ctx * @throws Exception */
        @Override
        public void run(SourceContext<Long> ctx) throws Exception { 
     
            while(isRunning){ 
     
                ctx.collect(count);
                count++;
                //每秒产生一条数据
                Thread.sleep(1000);
            }
        }
     
        /** * 取消一个cancel的时候调用的方法 */
        @Override
        public void cancel() { 
     
            isRunning=false;
        }
    }

### 3、测试 ###

两种方式的测试方法都是一样的，区别在于在env.addSource(new MyNoParalleSource());中添加自己实现的自定义Source即可

package com.test.streaming.customSource;
     
    import org.apache.flink.api.common.functions.MapFunction;
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.windowing.time.Time;
     
    /** * @Author:renxin.tang * @Desc:自定义的数据源测试MyNoParalleSource * @Date: Created in 15:49 2019/3/29 */
    public class StreamingDemoWithMyNoParalleSource { 
        public static void main(String[] args) throws Exception { 
            // 获取运行环境
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            //从collection中获取数据源
            DataStreamSource<Long> text = env.addSource(new MyNoParalleSource());// 测试 并行度的使用new MyParalleSource()
     
            SingleOutputStreamOperator<Long> num = text.map(new MapFunction<Long, Long>() { 
                @Override
                public Long map(Long value) throws Exception { 
                    System.out.println("接收到数据：" + value);
                    return value;
                }
            });
            // 每两秒钟处理一次数据
            SingleOutputStreamOperator<Long> sum = num.timeWindowAll(Time.seconds(2)).sum(0);
            sum.print().setParallelism(1);
     
            String jobname = StreamingDemoWithMyNoParalleSource.class.getSimpleName();
            env.execute(jobname);
     
        }
    }