spark系列-結(jié)構(gòu)化數(shù)據(jù)流-集成kafka數(shù)據(jù)源

IT那活兒發(fā)布于2023-01-11 13:20 / 2504人閱讀

點(diǎn)擊上方“IT那活兒”，關(guān)注后了解更多內(nèi)容，不管IT什么活兒，干就完了！！！

步驟

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

1. 添加依賴

2. 創(chuàng)建kafka數(shù)據(jù)流

3. 創(chuàng)建kafka批處理數(shù)據(jù)集

4. 數(shù)據(jù)流每行包含的內(nèi)容

選項(xiàng)配置

1. 必選項(xiàng)

對于kafka源，對于批處理，流處理以下選項(xiàng)是必選：

2. 可選項(xiàng)

消費(fèi)者池

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

消費(fèi)者緩存關(guān)鍵是topic name、topic partition、group id。

以下選項(xiàng)可以配置消費(fèi)者池（consumer pool）：

提取數(shù)據(jù)時數(shù)據(jù)池的配置選項(xiàng)：

向kafka寫入消息

向kafka寫入消息時不可避免會出現(xiàn)重復(fù)數(shù)據(jù)，一個可能解決方案是引入一個主鍵，再讀取時可以執(zhí)行重復(fù)數(shù)據(jù)消除。

向kafka寫入消息的schema應(yīng)該包括以下列：

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

如上只有value列是必須的，如果沒有指定key則默認(rèn)是null，如果指定了topic則將寫入指定的topic，除非配置選項(xiàng)中指定了topic，否則必須指定topic列，配置選項(xiàng)中指定的topic優(yōu)先級高，如果未指定partition列，則由生產(chǎn)者進(jìn)行計(jì)算處理，可以通過設(shè)置kafka.partitioner.class選項(xiàng)指定分區(qū)器，如果未指定，則使用默認(rèn)分區(qū)器。

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

以下是可選選項(xiàng)：

基于流查詢創(chuàng)建kafka sink：

//選項(xiàng)中指定topic。

基于批查詢創(chuàng)建kafka sink：

//選項(xiàng)中指定topic。

//指定topic列。

生產(chǎn)者池

有消費(fèi)者池，也有生產(chǎn)者池，生產(chǎn)者池配置：

Kafka特定配置

消費(fèi)者配置選項(xiàng)。

生產(chǎn)者配置選項(xiàng)。

安全

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

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

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

該過程使用委托令牌初始化，當(dāng)設(shè)置了spark.kafka.clusters.${cluster}.auth..bootstrap.servers，spark按照優(yōu)先級考慮一下登錄選項(xiàng)。

JAAS登錄配置，通過例子進(jìn)行說明：

1）keytab file

2）Kerberos 憑據(jù)緩存

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

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

SASL SSL（默認(rèn)協(xié)議），SSL，SASL PLAINTEXT（用于測試）。

獲取令牌成功之后，spark向其他節(jié)點(diǎn)進(jìn)行分發(fā)，令牌使用SCRAM 登錄模塊進(jìn)行登錄認(rèn)證，因此必須進(jìn)行相應(yīng)的配置（spark.kafka.clusters.${cluster}.sasl.token.mechanism（默認(rèn)值是SCRAM-SHA-512）），該參數(shù)必須與kafka broker配置參數(shù)匹配。

當(dāng)令牌在executor上可用時，spark會按優(yōu)先順序考慮以下登錄選項(xiàng)：

JAAS login configuration，

委托令牌，有關(guān)詳細(xì)信息，請參閱spark.kafka.clusters.${cluster}.target.bootstrap.servers.regex參數(shù)。

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

配置選項(xiàng)：