Flink学习12：DataStreaming API

 1.概念

2.DatStream编程

 需要代码实现的就3部分，1.数据源  2.转换 3.sink指定输出格式

2.1 创建环境

val env = ExecutionEnvironment.getExecutionEnvironment

语句比较固定

 2.2 数据源

 内置数据源主要有3种：

 2.2.1 文本数据源

import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
object readText {
  def main(args: Array[String]): Unit = {

    //create env
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //create datasource
    val dataStream = env.readTextFile(filePath = "C:\\doc\\temp\\1.正行项目介绍.txt")
    

    //print
    dataStream.print()
    

    //execute
    env.execute()


  }

}

2.2.2  套接字流

使用Ubantu系统自带的NC生成一个socket数据源

然后编写代码，监听socket数据

import org.apache.flink.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
object socketSourceTest {
  def main(args: Array[String]): Unit = {

    //create env
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //create datasource
    val socketDataStream = env.socketTextStream("localhost", 9999, '\n')

    //print
    socketDataStream.print()

    //execute
    env.execute()

  }

}

输出结果

2.2.3 集合数据源

 

import org.apache.flink.api.scala._
object CollectionSourceTest {

  def main(args: Array[String]): Unit = {


    //create env
    val env = ExecutionEnvironment.getExecutionEnvironment

    //collection data source
    val collectionDataStream = env.fromElements(Tuple1(1L, 2L), Tuple1(3L, 4L))


    //print
    collectionDataStream.print()

  }

}
输出结果：