Flink 常见数据源

    val source: DataStream[String] = env.socketTextStream("node01", 9999)

 //从指定的文件中加载数据    val source: DataStream[String] = env.readTextFile("hello.txt")

1. 先定义一个自定义的类继承SourceFunction.

/**  * 自定义数据源加载数据  */object MySourceDemo {
  def main(args: Array[String]): Unit = {    //1.获取流处理的执行环境    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    //加载自定义数据源    val source: DataStream[Int] = env.addSource(new MySourceFunction)    //打印    source.print()    //执行程序    env.execute()  }}//自定义数据源class MySourceFunction extends SourceFunction[Int] {  /**    * 自定义数据源,通过这个方法来生成我们的数据    * @param ctx 采集器.    */  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {    var i = 0    while (true) {      //使用采集器,将数据采集走.外部source就可以获取到数据      ctx.collect(i)      i = i + 1      Thread.sleep(500)    }  }
  /**    * 如果需要取消,调用此方法.    */  override def cancel(): Unit = ???}

//加载多并行度数据源    val source: DataStream[Int] = env.addSource(new MyParallelSourceFunction).setParallelism(2)

//自定义多并行度数据源class MyParallelSourceFunction extends ParallelSourceFunction[Int] {