1.集合数据源

一般用于做TestDemo的时候

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);

// 从集合中读取数据
DataStream<SensorReading> dataStream = env.fromCollection(Arrays.asList(
        new SensorReading("sensor_1", 1547718199L, 35.8),
        new SensorReading("sensor_6", 1547718201L, 15.4),
        new SensorReading("sensor_7", 1547718202L, 6.7),
        new SensorReading("sensor_10", 1547718205L, 38.1)
));

//自定义的源数据，类型必须一致
DataStream<String> integerDataStream = env.fromElements("1", "1","1","1", "1");

// 打印输出
dataStream.print("data");
integerDataStream.print("int");

// 执行
env.execute();

2.文件数据源

不做阐述,参考WordCount

3.Kafka数据源

重点掌握,生产中最常使用的,毕竟目前主流的方式就是Kafka+Flink

// 创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置并行度1
env.setParallelism(1);

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// 下面这些次要参数
properties.setProperty("group.id", "consumer-group");
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("auto.offset.reset", "latest");

// flink添加外部数据源
//topic 数据类型 配置文件
DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<String>("test", new SimpleStringSchema(),properties));

// 打印输出
dataStream.print();

env.execute();

4.自定义数据源

笔者在生产中使用的比较少,但是还是了解下比较好

/**
 * 自定义Source,可以用来模拟Kafka的输出
 */
public class CustomSource {
    public static void main(String[] args) throws Exception{

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 从文件读取数据
        DataStream<SensorReading> dataStream = env.addSource( new MySensorSource() );

        // 打印输出
        dataStream.print();

        env.execute();
    }

    public static class MySensorSource implements SourceFunction<SensorReading> {
        // 定义一个标识位，用来控制数据的产生
        private boolean running = true;

        @Override
        public void run(SourceContext<SensorReading> ctx) throws Exception {
            // 定义一个随机数发生器
            Random random = new Random();

            // 设置10个传感器的初始温度
            HashMap<String, Double> sensorTempMap = new HashMap<>();
            for( int i = 0; i < 10; i++ ){
                sensorTempMap.put("sensor_" + (i+1), 60 + random.nextGaussian() * 20);
            }

       while (running){
                for( String sensorId: sensorTempMap.keySet() ){
                    // 在当前温度基础上随机波动
                    Double newtemp = sensorTempMap.get(sensorId) + random.nextGaussian();
                    sensorTempMap.put(sensorId, newtemp);
                    ctx.collect(new SensorReading(sensorId, System.currentTimeMillis(), newtemp));
                }
                // 控制输出频率
                Thread.sleep(1000L);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }
}