大数据之Flink(4) | Source

- 日理万妓 2022-12-16 03:59 154阅读 0赞

### Source ###

*   *  从集合读取数据
     *  从文件读取数据
     *  从socket中读取数据
     *  以kafka消息队列的数据作为来源
     *  自定义Source

## 从集合读取数据 ##

import org.apache.flink.streaming.api.scala._
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object SourceTest { 
      def main(args: Array[String]): Unit = { 
        // 创建执行环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        //1.从集合中读取数据
        val stream1 = env.fromCollection(List(
          SensorReading("sensor_1", 1547718199, 35.8),
          SensorReading("sensor_6", 1547718201, 15.4),
          SensorReading("sensor_7", 1547718202, 6.7),
          SensorReading("sensor_10", 1547718205, 38.1),
          SensorReading("sensor_1", 1547718207, 37.2),
          SensorReading("sensor_1", 1547718212, 33.5),
          SensorReading("sensor_1", 1547718215, 38.1)
      ))
        //打印输出
        stream1.print("a1")
        env.execute("source test job")
    }
    }

## 从文件读取数据 ##

val stream2 = env.readTextFile("D:\\idea\\Flink-Learning\\src\\main\\resources\\sensor.txt")

## 从socket中读取数据 ##

val stream3 = env.socketTextStream("hadoop102", 9999)

## 以kafka消息队列的数据作为来源 ##

添加依赖

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka-0.11_2.11</artifactId>
        <version>1.10.0</version>
    </dependency>

具体代码如下：

val properties = new Properties()
        properties.setProperty("bootstrap.servers", "hadoop12: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")
        val stream4 = env.addSource(new FlinkKafkaConsumer011[String]("sensor",new SimpleStringSchema(),properties))

在hadoop12上输入

bin/kafka-console-producer.sh --broker-list hadoop12:9092 --topic sensor

## 自定义Source ##

import org.apache.flink.streaming.api.functions.source.SourceFunction
    import org.apache.flink.streaming.api.scala._
    import scala.util.Random
    
    //输入数据的样例类
    case class SensorReading(id:String,timestamp:Long,temperature:Double)
    
    object SourceTest { 
      def main(args: Array[String]): Unit = { 
        // 创建执行环境
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        env.setParallelism(1)
    
        //自定义source
        val stream5 = env.addSource(new MySensorSource())
        
        //打印输出
        stream5.print("stream5")
        env.execute("source test job")
    }
    }
    
    //实现一个自定义的SourceFunction,自动生成测试数据
    class MySensorSource() extends SourceFunction[SensorReading]{ 
      //定义一个flag，表示数据源是否正常运行
      var running:Boolean = true
    
      //随机生成SensorReading数据
      override def run(sourceContext: SourceFunction.SourceContext[SensorReading]): Unit = { 
        //定义一个随机数发生器
        val rand = new Random()
    
        //随机生成10个传感器的温度值，并且不停在之前温度的基础上更新(随机上线波动)
        //首先生成10个传感器的初始温度
        var curTemps = 1.to(10).map(
          i => ("sensor_"+i,60 + rand.nextGaussian() * 20)
        )
    
        //无线循环，生成随机数据流
        while(running){ 
          //在当前温度基础上，随机生成微小波动
          curTemps = curTemps.map(
            data => (data._1,data._2 + rand.nextGaussian())
          )
          //获取当前系统时间
          val curTs = System.currentTimeMillis()
          //包装成样例类，用ctx=sourceContext发出数据
          curTemps.foreach(
            data => sourceContext.collect(SensorReading(data._1,curTs,data._2))
          )
          //定义间隔时间
          Thread.sleep(1000L)
        }
      }
    
      override def cancel(): Unit = running = false
    }