[大数据]flink的算子，Source算子

一.flink算子执行过程

env -> source -> transform -> sink

1.env

根据不同的运行方式采取不同的运行环境

a.批处理

1）运行环境：ExecutionEnvironment

2）所有数据进行整体处理

3）返回的数据类型为：DataSet

b.流处理

1）运行环境：StreamExecutionEnvironment

2）流处理必须添加开启启动执行的语句： env.execute(" ")

3）不管是什么数据源都得按实时流数据进行处理

4）返回的数据类型：DataStream

2.Source算子

SourceFunction：函数

RichSourceFunction:富函数 比SourceFunction方法多，重要的两个方法：open(),close()

读取原数据的算子，创建DataStream，类似spark中的创建算子

处理的数据：文件，集合，元素，socket，kafka，自定义

写代码：主方法里进行创建环境

1）文件

a.不设置并行度的结果，如果不设置并行度，系统默认并行度为电脑的cup核数8

file:5> sensor_6,1547718201,15.40
file:6> sensor_7,1547718202,6.72
file:1> sensor_8,1547718204,12.72
file:4> sensor_1,1547718199,30.80
file:7> sensor_8,1547718203,28.72
file:2> sensor_8,1547718205,6.72

?b.设置并行度? ? env.setParallelism(1)

file> sensor_1,1547718199,30.80
file> sensor_6,1547718201,15.40
file> sensor_7,1547718202,6.72
file> sensor_8,1547718203,28.72
file> sensor_8,1547718204,12.72
file> sensor_8,1547718205,6.72

 /**
   * 读取文件类型的数据源
   * @param env 运行环境
   * @return 返回元数据
   */
  def source_File(env:StreamExecutionEnvironment) = {
    //用source算子读取原文件
    val dataStream_File:DataStream[String] =  env.readTextFile("D:/linuxResources/data/data_flink/sensor.txt")
    //输出读取到的数据
    dataStream_File.print("file")
  }


  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)

    source_File(env)

    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

2） 集合

注意：读取集合类型的数据需要进行隐式转换

?import org.apache.flink.streaming.api.scala._

 /**
   * 读取集合类型的数据源
   * @param env 运行环境
   * @return
   */
  def source_Collection(env:StreamExecutionEnvironment) = {
    //list集合数据
    val sensorList = List(
      "sensor_1,1547718199,30.80",
      "sensor_6,1547718201,15.40",
      "sensor_7,1547718202,6.72",
      "sensor_8,1547718203,28.72",
      "sensor_8,1547718204,12.72",
      "sensor_8,1547718205,6.72"
    )
    //用source算子读取集合类型的数据
    //读取集合类型的数据需要进行隐式转换  import org.apache.flink.streaming.api.scala._
    val dataStream:DataStream[String] = env.fromCollection(sensorList)
    //对得到的数据进行输出
    dataStream.print()
  }

  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)
    source_Collection(env)
    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

3）元素

注意：集合可以直接变成元素获取

/**
   * 读取元素类型的原数据
   * @param env
   * @return
   */
  def source_Items(env:StreamExecutionEnvironment) = {
    //元素数据原
    val dataStream_items:DataStream[String] = env.fromElements(
      "sensor_1,1547718199,30.80",
            "sensor_6,1547718201,15.40",
            "sensor_7,1547718202,6.72",
            "sensor_8,1547718203,28.72",
            "sensor_8,1547718204,12.72",
            "sensor_8,1547718205,6.72"
    )
    //对获取的元素进行输出
    dataStream_items.print()
  }


  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)
    source_Items(env)

    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

4）socket

a.在虚拟机上开启端口8888

nc -lk 8888?

b. 在端口上输入数据，控制台接收数据，实现实时监控数据

? ? ? ? ??

/**
   * 读取socket的数据源
   * @param env
   * @return
   */
  def source_socket(env:StreamExecutionEnvironment) = {
    //读取socket的数据源
    val dataStream_source:DataStream[String] = env.socketTextStream("node180",8888)
    //获取读到的数据
    dataStream_source.print()
  }

  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)
    source_socket(env)

    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

5）kafka

a.从kafka读取数据源,需要先导包，三点坐标

<dependency>
? ? ? ? ? ? <groupId>org.apache.spark</groupId>
? ? ? ? ? ? <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
? ? ? ? ? ? <version>2.3.3</version>
? ? ? ? </dependency>

b.启动kafka

每一个命令分别开启一个窗口进行执行

（1）启动zookeeper

（2）启动kafka，&：后台执行

（3）开启生产者，代码充当消费者

（4）原来的数据不想要，进行删除

?（1）?zkServer.sh start

?（2）kafka-server-start.sh kafka/config/server.properties ?&

（3）kafka-console-producer.sh --topic 主题名(myTopic) --broker-list? ip地址(node160):9092

（4）crontab -r

（5）关闭生产者，不会影响什么，将kafka的服务进行关闭，流处理会自动认为数据原没有了，自动终止程序

/**
   * 充当kafka的消费者，读取kafka的数据
   * @param env
   * @return
   */
  def source_kafka(env:StreamExecutionEnvironment) = {
    //设置kafka所需要的配置项
    val prop:Properties = new Properties()
    prop.setProperty("bootstrap.servers","node160:9092")
    prop.setProperty("group.id","myTopic")
    prop.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    prop.setProperty("auto.offset.reset", "latest")

    //从kafka读取数据源生成DataStream
    val dataStream_kafka:DataStream[String] = env.addSource(new FlinkKafkaConsumer[String](
      "myTopic",
      new SimpleStringSchema(),
      prop
    ))

    //输出获取到的数据
    dataStream_kafka.print()
  }

  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)
    source_kafka(env)

    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

?6）自定义

自己产生数据并且获取数据，自定义类需要实现(Rich)SourceFunction，Rich)SourceFunction 是 java 的 interface，但是在scala中第一个继承的类必须使用extends

/**
 * 正常使用SourceFunction就足够，因为此方法中没有需要初始化和最终执行一次的内容
 * 但是使用RichSourceFunction也没有问题
 * (Rich)SourceFunction 是 java 的 interface，但是在scala中第一个继承的类必须使用extends
 */
class MySource(sleepTime:Int) extends RichSourceFunction[String]{
  //初始化标识符
  var flag = true
  //定义一个随机数
  val rand = new Random() //import scala.util.Random
  //自定义产生数据
  //ctx是SourceFunction的一个内部类，也就是上下文
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    //实现每2秒发送一条数据
    while(flag){
      //产生一条sensorId数据
      val sensor_id = "sensor_"+rand.nextInt(10)
      //产生时间戳 Math.abs()绝对值
      //long类型的数据有正有负，需要取绝对值
      val sensor_timestamp = Math.abs(rand.nextLong())
      //产生温度数据 0-30之间的温度
      val sensor_temp = rand.nextDouble()*30
      //数据类型是string类型，拼成字符串
      val data = sensor_id+","+sensor_timestamp+","+sensor_temp
      //将获取到的数据发送出去
      ctx.collect(data)
      //每2秒发送一条数据
      //Thread.sleep(2000)
      //自动实现几秒生成一条数据
      Thread.sleep(sleepTime*1000)
    }
  }

  override def cancel(): Unit = {
    flag = false
  }
}
/**
   * 读取用户自定义的数据源
   * @param env 运行环境
   * @return
   */
  def source_UserDefine(env:StreamExecutionEnvironment) = {
    //source进行读取自定义的数据源
    val dataStream_UD:DataStream[String] = env.addSource(new MySource(3))
    //将获取到的数据源进行输出
    dataStream_UD.print()
  }

  def main(args: Array[String]): Unit = {
    //1.创建实时流处理的环境
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度为1
    //输出结果
    //如果不设置并行度，系统默认并行度为电脑的cup核数8，
    // 输出结果：file:5> sensor_6,1547718201,15.40
    //file:6> sensor_7,1547718202,6.72
    env.setParallelism(1)
    source_UserDefine(env)

    //2.流处理必须添加开启启动执行的语句
    env.execute("source")
  }

开发者涨薪指南

48位大咖的思考法则、工作方式、逻辑体系