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

// 输入数据的样例类
case class SensorReading2(id: String, timestamp: Long, temperature: Double)

object SourceTest2 {
  def main(args: Array[String]): Unit = {
    // 创建执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val readings = List(
      SensorReading2("sensor_1", 1547718199, 35.8),
      SensorReading2("sensor_6", 1547718201, 15.4),
      SensorReading2("sensor_7", 1547718202, 6.7),
      SensorReading2("sensor_10", 1547718205, 38.1),
      SensorReading2("sensor_1", 1547718207, 37.2),
      SensorReading2("sensor_1", 1547718212, 33.5),
      SensorReading2("sensor_1", 1547718215, 38.1)
    )

    // 1. 从集合中读取数据
    val stream1: DataStream[SensorReading2] = env.fromCollection(readings)

    stream1.print("stream1")
    env.execute("source test job")
  }
}

输出结果:

stream1> SensorReading2(sensor_1,1547718199,35.8)
stream1> SensorReading2(sensor_6,1547718201,15.4)
stream1> SensorReading2(sensor_7,1547718202,6.7)
stream1> SensorReading2(sensor_10,1547718205,38.1)
stream1> SensorReading2(sensor_1,1547718207,37.2)
stream1> SensorReading2(sensor_1,1547718212,33.5)
stream1> SensorReading2(sensor_1,1547718215,38.1)