flink file source

 1 import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
 2 
 3 object FlinkDemo04_CommonSource_fromFile {
 4     case class Flight(avgTicketPrice:String, cancelled:String, carrier:String, dest:String, destAirportID:String, origin:String, originAirportID:String)
 5     def main(args: Array[String]): Unit = {
 6         
 7         //1 创建环境
 8         val env = StreamExecutionEnvironment.getExecutionEnvironment
 9         //2 获取DataStream
10         val dStream: DataStream[String] = env.readTextFile("I:\projectImplement\dataWareHouse\test-es\data\630data.csv")
11         //3 计算
12         import org.apache.flink.api.scala._
13         dStream.map{
14             line=>
15             val vals = line.split(",")
16             Flight(vals(0),vals(1),vals(2),vals(3),vals(4),vals(5),vals(6))
17         }.print()
18         //4 执行
19         env.execute("stream job")
20     }
21 }
View Code