快速入门Flink (7) —— 小白都喜欢看的Flink流处理之DataSources和DataSinks

<code class="javascript">yum install -y nc</code>

<code class="javascript">import org.apache.flink.api.java.tuple.Tupleimport org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.streaming.api.windowing.windows.TimeWindow/* * @Author: Alice菌 * @Date: 2020/7/9 08:40 * @Description:      */// 入门案例，单词统计object StreamWordCount {  def main(args: Array[String]): Unit = {    // 1、 创建流处理的执行环境    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、 构建数据源，使用的socket    val socketDataStream: DataStream[String] = env.socketTextStream("node01",9999, 0)    // 3、 数据的处理    val wordDataStream: DataStream[(String, Int)] = socketDataStream.flatMap(_.split(" ")).map(_ -> 1)    //4. 使用keyBy 进行分流（分组）    // 在批处理中针对于dataset， 如果分组需要使用groupby    // 在流处理中针对于datastream， 如果分组（分流）使用keyBy    val groupedDataStream: KeyedStream[(String, Int), Tuple] = wordDataStream.keyBy(0)    //5. 使用timeWinodw 指定窗口的长度（每5秒计算一次）    // spark-》reduceBykeyAndWindow    val windowDataStream: WindowedStream[(String,Int),Tuple,TimeWindow]= groupedDataStream.timeWindow(      Time.seconds(5)    )    //6. 使用sum执行累加    val sumDataStream: DataStream[(String, Int)] = windowDataStream.sum(1)    sumDataStream.print()    env.execute("StreamWordCount")  }}</code>

<code class="javascript">import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}import scala.collection.immutable.{Queue, Stack}import scala.collection.mutableimport scala.collection.mutable.{ArrayBuffer, ListBuffer}import org.apache.flink.api.scala._/* * @Author: Alice菌 * @Date: 2020/8/8 17:02 * @Description:      */object StreamDataSourceDemo {  def main(args: Array[String]): Unit = {    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 0. 用element创建DataStream    val ds0: DataStream[String] = senv.fromElements("spark","flink")    ds0.print()    // 1. 用Tuple创建DataStream    val ds1: DataStream[(Int, String)] = senv.fromElements((1,"spark"),(2,"flink"))    ds1.print()    // 2. 用Array创建DataStream    val ds2: DataStream[String] = senv.fromCollection(Array("spark","flink"))    ds2.print()    // 3. 用ArrayBuffer 创建DataStream    val ds3: DataStream[String] = senv.fromCollection(ArrayBuffer("spark","flink"))    ds3.print()    // 4. 用List创建DataStream    val ds4: DataStream[String] = senv.fromCollection(List("spark","flink"))    ds4.print()    // 5. 用List创建DataStreamm    val ds5: DataStream[String] = senv.fromCollection(ListBuffer("spark","flink"))    ds5.print()    // 6. 用Vector创建DataStream    val ds6: DataStream[String] = senv.fromCollection(Vector("spark","flink"))    ds6.print()    // 7. 用Queue创建DataStream    val ds7: DataStream[String] = senv.fromCollection(Queue("spark","flink"))    ds7.print()    // 8. 用Stack创建DataStream    val ds8: DataStream[String] = senv.fromCollection(Stack("spark", "flink"))    // 9. 用Stream创建DataStream（Stream相当于lazy List，避免在中间过程中生 成不必要的集合）    val ds9: DataStream[String] = senv.fromCollection(Stream("spark","flink"))    ds9.print()    // 10. 用Seq创建DataStream    val ds10: DataStream[String] = senv.fromCollection(Seq("spark","flink"))    ds10.print()    // 11. 用Set创建DataStream(不支持)    // val ds11: DataStream[String] = senv.fromCollection(Seq("spark", "flink"))    // ds11.print()    // 12.用Iterable创建DataStream(不支持)    // val ds12: DataStream[String] = senv.fromCollection(Iterable("spark", "flink"))    // ds12.print()    // 13.用ArraySeq创建DataStream    val ds13: DataStream[String] = senv.fromCollection(mutable.ArraySeq("spark","flink"))    ds13.print()    // 14.用 ArrayStack 创建DataStream    val ds14: DataStream[String] = senv.fromCollection(mutable.ArrayStack("spark","flink"))    ds14.print()    // 15.用Map 创建 DataStream(不支持)    //val ds15: DataStream[(Int, String)] = senv.fromCollection(Map(1 -> "spark", 2 -> "flink"))    //ds15.print()    // 16.用Range创建DataStream    val ds16: DataStream[Int] = senv.fromCollection(Range(1,9))    ds16.print()    // 17.用fromElements创建DataStream    val ds17: DataStream[Long] = senv.generateSequence(1,9)    ds17.print()    senv.execute("StreamDataSourceDemo")  }}</code>

<code class="javascript">import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}/* * @Author: Alice菌 * @Date: 2020/8/8 17:42 * @Description:    基于文件的source */object StreamFileSourceDemo {  def main(args: Array[String]): Unit = {    // 1、构建流处理的环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、基于文件的source，构建数据集    val textDStream: DataStream[String] = senv.readTextFile("data/input/wordcount.txt")    // 3、打印输出    textDStream.print()          // 4、执行程序    senv.execute("StreamFileSourceDemo")    //3> Final Memory Finished at    //10> Total time BUILD SUCCESS    //4> Flink Flink Flink Flink Flink    //9> Final Memory Finished at    //1> Total time BUILD SUCCESS    //8> Total time BUILD SUCCESS    //12> Final Memory Finished at    //6> Hive Hive Hive Hive Hive  }}</code>

<code class="javascript">import org.apache.flink.api.java.tuple.Tupleimport org.apache.flink.streaming.api.scala._import org.apache.flink.streaming.api.windowing.time.Timeimport org.apache.flink.streaming.api.windowing.windows.TimeWindow/* * @Author: Alice菌 * @Date: 2020/7/9 08:40 * @Description:     基于网络套接字的 source */// 入门案例，单词统计object StreamWordCount {  def main(args: Array[String]): Unit = {    // 1、 创建流处理的执行环境    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、 构建数据源，使用的socket    val socketDataStream: DataStream[String] = env.socketTextStream("node01",9999, 0)    // 3、 数据的处理    val wordDataStream: DataStream[(String, Int)] = socketDataStream.flatMap(_.split(" ")).map(_ -> 1)    //4. 使用keyBy 进行分流（分组）    // 在批处理中针对于dataset， 如果分组需要使用groupby    // 在流处理中针对于datastream， 如果分组（分流）使用keyBy    val groupedDataStream: KeyedStream[(String, Int), Tuple] = wordDataStream.keyBy(0)    //5. 使用timeWinodw 指定窗口的长度（每5秒计算一次）    // spark-》reduceBykeyAndWindow    val windowDataStream: WindowedStream[(String,Int),Tuple,TimeWindow]= groupedDataStream.timeWindow(      Time.seconds(5)    )    //6. 使用sum执行累加    val sumDataStream: DataStream[(String, Int)] = windowDataStream.sum(1)    sumDataStream.print()    env.execute("StreamWordCount")  }}</code>

<code class="javascript">import org.apache.flink.streaming.api.functions.source.SourceFunctionimport org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.api.scala._/* * @Author: Alice菌 * @Date: 2020/8/8 21:51 * @Description:     自定义非并行数据源 */object StreamCustomerNoParallelSourceDemo {  def main(args: Array[String]): Unit = {    // 1、创建流处理的执行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、基于自定义数据源构建数据    val longDStream: DataStream[Long] = senv.addSource(new MyNoParallelSource()).setParallelism(1)    // 3、输出打印    longDStream.print()    // 4、执行程序    senv.execute("StreamCustomerNoParallelSourceDemo")        //10> 1    //11> 2    //12> 3    //1> 4    //2> 5    //3> 6    //4> 7    //5> 8    //6> 9      }    /*  创建一个并行度为1的数据源 * 实现从1开始产生递增数字  */  class MyNoParallelSource extends SourceFunction[Long]{    // 申明一个变量number    var number:Long = 1L    var isRunning:Boolean = true    override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {            while (isRunning){        ctx.collect(number)        number += 1        // 休眠1秒        Thread.sleep(1000)        if (number == 10){          cancel()        }      }    }         override def cancel(): Unit = {      isRunning = false    }  }}</code>

<code class="javascript">import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}import org.apache.flink.streaming.api.scala.StreamExecutionEnvironmentimport org.apache.flink.streaming.api.scala._/* * @Author: Alice菌 * @Date: 2020/8/8 22:05 * @Description:     自定义创建并行数据源 */object StreamCustomerParallelSourceDemo {  def main(args: Array[String]): Unit = {    // 1、创建流处理的执行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、基于自定义ParallelSource数据源创建并行的数据    val parallelSource: DataStream[Long] = senv.addSource(new MyParallelSource()).setParallelism(5)    // 3、打印输出    parallelSource.print()        // 4、执行程序    senv.execute("StreamCustomerParallelSourceDemo")  }  /*  创建一个并行度为1的数据源 * 实现从1开始产生递增数字   */  class MyParallelSource extends ParallelSourceFunction[Long] {    // 声明一个Long类型的变量    var number:Long = 1L    // 申明一个初始化为true的Boolean变量    var isRunning: Boolean = true        override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {      while (isRunning) {        ctx.collect(number)        number += 1        if (number > 20) {          cancel()        }      }    }    override def cancel(): Unit = {      isRunning = false    }  }}</code>

<code class="javascript">import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction,SourceFunction}import org.apache.flink.streaming.api.scala.{DataStream,StreamExecutionEnvironment}import org.apache.flink.api.scala._import org.apache.flink.configuration.Configuration/* * @Author: Alice菌 * @Date: 2020/8/8 22:23 * @Description:    创建并行数据源 */object StreamCustomerRichParallelSourceDemo {  def main(args: Array[String]): Unit = {    // 1、 创建流处理运行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、 基于 RichParallelSource 并行数据源构建数据集    val richParallelSource: DataStream[Long] = senv.addSource(new MyRichParallelSource()).setParallelism(2)    // 3、 打印输出    richParallelSource.map(line => {      println("接收到的数据:" + line)      line    })    // 4、执行程序    senv.execute("StreamCustomerRichParallelSourceDemo")  }  /*     创建一个并行度为1 的数据源     实现从 1 开始产生递增数字   */  class MyRichParallelSource extends RichParallelSourceFunction[Long] {    var count: Long = 1L    var isRunning: Boolean = true    override def run(ctx: SourceFunction.SourceContext[Long]): Unit = {      while (isRunning){        ctx.collect(count)        count += 1        Thread.sleep(1000)              }    }    override def cancel(): Unit = {      isRunning = false    }    override def open(parameters: Configuration): Unit = {      super.close()    }  }}</code>

<code class="javascript">import java.util.Propertiesimport org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}import org.apache.flink.api.scala._import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011/* * @Author: Alice菌 * @Date: 2020/8/8 22:51 * @Description:    基于 kafka 的 source 操作 */object StreamKafkaSourceDemo {  def main(args: Array[String]): Unit = {    // 1、构建流处理执行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 指定消费者主题    val topic: String = "test"    // 设置参数    val props: Properties = new Properties    props.setProperty("bootstrap.servers", "node01:9092")    props.setProperty("group.id", "test")    props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")    props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")    // 基于 Flink，创建 Kafka消费者    val kafkaConsumer: FlinkKafkaConsumer011[String] = new FlinkKafkaConsumer011[String](topic,new SimpleStringSchema(),props)    // Flink 从 topic 中最新的数据开始消费    kafkaConsumer.setStartFromLatest()    // 构建基于 kafka 的数据源    val kafkaDataStream: DataStream[String] = senv.addSource(kafkaConsumer)    // 打印输出消费的数据    kafkaDataStream.print()    // 执行流处理的程序    senv.execute("StreamKafkaSourceDemo")  }}</code>

<code class="javascript">cd /export/servers/kafka_2.11-1.0.0bin/kafka-console-producer.sh --broker-list node01:9092,node02:9092,node03:9092 --topic test</code>

<code class="javascript">import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}import org.apache.flink.configuration.Configurationimport org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}/* * @Author: Alice菌 * @Date: 2020/8/8 23:52 * @Description:     基于mysql的source操作 */object StreamFromMysqlSource {  def main(args: Array[String]): Unit = {    // 1、创建流处理执行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    import org.apache.flink.api.scala._        // 2、添加自定义的 mysql 数据源对象    val studentDataStream: DataStream[Student] = senv.addSource(new MysqlSource())    studentDataStream.print()    senv.execute("StreamFromMysqlSource")  }  // 3、创建mysql自定义数据源对象  class MysqlSource extends RichSourceFunction[Student](){    // 3.1 声明Connection对象    var connection:Connection = _    // 3.2 声明PreparedStatement对象    var ps: PreparedStatement = _        // 在 open 方法中进行配置链接信息 drive  url username password    // 加载驱动 Class.forName(),DriveManager 获取链接，调用prepareStatement,预编译执行sql    override def open(parameters: Configuration): Unit = {      val driver: String = "com.mysql.jdbc.Driver"      val url: String = "jdbc:mysql://localhost:3306/blogs"      val username: String = "root"      val password: String = "root"      Class.forName(driver)      connection = DriverManager.getConnection(url,username,password)      val sql: String =        """          |select nid,ntitle,content from notice        """.stripMargin             ps = connection.prepareStatement(sql)          }    // 在run方法中进行查询，结果封装成样例类，ctx进行collect    override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {      // 执行 SQL 查询      val queryResultSet: ResultSet = ps.executeQuery()             while (queryResultSet.next()){                // 分别获取到查询的值        val nid: Int = queryResultSet.getInt("nid")        val ntitle: String = queryResultSet.getString("ntitle")        val content: String = queryResultSet.getString("content")        // 将获取到的值，封装成样例类        val student: Student = Student(nid,ntitle,content)        ctx.collect(student)      }    }    override def close(): Unit = {      if (connection != null){        connection.close()      }      if (ps != null){        ps.close()      }    }    override def cancel(): Unit = {    }  }  case class Student(nid: Int, ntitle: String, content: String) {    override def toString: String = {      "文章id:" + nid + " 标题:" + ntitle + " 内容:" + content }  }}</code>

<code class="javascript">import java.util.Propertiesimport org.apache.flink.api.common.serialization.SimpleStringSchemaimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011/* * @Author: Alice菌 * @Date: 2020/8/10 10:08 * @Description:     sink 到 kafka */object StreamKafkaSink {  def main(args: Array[String]): Unit = {    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 导入隐式转换    import org.apache.flink.api.scala._    val source: DataStream[String] = senv.fromElements("1,小丽,北京,女")    val properties: Properties = new Properties()    properties.setProperty("bootstrap.servers","node01:9092")    val flinkKafkaProducer: FlinkKafkaProducer011[String] = new FlinkKafkaProducer011[String]("test",new SimpleStringSchema(),properties)    source.addSink(flinkKafkaProducer)    // 打印    source.print()    // 执行    senv.execute("StreamKafkaSink")  }}</code>

<code class="javascript">cd /export/servers/kafka_2.11-1.0.0bin/kafka-console-consumer.sh --from-beginning --topic test --zookeeper node01:2181,node02:2181,node03:2181</code>

<code class="javascript">import java.sql.{Connection, DriverManager, PreparedStatement}import org.apache.flink.api.scala._import org.apache.flink.configuration.Configurationimport org.apache.flink.streaming.api.functions.sink.RichSinkFunctionimport org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}/* * @Author: Alice菌 * @Date: 2020/8/10 10:29 * @Description:      */object StreamMysqlSink {  // 定义一个样例类，用于封装数据  case class Student(id:Int,name:String,addr:String,sex:String)  def main(args: Array[String]): Unit = {    // 1、创建执行环境    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment    // 2、准备数据    val studentDStream: DataStream[Student] = senv.fromElements(      Student(4, "小明", "上海", "男"),      Student(5, "小青", "广州", "女"),      Student(6, "小月", "深圳", "女")    )    studentDStream.addSink(new StudentSlinkToMySql)    senv.execute("StreamMysqlSink")  }  class StudentSlinkToMySql extends RichSinkFunction[Student]{    private var connection:Connection = _    private var ps:PreparedStatement = _    override def open(parameters: Configuration): Unit = {      // 设置驱动，连接地址，用户名，密码      var driver: String = "com.mysql.jdbc.Driver"      var url:String = "jdbc:mysql://localhost:3306/blogs?characterEncoding=utf-8&useSSL=false"      var username: String = "root"      var password: String = "root"      // 1、加载驱动      Class.forName(driver)      // 2、创建连接      connection = DriverManager.getConnection(url,username,password)      // 书写SQL语句      val sql: String = "insert into student(id,name,addr,sex) values(?,?,?,?);"      // 3、获得执行语句      ps = connection.prepareStatement(sql)    }    // 关闭连接操作    override def close(): Unit = {      if (connection != null){        connection.close()      }      if (ps != null){        ps.close()      }    }    // 每个元素的插入，都要触发一次 invoke，这里主要进行 invoke 插入    override def invoke(stu: Student): Unit = {      try{        // 4、组装数据，执行插入操作        ps.setInt(1,stu.id)        ps.setString(2,stu.name)        ps.setString(3,stu.addr)        ps.setString(4,stu.sex)        ps.executeUpdate()      } catch {        case e:Exception => println(e.getMessage)      }    }  }}</code>