2. 程式人生 > >Flink的流處理API(二)

Flink的流處理API(二)

阿新 • • 發佈：2020-08-03

一、Environment

1，getExecutionEnvironment

　　getExecutionEnvironment會根據查詢執行的方式決定返回什麼樣的執行環境，是最常用的一種建立執行環境的方式。

2，createLocalEnvironment

　　返回本地執行環境，需要在呼叫時指定預設的並行度。

val env = StreamExecutionEnvironment.createLocalEnvironment(1) //parallelism

3，createRemoteEnvironment

　　返回叢集執行環境，將Jar提交到遠端伺服器。需要在呼叫時指定JobManager的IP和埠號，並指定要在叢集中執行的Jar包。

//hostname port jarFiles
val env = ExecutionEnvironment.createRemoteEnvironment(host, port,"/flink/wc.jar")

4，maven依賴

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-scala_2.11</artifactId>
        <version>1.7.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-streaming-scala_2.11</artifactId>
        <version>1.7.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-kafka-0.11_2.12</artifactId>
        <version>1.7.0</version>
    </dependency>
</dependencies>
<build>
    <plugins>
        <!-- 該外掛用於將Scala程式碼編譯成class檔案 -->
        <plugin>
            <groupId>net.alchim31.maven</groupId>
            <artifactId>scala-maven-plugin</artifactId>
            <version>3.4.6</version>
            <executions>
                <execution>
                    <!-- 宣告繫結到maven的compile階段 -->
                    <goals>
                        <goal>compile</goal>
                        <goal>testCompile</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-assembly-plugin</artifactId>
            <version>3.0.0</version>
            <configuration>
                <descriptorRefs>
                    <descriptorRef>jar-with-dependencies</descriptorRef>
                </descriptorRefs>
            </configuration>
            <executions>
                <execution>
                    <id>make-assembly</id>
                    <phase>package</phase>
                    <goals>
                        <goal>single</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

 

二、Source

1，基本讀取方式

//檔案中讀取
val fileDs = env.readTextFile("in/tbStock.txt")
//埠讀取
val socketDs = env.socketTextStream("localhost",777)
//集合中獲取
val collectDs = env.fromCollection(List("aaa","bbb","ccc","aaa"))

2，kafka source

//kafka配置檔案
val properties = new Properties()
properties.setProperty("bootstrap.servers", "hadoop102:9092")
properties.setProperty("group.id", "consumer-group")
properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
properties.setProperty("auto.offset.reset", "latest")
//接收kafka的topic-demo這個topic發來的資料
val kafkaDataStream: DataStream[String] = env.addSource(new FlinkKafkaConsumer011[String]("topic-demo", new SimpleStringSchema(), properties))

3，Flink Kafa如何實現exactly-once

可參考: https://www.aboutyun.com/forum.php?mod=viewthread&tid=27395

　　Flink通過checkpoint來儲存資料是否處理完成的狀態

　　由JobManager協調各個TaskManager進行checkpoint儲存，checkpoint儲存在 StateBackend中，預設StateBackend是記憶體級的，也可以改為檔案級的進行持久化儲存。

　　執行過程實際上是一個兩段式提交，每個運算元執行完成，會進行“預提交”，直到執行完sink操作，會發起“確認提交”，如果執行失敗，預提交會放棄掉。

　　如果宕機需要通過StateBackend進行恢復，只能恢復所有確認提交的操作。

   

 

4，自定義source

env.addSource(new MySource)
//自定義source
class MySource extends SourceFunction[(String,Double)] {
  //flag: 表示資料來源是否還在正常執行
  var running: Boolean = true
  override def cancel(): Unit = {
    running = false
  }
  override def run(ctx: SourceFunction.SourceContext[(String,Double)]): Unit = {
    //初始化一個隨機數發生器
    val rand = new Random()
    var curTemp = 1.to(10).map(
      i => ("item_" + i, 65 + rand.nextGaussian() * 20)
    )
    while (running) {
      curTemp.foreach(
        t => ctx.collect(t)
      )
      Thread.sleep(5000)  //每5秒鐘產生一組資料
    }
  }
}

 

三、Transform

1，基本轉換運算元

//map
val streamMap = stream.map { x => x * 2 }
//flatmap
val streamFlatMap = stream.flatMap{
    x => x.split(" ")
}
//filter
val streamFilter = stream.filter{
    x => x == 1
}

2，KeyBy與Reduce

　　keyBy(DataStream → KeyedStream)：輸入必須是Tuple型別，邏輯地將一個流拆分成不相交的分割槽，每個分割槽包含具有相同key的元素，在內部以hash的形式實現的。

　　reduce(KeyedStream → DataStream)：一個分組資料流的聚合操作，合併當前的元素和上次聚合的結果，產生一個新的值，返回的流中包含每一次聚合的結果，而不是隻返回最後一次聚合的最終結果。

val keyedStream: KeyedStream[(String, Int), Tuple] = startUplogDstream.map(startuplog=>(startuplog.ch,1)).keyBy(0)
//reduce //sum
keyedStream.reduce{  (ch1,ch2)=>
  (ch1._1,ch1._2+ch2._2)
}.print()

3，Split和Select

　　split(DataStream → SplitStream)：根據某些特徵把一個DataStream拆分成兩個或者多個DataStream。

　　select(SplitStream→DataStream)：從一個SplitStream中獲取一個或者多個DataStream。

//根據Item的id進行拆分
val splitStream:SplitStream[Item] = dStream.split {
  item =>
    List(item.id)
}
//獲取標記為item_1的資料集
splitStream.select("item_1").print()

4，Connect和CoMap

　　connect(DataStream,DataStream → ConnectedStreams)：連線兩個保持他們型別的資料流，兩個資料流被Connect之後，只是被放在了一個同一個流中，內部依然保持各自的資料和形式不發生任何變化，兩個流相互獨立。

　　CoMap,CoFlatMap(ConnectedStreams → DataStream)：作用於ConnectedStreams上，功能與map和flatMap一樣，對ConnectedStreams中的每一個Stream分別進行map和flatMap處理。

val connStream: ConnectedStreams[StartUpLog, StartUpLog] = appStoreStream.connect(otherStream)
val allStream: DataStream[String] = connStream.map(
  (log1: StartUpLog) => log1.ch,
  (log2: StartUpLog) => log2.ch
)

4，Union

　　DataStream → DataStream：對兩個或者兩個以上的DataStream進行union操作，產生一個包含所有DataStream元素的新DataStream。注意:如果你將一個DataStream跟它自己做union操作，在新的DataStream中，你將看到每一個元素都出現兩次。

val unionStream: DataStream[StartUpLog] = appStoreStream.union(otherStream)
unionStream.print("union:::")

5，Connect與 Union 區別：

　　1)Union之前兩個流的型別必須是一樣，Connect可以不一樣，在之後的coMap中再去調整成為一樣的。

　　2)Connect只能操作兩個流，Union可以操作多個

 

四、實現UDF函式

1，函式類(Function Classes)

　　Flink暴露了所有udf函式的介面(實現方式為介面或者抽象類)。例如:MapFunction, FilterFunction, ProcessFunction 等等。

val flinkTweets = tweets.filter(new FlinkFilter)
//自定義filter類
class FlinkFilter extends FilterFunction[String] {
    override def filter(value: String): Boolean = { value.contains("flink")
   }
}

2，匿名函式(Lamda Functions)

val flinkTweets = tweets.filter(_.contains("flink"))

3，富含數(Rich Functions)

　　富函式是 DataStream API 提供的一個函式類的介面，所有 Flink 函式類都有其 Rich 版本。它與常規函式的不同在於，可以獲取執行環境的上下文，並擁有一些生命週期方法，所以可以實現更復雜的功能。

　　open()方法是 rich function 的初始化方法，當一個運算元例如map或者filter被呼叫之前open()會被呼叫。

　　close()方法是生命週期中的最後一個呼叫的方法，做一些清理工作。

　　getRuntimeContext()方法提供了函式的 RuntimeContext 的一些資訊，例如函式執行的並行度，任務的名字，以及 state 狀態。

 

五、Sink

　　Flink 沒有類似於spark中foreach方法，讓使用者進行迭代的操作。雖有對外的輸出操作都要利用Sink完成。最後通過類似如下方式完成整個任務最終輸出操作。

1，kafka

dstream.addSink(new FlinkKafkaProducer011[String]("linux01:9092","test", new SimpleStringSchema()))

2，redis

<dependency>
    <groupId>org.apache.bahir</groupId>
    <artifactId>flink-connector-redis_2.11</artifactId>
    <version>1.0</version>
</dependency>
val config = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").setPort(6379).build()
resultDStream.addSink(new RedisSink[Item](config,new MyRedisMapper))
//定義redisMapper
class MyRedisMapper extends RedisMapper[Item] {
  override def getCommandDescription: RedisCommandDescription = {
    new RedisCommandDescription(RedisCommand.HSET,"item_test") //hkey
  }
  override def getKeyFromData(data: Item): String = data.id 
  override def getValueFromData(data: Item): String = data.toString
}

3，Elasticsearch

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-elasticsearch6_2.11</artifactId>
    <version>1.7.2</version>
</dependency>
//定義es的host集合
val list = new util.ArrayList[HttpHost]()
list.add(new HttpHost("linux01", 9200))
//定義esBuilder
val esBuilder = new ElasticsearchSink.Builder[Item](list,new ElasticsearchSinkFunction[Item] {
  override def process(element: Item, ctx: RuntimeContext, indexer: RequestIndexer): Unit = {
    //定義es資料儲存方式和儲存值
    val json = new util.HashMap[String, String]()
    json.put("data", element.toString)
    //定義儲存索引 type 和資料來源
    val indexRequest = Requests.indexRequest().index("indexName").`type`("_doc").source(json)
    indexer.add(indexRequest)
  }
})
resultDStream.addSink(esBuilder.build())

4，自定義sink(JDBC)

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>5.1.44</version>
</dependency>
resultDStream.addSink(new MyJDBCSink)
//自定義jdbcsink
class MyJDBCSink extends RichSinkFunction[Sensor]{
  var conn: Connection = _
  var insertStmt: PreparedStatement = _
  var updateStmt: PreparedStatement = _
  //open 簡歷連線
  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test", "root", "123456")
    insertStmt = conn.prepareStatement("INSERT INTO item_test (id, num) VALUES (?, ?)")
    updateStmt = conn.prepareStatement("UPDATE item_test SET num = ? WHERE id = ?")
  }
  //呼叫執行
  override def invoke(value: Sensor, context: SinkFunction.Context[_]): Unit = {
    updateStmt.setDouble(1, value.temp)
    updateStmt.setString(2, value.id)
    updateStmt.execute()
    if (updateStmt.getUpdateCount == 0) {
      insertStmt.setString(1, value.id)
      insertStmt.setDouble(2, value.temp)
      insertStmt.execute()
    }
  }
  //關閉資源
  override def close(): Unit = {
    insertStmt.close()
    updateStmt.close()
    conn.close()
  }
}

&n

相關推薦

Flink的流處理API(二)

一、Environment 1，getExecutionEnvironment 　　getExecutionEnvironment會根據查詢執行的方式決定返回什麼樣的執行環境，是最常用的一種建立執行環境的方式。 2，createLocalEnvironment 　　返回本地執行環境，需要在呼叫時指定預設的並行

雲星資料---Apache Flink實戰系列(精品版)】：Flink流處理API詳解與程式設計實戰002-Flink基於流的wordcount示例002

三、基於socket的wordcount 1.傳送資料 1.傳送資料命令 nc -lk 9999 2.傳送資料內容 good good study day day

Flink流處理（二）- 時間語義

嚴重 結果 rep 通知 ble eager 自動 行處理 應該 4. 時間語義（Time Semantics） 這章我們會介紹時間語義，以及在流中，對於時間的各種不同的概念的描述。同時我們也會討論一個流處理器在事件亂序的情況下，如何能提供精準的結果，以及如何使用流對歷史

Apache Flink流處理(二)

到目前為止，您已經瞭解了流處理如何解決傳統批處理的侷限性，以及它如何支援新的應用程式和體系結構。您已經熟悉了開源的流處理空間的演變，並對Flink流應用程式有了簡單的瞭解。在這一章，你將進入流世界中，並得到本書本書剩下部分所必要的基礎知識。 這一章仍然與Flink無關。它的目標是介紹流處

Apache Flink流處理(一)

Apache Flink是一個分散式流處理器，它使用直接且富有表現力的API來實現有狀態的流處理應用程式。它以容錯的方式高效地大規模執行此類應用程式。Flink於2014年4月加入Apache軟體基金會作為孵化專案，並於2015年1月成為頂級專案。從一開始，Flink就有一個非常活躍且不斷增

Flink視頻教程_基於Flink流處理的動態實時電商實時分析系統

分布 業務 電商分析 apr 進行 處理 密碼 教程 包括 Flink視頻教程_基於Flink流處理的動態實時電商實時分析系統 課程分享地址鏈接:https://pan.baidu.com/s/1cX7O-45y6yUPT4B-ACfliA 密碼:jqmk 在開始學習前給

flink流處理內容

Flink核心是一個流式的資料流執行引擎,其針對資料流的分散式計算提供了資料分佈、資料通訊以及容錯機制等功能 Flink提供了諸多更高抽象層的API以便使用者編寫分散式任務: DataSet API, 對靜態資料進行批處理操作,將靜態資料抽象成分散式的資料集,使用者可以方便地使用Flink提供的各種操作符

Flink流處理過程的部分原理分析

文章目錄 前言 流的時間有序性保證 視窗序列對齊 流資料的容錯：Checkpoint機制 Barrier State 引用 前言 在分散式領域，計算和儲存一直是兩大子領域。很多分散式

回顧2016--Apache Flink流處理在生產中的實踐

從2016年4月底開始接觸Flink，到現在已經8個多月了。從瞭解到熟悉，再到實際開發，這個過程就是我從0到實際開發使用Flink的過程。 上週，我們的Flink流計算程式終於上線了。也算是在實時流計算方面的一個成果。 下面，我將簡要介紹下公司如何使用Fli

Flink流處理的時間視窗

Flink流處理的時間視窗 對於流處理系統來說，流入的訊息是無限的，所以對於聚合或是連線等操作，流處理系統需要對流入的訊息進行分段，然後基於每一段資料進行聚合或是連線等操作。 訊息的分段即稱為視窗，流處理系統支援的視窗有很多型別，最常見的就是時間視窗，基於時間間隔對訊息進行分段處理。本節主要介紹Flink

Flink流處理的時間窗口

size 切分 系統 data 種類 提前 當前時間 應用 data- Flink流處理的時間窗口 對於流處理系統來說，流入的消息是無限的，所以對於聚合或是連接等操作，流處理系統需要對流入的消息進行分段，然後基於每一段數據進行聚合或是連接等操作。 消息的分段即稱為窗口，流處

基於Flink流處理的動態實時電商實時分析系統

在開始學習前給大家說下什麼是Flink？ 1.Flink是一個針對流資料和批資料的分散式處理引擎，主要用Java程式碼實現。 2.Apache Flink作為Apache的頂級專案，Flink集眾多優點於一身，包括快速、可靠可擴充套件、完全相容Hadoop、使

Flink流處理的動態實時億級全端使用者畫像系統視訊課程分享

基於Flink流處理的動態實時億級全端使用者畫像系統課程下載: https://pan.baidu.com/s/1YtMs-XG5-PsTFV9_7-AlfA 提取碼: 639m 專案中採用到的演算法包含Logistic Regression、Kmeans、TF-IDF等，Flink暫時支援的演算法比較少，

Flink流處理的動態實時億級全端用戶畫像系統視頻課程分享

分享 gis 架構師 log tps 終端 下載 hadoop2.6 算法比較 基於Flink流處理的動態實時億級全端用戶畫像系統課程下載: https://pan.baidu.com/s/1YtMs-XG5-PsTFV9_7-AlfA 提取碼: 639m 項目中采用到的算

Flink流處理之迭代案例

當前Flink將迭代的重心集中在批處理上，之前我們談及了批量迭代和增量迭代主要是針對批處理（DataSet）API而言的，並且Flink為批處理中的迭代提供了針對性的優化。但是對於流處理（DataStream），Flink同樣提供了對迭代的支援，這一節我們主要來

基於Flink流處理的動態實時億級全端用戶畫像系統

發展 完美 企業應用 應用 消費 spring 服務架構 習慣 link 基於Flink流處理的動態實時億級全端用戶畫像系統網盤地址：https://pan.baidu.com/s/1t_VNw9I6ML1o-jSqVlhBJQ 提取碼: 3t5q備用地址（騰訊微雲）：ht

Apache Flink 流處理例項

維基百科在 IRC 頻道上記錄 Wiki 被修改的日誌，我們可以通過監聽這個 IRC 頻道，來實時監控給定時間視窗內的修改事件。Apache Flink 作為流計算引擎，非常適合處理流資料，並且，類似於 Hadoop MapReduce 等框架，Flink 提供了非常良好的抽象，使得業務邏輯程式碼編寫非常簡單

Flink流處理（一）- Flink 簡介

query 架構 time conn esc bubuko 一次 upd tor 1. Flink 簡介 Flink 是一個分布式流處理器，提供直觀且易於使用的API，以供實現有狀態的流處理應用。它能夠以fault-tolerant的方式高效地運行在大規模系統中。 流處

再也不擔心寫出臃腫的Flink流處理程式啦，發現一款將Flink與Spring生態完美融合的腳手架工程-懶松鼠Flink-Boot

[TOC] 還在為開發Flink流處理應用程式時無法像開發Spring Boot程式那麼優雅的分層以及裝配Bean而煩惱嗎？ ## 你可能面臨如下苦惱： 1. 開發的Flink流處理應用程式，業務邏輯全部寫在Flink的操作符中，程式碼無法服用，無法分層 2. 要是有一天它可以像開發Spring Boo

（二）：Flink概述，Flink如何支援批流處理，程式流程

前言 以下都儘量對比Spark（或者大資料生態的其他技術）進行理解。 Flink簡介，Flink能做什麼 Flink簡介 Flink最初是一個名為Stratosphere的研究專案，目標是為柏林地區的一些大學建立下一代大資料分析平臺。 它於2014年4月16