快速入門Flink(7)——Flink中的流式處理DataSource與DataSinke-有解無憂

在這里插入圖片描述
前幾篇文章給大家講解了個關于Flink批處理相關的技術點，今天給大家將講解下關于流式處理的DataSource與DataSink ?（Flink專輯）

一、入門案例

使用Flink的流式處理來計算wordCount
實作步驟：

獲取Flink批處理運行環境
構建一個socket源
使用Flink操作進行單詞統計
列印

說明：如果 linux 上沒有安裝 nc 服務 ,使用 yum 安裝

yum install -y nc

參考代碼:

import org.apache.flink.streaming.api.scala._
/**
 * @author 流處理wordCount
 * @date 2020/8/26 22:03
 * @version 1.0
 */
 object StreamWordCount{
  def main(args: Array[String]): Unit = {
    //1.構建流處理的運行環境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用socket來接收資料
    val socketData: DataStream[String] = env.socketTextStream("node01", 9999)
    //3.對資料進行切分將每個單詞獲取出來后面加1 使用keyBy進行分組使用sum進行求核
    val result = socketData.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)
    //4.將結果輸出并啟動
    result.print("批處理wordCount")
    env.execute("批處理wordCount")
  }
}

二、Flink 在流處理上常見的 Source

注意：Flink 在流處理上常見的 Source ，Flink 在流處理上的 source 和在批處理上的 source 基本一致，

2.1 基本地集合的source

我在這就不給大家一一介紹了，我在這里給大家入門，想學習更多關于本地的Data Source請看?DataSource

import org.apache.flink.streaming.api.scala._
/**
 * @author
 * @date 2020/9/20 18:57
 * @version 1.0
 */
object StreamDataSource {
  def main(args: Array[String]): Unit = {
   //1.構建流處理運行環境
   val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用fromCollection 構建資料集
    val data = env.fromCollection(List("張三", "李四", "王五"))
   //3.輸出
    data.print()
    env.execute("StreamDataSource")
  }
}

2.2 基本地檔案的Source


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

/**
 * @author
 * @date 2020/9/20 19:05
 * @version 1.0
 */
object StreamFileSource {
  def main(args: Array[String]): Unit = {
    //1.構建流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用檔案構建資料集
    val dataSource = env.readTextFile("./data/wordcount.txt")
    //3.列印
    dataSource.print()
    env.execute("StreamFileSource")
  }
}

2.3 自定義Source

除了預定義的 Source 外，我們還可以通過實作 SourceFunction 來自定義 Source，然后通過 StreamExecutionEnvironment.addSource(sourceFunction)添加進來，比如讀取 Kafka 資料的 Source： addSource(new FlinkKafkaConsumer08<>); 我們可以實作以下三個介面來自定義 Source：

2.3.1 SourceFunction：創建非并行資料源

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._
/**
 * @author
 * @date 2020/9/20 19:22
 * @version 1.0
 */
object StreamCustomerNoParallelSource {
  def main(args: Array[String]): Unit = {
    //1.構建流式處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用自定義資料流
    val dataSource = env.addSource(new MyNoParallel()).setParallelism(1)
    //3.列印
    dataSource.print()
    //4.執行程式
    env.execute("StreamCustomerNoParallelSource")
  }
  class MyNoParallel() extends SourceFunction[Long] {
    // 定義一個變數
    var number: Long = 1L
    var isRunning: Boolean = true
    override def run(sourceContext: SourceFunction.SourceContext[Long]): Unit = {
      while (isRunning) {
        sourceContext.collect(number)
        number += 1
        Thread.sleep(1000)
        if (number == 10) {
          cancel()
        }
      }
    }
    override def cancel(): Unit = {
      isRunning = false
    }
  }
}

2.3.2 ParallelSourceFunction：創建并行資料源，

import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

/**
 * @author
 * @date 2020/9/20 20:40
 * @version 1.0
 */
object StreamCustomerParallelSource {
  def main(args: Array[String]): Unit = {
    //1.構建流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.基于自定義ParallelSource資料源創建并行的資料
    val source = env.addSource(new MyParallelSource()).setParallelism(1)
    //3.列印輸出
    source.print()
    //4.執行任務
    env.execute("StreamCustomerParallelSource")
  }

  class MyParallelSource() extends ParallelSourceFunction[Long] {
    //1.定義一個Long型別的變數
    var number: Long = 1L
    //2.定義一個變數
    var isRunning: Boolean = true

    override def run(sourceContext: SourceFunction.SourceContext[Long]): Unit = {
      while (isRunning) {
        sourceContext.collect(number)
        number += 1
        if (number > 20) {
          cancel()
        }
      }
    }

    override def cancel(): Unit = {
      isRunning = false
    }
  }
}

2.3.3 RichParallelSourceFunction：創建并行資料源，

import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

/**
 * @author
 * @date 2020/9/20 20:48
 * @version 1.0
 */
object StreamCustomerRichParallelSource {
  def main(args: Array[String]): Unit = {
    //1.構建流式處理資料集
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.基于RichParallelSource并行資料源構建資料集
    val dataSource = env.addSource(new RichParallelSource()).setParallelism(2)
    dataSource.map(line=>{
      println("接收到的資料：" + line)
      line
    })
    env.execute("StreamCustomerRichParallelSource")
  }
  class RichParallelSource() extends RichParallelSourceFunction[Long]{
    //1.定義一個Long型別的變數
    var number: Long = 1L
    //2.定義一個變數
    var isRunning: Boolean = true
    override def run(sourceContext: SourceFunction.SourceContext[Long]): Unit = {
      while (isRunning) {
        sourceContext.collect(number)
        number += 1
        Thread.sleep(1000)
        if (number > 20) {
          cancel()
        }
      }
    }
    override def cancel(): Unit = {
      isRunning=false
    }
  }
}

2.3.4 基于 kafka 的 source 操作

在這里我就不過多講解了關于Kafka的常用的命令，如果想學的可以點擊-> kfka常用的操作

代碼示例：

import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011

/**
 * @author 消費kafka中的資料
 * @date 2020/9/21 22:53
 * @version 1.0
 */
object StreamKafkaSource {
  def main(args: Array[String]): Unit = {
    //1.構建流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.指定消費主題
    var topic = "FlinkAsKafka"
    //2.1設定配置資訊
    val porps = new Properties()
    porps.setProperty("bootstrap.servers", "node01:9092,node02:9092,node03:9092")
    porps.setProperty("group.id", "test01")
    porps.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    porps.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    //3.基于Flink構建kafka消費者
    val kafka = new FlinkKafkaConsumer011[String](topic, new SimpleStringSchema(), porps)
    //4.設定Flink層最新的資料開始消費
    kafka.setStartFromLatest()
    //5.基于kafka構建資料源
    val data = env.addSource(kafka)
    //6.列印輸出
    data.print()
    env.execute("StreamKafkaSource")
  }
}

2.3.5 基于 mysql 的 source 操作

上面就是 Flink 自帶的 Kafka source，那么接下來就模仿著寫一個從 MySQL 中讀取資料的 Source，


import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._

/**
 * @author 基于MySQL的source操作
 * @date 2020/9/21 23:17
 * @version 1.0
 */
object StreamFromMysqlSource {

  case class User(id: String, user_id: String, user_name: String, phone: String, lan_id: String, region_id: String)


  def main(args: Array[String]): Unit = {
    //1.創建流式執行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.添加自定義mysql資料源
    val source = env.addSource(new MySqlSource())
    //3.輸出
    source.print()
    //4.任務執行
    env.execute("StreamFromMysqlSource")
  }

  class MySqlSource() extends RichSourceFunction[User] {
    //1.宣告Connection物件
    var connection: Connection = null
    //2.宣告 PreparedStatement 物件
    var ps: PreparedStatement = null

    override def open(parameters: Configuration): Unit = {
      var driver = "com.mysql.jdbc.Driver"
      var url = "jdbc:mysql://node01:3306/datax_web"
      var username = "root"
      var password = "123456"
      Class.forName(driver)
      connection = DriverManager.getConnection(url, username, password)
      var sql =
        """
          |SELECT id,user_id,user_name,phone,lan_id,region_id
          |FROM user
          |""".stripMargin
      ps = connection.prepareStatement(sql)
    }

    override def run(sourceContext: SourceFunction.SourceContext[User]): Unit = {
      val queryResultSet = ps.executeQuery()
      while (queryResultSet.next()) {
        val id = queryResultSet.getString("id")
        val user_id = queryResultSet.getString("user_id")
        val user_name = queryResultSet.getString("user_name")
        val phone = queryResultSet.getString("phone")
        val lan_id = queryResultSet.getString("lan_id")
        val region_id = queryResultSet.getString("region_id")
        sourceContext.collect(User(id, user_id, user_name, phone, lan_id, region_id))
      }
    }

    override def cancel(): Unit = {
      if (connection != null) {
        connection.close()
      }
      if (ps != null) {
        ps.close()
      }
    }
  }

}

三、Flink 常用的DataSink

3.1 將資料 sink 到本地檔案

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

/**
 * @author
 * @date 2020/9/22 22:41
 * @version 1.0
 */
object StreamFileSourceSinkFile {
  def main(args: Array[String]): Unit = {
    //1.構建流式處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.讀取本地檔案構建資料集
    val data = env.readTextFile("./data/wordcount.txt")
    //3.檔案輸出
    data.writeAsText("./data/wordcountSink.txt").setParallelism(1)
    //4.開始執行
    env.execute("StreamFileSourceSinkFile")
  }
}

3.2 Sink 到本地集合


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

/**
 * @author
 * @date 2020/9/22 22:50
 * @version 1.0
 */
object StreamFromCollectionSourceFile {
  def main(args: Array[String]): Unit = {
    //1.構建流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.使用FromCollection構建資料集
    val data = env.fromCollection(List((1, "張三"), (2, "李四"), (1, "趙劉")))
    //3.將檔案輸出
    data.writeAsText("./data/fromCollection.txt").setParallelism(1)
    //4.執行任務
    env.execute("StreamFromCollectionSourceFile")
  }
}

3.3 Sink將資料到 HDFS


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

/**
 * @author
 * @date 2020/9/22 22:41
 * @version 1.0
 */
object StreamFileSourceSinkFileHDFS {
  def main(args: Array[String]): Unit = {
    //1.構建流式處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.讀取本地檔案構建資料集
    val data = env.readTextFile("./data/wordcount.txt")
    //3.檔案輸出
    data.writeAsText("hdfs://node01:8020/data/wordcountSink.txt").setParallelism(1)
    //4.開始執行
    env.execute("StreamFileSourceSinkFile")
  }
}

3.4 Sink將資料到 Kafka

import java.util.Properties
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.kafka.{FlinkKafkaConsumer011, FlinkKafkaProducer011}

/**
 * @author
 * @date 2020/9/22 23:17
 * @version 1.0
 */
object StreamKafkaSink {
  def main(args: Array[String]): Unit = {
    //1.構建流處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.構建資料集
    val dataSource: DataStream[String] = env.fromElements("1,小麗,北京,女")
    //3.構建組態檔
    val prop = new Properties()
    prop.setProperty("bootstrap.servers", "node01:9092,node02:9092,node03:9092")
    //4.連接Kafka
    val producer: FlinkKafkaProducer011[String] = new FlinkKafkaProducer011[String]("FlinkAsKafka", new SimpleStringSchema(), prop)
    //5.將資料打入kafka
    dataSource.addSink(producer)
    //6.執行任務
    env.execute("StreamKafkaSink")
  }
}

3.4 Sink將資料到 MySQL

import java.sql.{Connection, DriverManager, PreparedStatement}
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

/**
 * @author
 * @date 2020/9/22 23:35
 * @version 1.0
 */
object StreamMysqlSink {

  case class Student(id: Int, name: String, addr: String, sex: String)

  def main(args: Array[String]): Unit = {
    //1.構建流式處理運行環境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //2.資料準備
    val dataSource: DataStream[Student] = env.fromElements(
      Student(1, "張三", "上海", "男"),
      Student(2, "李四", "北京", "女"),
      Student(3, "王五", "上海", "男"),
      Student(4, "趙劉", "廣東", "男")
    )
    dataSource.addSink(new StudentSinkToMysql)
    env.execute("StreamMysqlSink")
  }

  class StudentSinkToMysql extends RichSinkFunction[Student] {
    var connection: Connection = null
    var ps: PreparedStatement = null

    override def open(parameters: Configuration): Unit = {
      var driver = "com.mysql.jdbc.Driver"
      var url = "jdbc:mysql://node01:3306/text?characterEncoding=utf-8&useSSL=false"
      var username = "root"
      var password = "123456"
      //加載驅動
      Class.forName(driver)
      //創建連接
      connection = DriverManager.getConnection(url,username,password)
      ps = connection.prepareStatement("insert into student(id,name,addr,sex) values (?,?,?,?);")
    }

    override def close(): Unit = {
      if (connection != null) {
        connection.close()
      }
      if (ps != null) {
        ps.close()
      }
    }

    override def invoke(value: Student): Unit = {
      ps.setInt(1,value.id)
      ps.setString(2,value.name)
      ps.setString(3,value.addr)
      ps.setString(4,value.sex)
      ps.executeUpdate()
    }
  }
}

轉載請註明出處，本文鏈接：https://www.uj5u.com/qita/135638.html

標籤：其他

上一篇：4面京東拿下offer（Java后臺研發崗）：MySQL+jvm+Redis+多執行緒+網路協議

下一篇：聽說阿里中間件Java面試很難？看看大佬整理的經驗吧

快速入門Flink(7)——Flink中的流式處理DataSource與DataSinke