1 flink 从集合中读取数据

1.1 代码

package com.study.liucf.bounded.api.source

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

/**
 * @Author liucf
 * @Date 2021/9/5
 */

case class SensorReding(id:String,timeStamp:Long,temperature:Double)
object CollectionSource {
  def main(args: Array[String]): Unit = {
    //定义一个提供数据的List
    val  dataList = List(
      SensorReding("sensor_1",1630851513,36.1),
      SensorReding("sensor_2",1630851512,36.2),
      SensorReding("sensor_3",1630851513,36.3),
      SensorReding("sensor_4",1630851514,36.4),
      SensorReding("sensor_5",1630851515,36.5),
    )

    //定义执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //读取集合数据源
    val ds: DataStream[SensorReding] = env.fromCollection(dataList)
    //输出结果到标准控制台
    ds.print()
    //启动执行器
    env.execute("liucf collection source test")

  }

}

1.2 执行结果

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
4> SensorReding(sensor_2,1630851512,36.2)
6> SensorReding(sensor_4,1630851514,36.4)
5> SensorReding(sensor_3,1630851513,36.3)
7> SensorReding(sensor_5,1630851515,36.5)
3> SensorReding(sensor_1,1630851513,36.1)

Process finished with exit code 0

2 fink 从文件中读取数据

2.1 数据文件

2.2 代码

package com.study.liucf.bounded.api.source

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

/**
 * @Author liucf
 * @Date 2021/9/5
 */
object FileSource {
  def main(args: Array[String]): Unit = {
     //创建flink执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //从文件中读取数据
    val ds = env.readTextFile("src\\main\\resources\\sensor.txt")
    //输出结果到标准控制台
    ds.print()
    //启动flink执行
    env.execute("liucf File source test")
  }
}

2.3 输出结果

SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
6> sensor_4,1630851514,36.4
5> sensor_3,1630851513,36.3
2> sensor_1,1630851513,36.1
3> sensor_2,1630851512,36.2
8> sensor_5,1630851515,36.5

Process finished with exit code 0