Flink1.9整合Kafka

来源:https://www.cnblogs.com/tree1123/archive/2019/09/20/11556114.html
-Advertisement-
Play Games

本文基於Flink1.9版本簡述如何連接Kafka。 流式連接器 我們知道可以自己來開發Source 和 Sink ，但是一些比較基本的 Source 和 Sink 已經內置在 Flink 里。 預定義的source支持從文件、目錄、socket，以及 collections 和 iterators ...


file

本文基於Flink1.9版本簡述如何連接Kafka。

流式連接器

file

我們知道可以自己來開發Source 和 Sink ，但是一些比較基本的 Source 和 Sink 已經內置在 Flink 里。

預定義的source支持從文件、目錄、socket，以及 collections 和 iterators 中讀取數據。

預定義的sink支持把數據寫入文件、標準輸出（stdout）、標準錯誤輸出（stderr）和 socket。

連接器可以和多種多樣的第三方系統進行交互。目前支持以下系統:

  • Apache Kafka
  • Apache Cassandra(sink)
  • Amazon Kinesis Streams(source/sink)
  • Elasticsearch(sink)
  • Hadoop FileSystem (sink)
  • RabbitMQ(source/sink)
  • Apache NiFi(source/sink)
  • Twitter Streaming API(source)

請記住，在使用一種連接器時，通常需要額外的第三方組件，比如：數據存儲伺服器或者消息隊列。

Apache Bahir 中定義了其他一些連接器

  • Apache ActiveMQ(source/sink)
  • Apache Flume(sink)
  • Redis(sink)
  • Akka (sink)
  • Netty (source)

使用connector並不是唯一可以使數據進入或者流出Flink的方式。一種常見的模式是從外部資料庫或者 Web 服務查詢數據得到初始數據流，然後通過 Map 或者 FlatMap 對初始數據流進行豐富和增強，這裡要使用Flink的非同步IO。

而向外部存儲推送大量數據時會導致 I/O 瓶頸問題出現。在這種場景下，如果對數據的讀操作遠少於寫操作，可以讓外部應用從 Flink 拉取所需的數據，需要用到Flink的可查詢狀態介面。

本文重點介紹Apache Kafka Connector

Kafka連接器

此連接器提供對Apache Kafka提供的事件流的訪問。

Flink提供特殊的Kafka連接器，用於從/向Kafka主題讀取和寫入數據。Flink Kafka Consumer集成了Flink的檢查點機制，可提供一次性處理語義。為實現這一目標，Flink並不完全依賴Kafka 的消費者組的偏移量，而是在內部跟蹤和檢查這些偏移。

下表為不同版本的kafka與Flink Kafka Consumer的對應關係。

Maven Dependency Supported since Consumer and Producer Class name Kafka version
flink-connector-kafka-0.8_2.11 1.0.0 FlinkKafkaConsumer08 FlinkKafkaProducer08 0.8.x
flink-connector-kafka-0.9_2.11 1.0.0 FlinkKafkaConsumer09 FlinkKafkaProducer09 0.9.x
flink-connector-kafka-0.10_2.11 1.2.0 FlinkKafkaConsumer010 FlinkKafkaProducer010 0.10.x
flink-connector-kafka-0.11_2.11 1.4.0 FlinkKafkaConsumer011 FlinkKafkaProducer011 0.11.x
flink-connector-kafka_2.11 1.7.0 FlinkKafkaConsumer FlinkKafkaProducer >= 1.0.0

而從最新的Flink1.9.0版本開始，使用Kafka 2.2.0客戶端。

下麵簡述使用步驟。

導入maven依賴：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-kafka_2.11</artifactId>
  <version>1.9.0</version>
</dependency>

安裝Kafka：

可以參照 Kafka入門寶典（詳細截圖版）

相容性：

從Flink 1.7開始，它不跟蹤特定的Kafka主要版本。相反，它在Flink發佈時跟蹤最新版本的Kafka。如果您的Kafka代理版本是1.0.0或更高版本，則應使用此Kafka連接器。如果使用舊版本的Kafka（0.11,0.10,0.9或0.8），則應使用與代理版本對應的連接器。

升級Connect要註意Flink升級作業，同時

  • 在整個過程中使用Flink 1.9或更新版本。

  • 不要同時升級Flink和運營商。

  • 確保您作業中使用的Kafka Consumer和/或Kafka Producer分配了唯一標識符（uid）。

  • 使用stop with savepoint功能獲取保存點（例如，使用stop --withSavepoint）。

用法：

引入依賴後，實例化新的source（FlinkKafkaConsumer）和sink（FlinkKafkaProducer）。

Kafka Consumer

先分步驟介紹構建過程，文末附Flink1.9連接Kafka完整代碼。

Kafka consumer 根據版本分別叫做FlinkKafkaConsumer08 FlinkKafkaConsumer09等等
Kafka >= 1.0.0 的版本就叫FlinkKafkaConsumer。

構建FlinkKafkaConsumer

java示例代碼如下：

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181");
properties.setProperty("group.id", "test");
DataStream<String> stream = env
    .addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties));

scala：

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181")
properties.setProperty("group.id", "test")
stream = env
    .addSource(new FlinkKafkaConsumer[String]("topic", new SimpleStringSchema(), properties))
    .print()

必須有的：

1.topic名稱

2.用於反序列化Kafka數據的DeserializationSchema / KafkaDeserializationSchema

3.配置參數：“bootstrap.servers” “group.id” （kafka0.8還需要 “zookeeper.connect”）

配置消費起始位置

java：

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>(...);
myConsumer.setStartFromEarliest();     // start from the earliest record possible
myConsumer.setStartFromLatest();       // start from the latest record
myConsumer.setStartFromTimestamp(...); // start from specified epoch timestamp (milliseconds)
myConsumer.setStartFromGroupOffsets(); // the default behaviour

//指定位置
//Map<KafkaTopicPartition, Long> specificStartOffsets = new HashMap<>();
//specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L);
//myConsumer.setStartFromSpecificOffsets(specificStartOffsets);

DataStream<String> stream = env.addSource(myConsumer);

scala：

val env = StreamExecutionEnvironment.getExecutionEnvironment()

val myConsumer = new FlinkKafkaConsumer[String](...)
myConsumer.setStartFromEarliest()      // start from the earliest record possible
myConsumer.setStartFromLatest()        // start from the latest record
myConsumer.setStartFromTimestamp(...)  // start from specified epoch timestamp (milliseconds)
myConsumer.setStartFromGroupOffsets()  // the default behaviour

//指定位置
//val specificStartOffsets = new java.util.HashMap[KafkaTopicPartition, java.lang.Long]()
//specificStartOffsets.put(new KafkaTopicPartition("myTopic", 0), 23L)
//myConsumer.setStartFromSpecificOffsets(specificStartOffsets)

val stream = env.addSource(myConsumer)
檢查點

啟用Flink的檢查點後，Flink Kafka Consumer將使用主題中的記錄，並以一致的方式定期檢查其所有Kafka偏移以及其他操作的狀態。如果作業失敗，Flink會將流式程式恢復到最新檢查點的狀態，並從存儲在檢查點中的偏移量開始重新使用Kafka的記錄。

如果禁用了檢查點，則Flink Kafka Consumer依賴於內部使用的Kafka客戶端的自動定期偏移提交功能。

如果啟用了檢查點，則Flink Kafka Consumer將在檢查點完成時提交存儲在檢查點狀態中的偏移量。

java

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(5000); // checkpoint every 5000 msecs

scala

val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.enableCheckpointing(5000) // checkpoint every 5000 msecs
分區發現

Flink Kafka Consumer支持發現動態創建的Kafka分區，並使用一次性保證消費它們。

還可以使用正則：

java

final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "test");

FlinkKafkaConsumer011<String> myConsumer = new FlinkKafkaConsumer011<>(
    java.util.regex.Pattern.compile("test-topic-[0-9]"),
    new SimpleStringSchema(),
    properties);

DataStream<String> stream = env.addSource(myConsumer);
...

scala

val env = StreamExecutionEnvironment.getExecutionEnvironment()

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
properties.setProperty("group.id", "test")

val myConsumer = new FlinkKafkaConsumer08[String](
  java.util.regex.Pattern.compile("test-topic-[0-9]"),
  new SimpleStringSchema,
  properties)

val stream = env.addSource(myConsumer)
...
時間戳和水印

在許多情況下，記錄的時間戳（顯式或隱式）嵌入記錄本身。另外，用戶可能想要周期性地或以不規則的方式發出水印。

我們可以定義好Timestamp Extractors / Watermark Emitters，通過以下方式將其傳遞給您的消費者：

java

Properties properties = new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181");
properties.setProperty("group.id", "test");

FlinkKafkaConsumer08<String> myConsumer =
    new FlinkKafkaConsumer08<>("topic", new SimpleStringSchema(), properties);
myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter());

DataStream<String> stream = env
    .addSource(myConsumer)
    .print();

scala

val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092")
// only required for Kafka 0.8
properties.setProperty("zookeeper.connect", "localhost:2181")
properties.setProperty("group.id", "test")

val myConsumer = new FlinkKafkaConsumer08[String]("topic", new SimpleStringSchema(), properties)
myConsumer.assignTimestampsAndWatermarks(new CustomWatermarkEmitter())
stream = env
    .addSource(myConsumer)
    .print()

Kafka Producer

Kafka Producer 根據版本分別叫做FlinkProducer011 FlinkKafkaProducer010等等
Kafka >= 1.0.0 的版本就叫FlinkKafkaProducer 。

構建FlinkKafkaConsumer

java

DataStream<String> stream = ...;

FlinkKafkaProducer011<String> myProducer = new FlinkKafkaProducer011<String>(
        "localhost:9092",            // broker list
        "my-topic",                  // target topic
        new SimpleStringSchema());   // serialization schema

// versions 0.10+ allow attaching the records' event timestamp when writing them to Kafka;
// this method is not available for earlier Kafka versions
myProducer.setWriteTimestampToKafka(true);

stream.addSink(myProducer);

scala

val stream: DataStream[String] = ...

val myProducer = new FlinkKafkaProducer011[String](
        "localhost:9092",         // broker list
        "my-topic",               // target topic
        new SimpleStringSchema)   // serialization schema

// versions 0.10+ allow attaching the records' event timestamp when writing them to Kafka;
// this method is not available for earlier Kafka versions
myProducer.setWriteTimestampToKafka(true)

stream.addSink(myProducer)

需要指定broker list ， topic，序列化類。

自定義分區：預設情況下，將使用FlinkFixedPartitioner將每個Flink Kafka Producer並行子任務映射到單個Kafka分區。

可以實現FlinkKafkaPartitioner類自定義分區。

Flink1.9消費Kafka完整代碼：

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import java.util.Properties;

public class KafkaConsumer {

    public static void main(String[] args) throws Exception {
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "test");
        //構建FlinkKafkaConsumer
        FlinkKafkaConsumer<String> myConsumer = new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties);
        //指定偏移量
        myConsumer.setStartFromEarliest();


        DataStream<String> stream = env
                .addSource(myConsumer);

        env.enableCheckpointing(5000);
        stream.print();

        env.execute("Flink Streaming Java API Skeleton");
    }

項目地址：https://github.com/tree1123/flink_demo_1.9

更多Flink知識，歡迎關註實時流式計算

file


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Mybatis之ResultMap的使用詳解
    Mybatis的定義 MyBatis 是一款優秀的持久層框架，它支持定製化 SQL、存儲過程以及高級映射。MyBatis 避免了幾乎所有的 JDBC 代碼和手動設置參數以及獲取結果集。MyBatis 可 以使用簡單的 XML 或註解來配置和映射原生信息，將介面和 Java 的 POJOs(Plain ...
  • 第十六章、初識資料庫
    [TOC] 第十六章、初識資料庫 一、資料庫 二、資料庫的組成 三、資料庫的分類 四、卸載資料庫 五、安裝資料庫 六、連接資料庫 七、用戶信息查看 八、資料庫的基本操作 九、表的基本操作 十、記錄的基本操作 1）查看某個資料庫中的某個表的所有記錄，如果在對應資料庫中，可以直接查找表 mysql : ...
  • 〈三〉ElasticSearch的認識：搜索、過濾、排序
    ElasticSearch第三篇，關於搜索、過濾、排序簡單講解。 ...
  • CentOS 7.6安裝MySQL 5.7GA版
    centos7 內部集成了mariadb，而安裝mysql的話會和mariadb的文件衝突，所以需要先卸載掉mariadb。 ...
  • win中Oracle簡易客戶端和plsql的配置
    連接資料庫有2種方式：在本機安裝Oracle資料庫或者是安裝一個oracle簡易客戶端當然，簡易客戶端跟oracle資料庫比較少了一些功能連接方式：1）簡易連接sqlplus scott/[email protected]:1521/study註意最後的study是服務名，別搞錯了這種ora-12514... ...
  • 鬆軟科技課堂:sqlserver--數據類型
    SQL Server 數據類型(文章來源：鬆軟科技www.sysoft.net.cn) Character 字元串： Unicode 字元串： Binary 類型： Number 類型： 固定精度和比例的數字。允許從 -10^38 +1 到 10^38 -1 之間的數字。 p 參數指示可以存儲的最大 ...
  • SQL DROP INDEX 語句
    SQL DROP INDEX 語句 我們可以使用 DROP INDEX 命令刪除表格中的索引。 用於 Microsoft SQLJet (以及 Microsoft Access) 的語法: 用於 MS SQL Server 的語法: 用於 IBM DB2 和 Oracle 語法: 用於 MySQL ...
  • MySQL的邏輯查詢語句的執行順序
    一 SELECT語句關鍵字的定義順序 二 SELECT語句關鍵字的執行順序 三 準備表和數據 \1. 新建一個測試資料庫TestDB； 2.創建測試表table1和table2； 3.插入測試數據； 準備工作做完以後，table1和table2看起來應該像下麵這樣 四 準備SQL邏輯查詢測試語句 + ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類