引言

隨著在CDH平臺上物聯網(IoT)使用案例的不斷增加，針對這些工作負載的安全性顯得至關重要。本篇博文對如何以安全的方式在Spark中使用來自Kafka的數據，以及針對物聯網(IoT)使用案例的兩個關鍵組件進行了說明。

Cloudera Distribution of Apache Kafka 2.0.0版本(基于Apache Kafka 0.9.0)引入了一種新型的Kafka消費者API，可以允許消費者從安全的Kafka集群中讀取數據。這樣可以允許管理員鎖定其Kafka集群，并要求客戶通過Kerberos進行身份驗證。此外，也可以允許客戶在與Kafka brokers(通過SSL/TLS)通信時加密數據隨后，在Cloudera Distribution of Apache Kafka 2.1.0版本中，Kafka通過Apache Sentry引入了支持授權功能。這樣可以允許Kafka管理員鎖定某些主題，并針對特定角色和用戶授予權限，充分發揮基于角色的訪問控制功能。

而現在，從Cloudera Distribution of Spark 2.1的第一次發行版開始，我們已經具備了從Spark中的Kafka內安全讀取數據的功能。

要求

• Cloudera Distribution Spark 2.1第一次發行版或更高版本。
• Cloudera Distribution Kafka 2.1.0版本或更高版本。

體系架構

使用Spark中新的直接連接器可以支持從安全的Kafka集群中獲取消息。直接連接器不使用單獨的進程(亦稱為接收器)讀取數據。相反，Spark驅動程序將跟蹤各種Kafka主題分區的偏移量，并將偏移量發送到從Kafka中直接讀取數據的執行程序中。直接連接器的簡單描述如下所示。 

圖1：Spark中的Kafka直接連接器

• Spark節點和Kafka 代理人(broker)不一定在同一地點。
• 一個Spark分區對應一個Kafka主題分區。
• 如果出于某種原因，多個主題分區位于單個Kafka節點上，則有多個Spark執行程序可能會命中該節點(不過沒關系)。
• 上圖只是一個簡單的說明。

非常值得注意的一點是，Spark是以分布式的方式訪問Kafka中的數據。Spark中的每一個任務都會從某個Kafka主題的特定分區中讀取數據，該特定分區稱為主題分區。主題分區理想地均勻分布在Kafka 代理人(broker)之間。

但是，為了以分布式的方式從安全的Kafka中讀取數據，我們需要在Kafka(KAFKA-1696)中使用Hadoop風格的授權令牌，在寫本篇博文時(2017年春季)還不支持這一功能。

我們已經考慮了各種解決這個問題的方法，但是最終決定采用從Kafka中安全讀取數據的建議解決方案(至少應實現Kafka授權令牌的支持)將是Spark應用程序分發用戶的keytab，以便執行程序可以訪問。然后，執行程序將使用共享的用戶密鑰表，與Kerberos密鑰分發中心(KDC)進行身份驗證，并從Kafka 代理人(broker)中讀取數據。YARN分布式緩存用于從客戶端(即網關節點)向驅動程序和執行程序發送和共享密鑰表。下圖顯示了當前解決方案的一覽圖。 

圖2：當前解決方案(假設YARN集群模式)

這種方法存在以下一些常見的問題：

a. 這不能被認為是發送鑰匙表的最佳安全實踐

b.在具有大量Kafka主題分區的情況下，所有執行程序可能會同時嘗試登錄KDC，存在導致重送攻擊的潛在風險(類似于DDOS攻擊)。

關于問題a)，Spark已經使用分布式緩存將用戶的密鑰表從客戶端(亦稱為網關)節點發送到驅動程序，并且由于缺少授權令牌，所以沒有辦法繞過。管理員可以選擇自己在Spark外部將密鑰表分發到Spark執行程序節點(即YARN節點，因為Spark在YARN上運行)，并調整優化共享的示例應用程序以緩解該問題。

關于問題b)，我們在Kafka主題中測試了1000多個主題分區，并且在增加分區數量后未見對KDC服務器產生不利影響。

與Apache Sentry集成

例應用程序假設沒有使用任何Kafka授權。如果使用了Kafka授權的話(通過Apache Sentry)，則必須確保應用程序中指定的消費者小組已經獲得Sentry授權。例如，如果應用程序的消費者小組的名稱是my-consumer-group，則必須同時對my-consumer-group和spark-executor-my-consumer-group授予訪問權限(即您的消費者小組名稱前綴為spark-executor-)。這是因為Spark驅動器使用是該應用程序指定的消費者小組，但spark執行程序在此集成中使用的是不同的消費者小組，該集成在驅動程序消費者小組的名稱前指定的前綴是spark-executor-。

結論

簡而言之，您可以使用Cloudera Distribution of Apache Kafka 2.1.0 版本(或更高版本)和Cloudera Distribution of Apache Spark 2.1第一次發行版(或更高版本)，以安全的方式從Kafka中使用Spark內的數據——包括身份驗證(使用Kerberos進行身份認證)、授權(使用Sentry進行授權)以及線上加密(使用SSL/TLS進行加密)。