spark系列-結構化數據流-集成kafka數據源

IT那活兒發布于2023-01-11 13:20 / 2493人閱讀

點擊上方“IT那活兒”，關注后了解更多內容，不管IT什么活兒，干就完了！！！

步驟

兼容的kafka版本是0.10.0及以上。

1. 添加依賴

2. 創建kafka數據流

3. 創建kafka批處理數據集

4. 數據流每行包含的內容

選項配置

1. 必選項

對于kafka源，對于批處理，流處理以下選項是必選：

2. 可選項

消費者池

初始化消費者非常耗時，尤其在處理時間是關鍵因素的流場景中，所以spark利用ApacheCommons將kafka消費者集中在executors上。

消費者緩存關鍵是topic name、topic partition、group id。

以下選項可以配置消費者池（consumer pool）：

提取數據時數據池的配置選項：

向kafka寫入消息

向kafka寫入消息時不可避免會出現重復數據，一個可能解決方案是引入一個主鍵，再讀取時可以執行重復數據消除。

向kafka寫入消息的schema應該包括以下列：

注意：如果配置中未指定topic那么topic也是必須的。

如上只有value列是必須的，如果沒有指定key則默認是null，如果指定了topic則將寫入指定的topic，除非配置選項中指定了topic，否則必須指定topic列，配置選項中指定的topic優先級高，如果未指定partition列，則由生產者進行計算處理，可以通過設置kafka.partitioner.class選項指定分區器，如果未指定，則使用默認分區器。

以下選項對于kafka sink不管是批查詢還是流查詢都是必須的：

以下是可選選項：

基于流查詢創建kafka sink：

//選項中指定topic。

基于批查詢創建kafka sink：

//選項中指定topic。

//指定topic列。

生產者池

有消費者池，也有生產者池，生產者池配置：

Kafka特定配置

消費者配置選項。

生產者配置選項。

安全

0.9.0.0版本kafka引入了一些提高kafka集群安全性的選項，值得注意的是安全性是可選的，默認是關閉的，spark支持以下針對kafka集群認證方法：

委托令牌（Delegation token），通過這種方式，可以通過spark參數配置應用程序，并且可能不需要JAAS登錄配置，關于委托令牌可以通過以下鏈接：

http://kafka.apache.org/documentation/#security_delegation_token

該過程使用委托令牌初始化，當設置了spark.kafka.clusters.${cluster}.auth..bootstrap.servers，spark按照優先級考慮一下登錄選項。

JAAS登錄配置，通過例子進行說明：

1）keytab file

2）Kerberos 憑據緩存

委托令牌（delegation token）可以通過設置spark.security.credentials.kafka.enabled為false（默認還是true）。

Spark可以配置為使用以下身份驗證協議來獲取令牌（它必須與Kafka broker配置匹配）：

SASL SSL（默認協議），SSL，SASL PLAINTEXT（用于測試）。

獲取令牌成功之后，spark向其他節點進行分發，令牌使用SCRAM 登錄模塊進行登錄認證，因此必須進行相應的配置（spark.kafka.clusters.${cluster}.sasl.token.mechanism（默認值是SCRAM-SHA-512）），該參數必須與kafka broker配置參數匹配。

當令牌在executor上可用時，spark會按優先順序考慮以下登錄選項：

JAAS login configuration，

委托令牌，有關詳細信息，請參閱spark.kafka.clusters.${cluster}.target.bootstrap.servers.regex參數。

如果上述任何一項都不適用，則假定為不安全連接。

配置選項：

JAAS登錄配置必須放置在Spark嘗試訪問Kafka群集的所有節點上。這提供了應用任何自定義身份驗證邏輯的可能性，維護成本更高。這可以通過幾種方式實現。一種可能性是提供額外的JVM參數，例如：

本文作者：潘宗昊

本文來源：IT那活兒（上海新炬王翦團隊）

云服務器 GPU云服務器 spark入門實戰系列云數據服務器系列數據分析與數據分析系列教材 kafka數據遷移

文章版權歸作者所有，未經允許請勿轉載,若此文章存在違規行為，您可以聯系管理員刪除。

轉載請注明本文地址：http://specialneedsforspecialkids.com/yun/129583.html

大數據入門指南（GitHub開源項目）

摘要：項目地址前言大數據技術棧思維導圖大數據常用軟件安裝指南一分布式文件存儲系統分布式計算框架集群資源管理器單機偽集群環境搭建集群環境搭建常用命令的使用基于搭建高可用集群二簡介及核心概念環境下的安裝部署和命令行的基本使用常用操作分區表和分桶表視圖項目GitHub地址：https://github.com/heibaiying... 前言大數據技術棧思維導圖大數據常用軟件安裝指...

guyan0319 2019-08-19 11:46 評論0 收藏0
大數據框架hadoop服務角色介紹

摘要：大數據框架服務角色介紹翻了一下最近一段時間寫的分享，發行版本下載安裝運行環境部署等相關內容幾乎都已經寫了一遍了。這些數據通常是由于吞吐量的要求而通過處理日志和日志聚合來解決。大數據框架hadoop服務角色介紹翻了一下最近一段時間寫的分享，DKHadoop發行版本下載、安裝、運行環境部署等相關內容幾乎都已經寫了一遍了。雖然有的地方可能寫的不是很詳細，個人理解水平有限還請見諒吧！我記得在...

atinosun 2019-08-16 13:40 評論0 收藏0
阿里云正式推出消息隊列Kafka：全面融合開源生態

摘要：在全面兼容Apache Kafka生態的基礎上，消息隊列Kafka徹底解決ApacheKafka穩定性不足的長期痛點，并且支持消息無縫遷移到云上。近日，阿里云宣布正式推出消息隊列Kafka，全面融合開源生態。在全面兼容Apache Kafka生態的基礎上，消息隊列Kafka還具備了超易用，超高可用可靠性，擴縮容不操心，全方位安全診斷，數據安全有保障的特點。可用行達99.9%，數據可靠行99...

aisuhua 2019-06-21 16:48 評論0 收藏0