[[415793]]

【51CTO.com快譯】數(shù)據(jù)集成，通常在企業(yè)的信息架構(gòu)中扮演著重要的角色。具體而言，企業(yè)的分析流程在很大程度上會依賴于此類集成模式，以便從交易系統(tǒng)中，提取方便分析與加載的數(shù)據(jù)格式。

過去，在傳統(tǒng)的架構(gòu)范式中，由于系統(tǒng)之間缺乏互連，事務(wù)和分析經(jīng)常出現(xiàn)延遲，我們只能依賴Batch以實現(xiàn)集成。在Batch模式中，大文件(即數(shù)據(jù)的dump文件)通常是由操作系統(tǒng)生成，并且通過驗證、清理、標(biāo)準(zhǔn)化、以及轉(zhuǎn)換等處理，進而輸出文件以供系統(tǒng)分析。由于此類大文件的讀取會占用大量的內(nèi)存，因此數(shù)據(jù)架構(gòu)師通常會依賴一些暫存類型的數(shù)據(jù)庫，來持續(xù)存儲已完成處理的數(shù)據(jù)輸出。

近年來，隨著以Hadoop為代表的分布式計算的廣泛發(fā)展與應(yīng)用，MapReduce通過在商用硬件上的水平擴展，以分布處理的方式，解決了高內(nèi)存的使用需求。如今，隨著計算技術(shù)的進一步發(fā)展，我們已可以在內(nèi)存中運行MapReduce，并使之成為了處理大型數(shù)據(jù)文件的標(biāo)準(zhǔn)。

就在Batch方式進行演變的過程中，非批(non-batch)處理方式也得到了重大進展。多年來，面向用戶的物聯(lián)網(wǎng)設(shè)備已逐漸成為數(shù)據(jù)系統(tǒng)中的重要一環(huán)。大量數(shù)據(jù)源于物聯(lián)網(wǎng)設(shè)備的采集，而事件驅(qū)動型架構(gòu)也成為了基于微服務(wù)的云原生開發(fā)方法的流行選擇。由于數(shù)據(jù)處理頻率的成倍增加，數(shù)據(jù)流的處理能力成為了數(shù)據(jù)集成工作的主要非功能性需求。因此，曾經(jīng)是大文件數(shù)據(jù)集成問題，已演變成為了流處理需求。這就需要我們提供一個具有足夠緩沖區(qū)的數(shù)據(jù)管道，通過持久性來避免數(shù)據(jù)包的丟失。

在那些以云服務(wù)為主體的平臺上，各種組件的水平擴展能力，相對于數(shù)據(jù)流和使用者而言，要比垂直擴展更加重要。因此，對流的水平可伸縮性以及流的使用者有明確的關(guān)注。這也是諸如Kafka之類的數(shù)據(jù)流解決方案、以及Kubernetes集群需要向使用者(consumer)提供的。目前，Lambda架構(gòu)的Speed層、以及Kappa架構(gòu)的構(gòu)建也都在向此方法發(fā)展。

采用Spark、Kafka和k8s構(gòu)建下一代數(shù)據(jù)管道的目的，本文將討論相關(guān)架構(gòu)模式，以及對應(yīng)的示例代碼，您可以跟著一步一步在自己的環(huán)境中搭建與實現(xiàn)。

Lambda架構(gòu)

Lambda架構(gòu)主要兩個層次：Batch和Stream。Batch能夠按照預(yù)定的批次轉(zhuǎn)換數(shù)據(jù)，而Stream負責(zé)近乎實時地處理數(shù)據(jù)。Batch層通常被使用的場景是：在源系統(tǒng)中批量發(fā)送的數(shù)據(jù)，需要訪問整個數(shù)據(jù)集，以進行所需的數(shù)據(jù)處理，不過因為數(shù)據(jù)集太大，無法執(zhí)行流式處理。相反，那些帶有小塊數(shù)據(jù)包的高速數(shù)據(jù)需要在Speed層被處理。這些數(shù)據(jù)包要么相互獨立，要么按照速度相近的方式形成了對應(yīng)的上下文。顯然，這兩種類型的數(shù)據(jù)處理方式，都屬于計算密集型，盡管Batch層的內(nèi)存需求要高于Speed層。與之對應(yīng)的架構(gòu)方案需要具備可擴展性、容錯性、性能優(yōu)勢、成本效益、靈活性、以及分布式。 

 

圖 1：Lambda架構(gòu)

由上圖可知，由于Lambda需要兩個單獨的組件，來進行Batch和Speed層面的數(shù)據(jù)處理，因此其架構(gòu)較為復(fù)雜。如果我們能夠用某個單一的技術(shù)組件，來同時滿足這兩個目的，則會大幅降低復(fù)雜性。而這正是Apache Spark大顯身手之處。

分布式計算的最新選擇

憑借著包括SparkSQL和SparkStreaming在內(nèi)的一系列庫，Apache Spark作為一種有效的方案，可通過內(nèi)存計算，來實現(xiàn)分布式Lambda架構(gòu)。其中，SparkSQL能夠支持各種Batch操作，例如：通過分布式架構(gòu)加載、驗證、轉(zhuǎn)換、聚合、以及映射數(shù)據(jù)，進而減少對于單臺機器的內(nèi)存需求。同樣，基于SparkStreaming的作業(yè)任務(wù)，可以近乎實時地處理來自Kafka等來源的數(shù)據(jù)流，并將分析結(jié)果提供給諸如：數(shù)據(jù)倉庫或數(shù)據(jù)湖等更為持久的組件。 

 

圖 2：Batch和Speed層的上下文

Kubernetes是一種云平臺集群管理器，其最新版本的Spark，可以運行在由 Kubernetes管理的集群上?？梢哉f，基于Kubernetes的Spark是在云端實現(xiàn)Lambda架構(gòu)的絕佳組合。

雖然我們可以單獨地使用Kubernetes進行分布式計算，但是在這種情況下我們?nèi)孕枰蕾嚩ㄖ频慕鉀Q方案。例如，在Batch層中，Spring Batch框架可以與Kubernetes集群結(jié)合使用，進而將工作任務(wù)分發(fā)到多個集群節(jié)點處。類似地，Kubernetes也可以將流數(shù)據(jù)分發(fā)到多個針對Speed層，而并行運行的Pod。Pod可以通過在其中生成容器，以實現(xiàn)輕松地水平擴展，進而能夠根據(jù)數(shù)據(jù)的體量和速率去調(diào)整集群。

Spark，針對Lambda架構(gòu)的一站式解決方案

針對Batch和Speed層的非功能性需求，Apache Spark具有如下特性：

• 可擴展性：Spark集群可以按需進行擴、縮容。由于它由一個主節(jié)點和一組工作節(jié)點組成，因此這些工作節(jié)點會隨著工作負載的增加，而提高水平擴展的能力。
• 容錯性：Spark框架能夠處理由于工作節(jié)點的崩潰，而導(dǎo)致的集群故障。由于每個數(shù)據(jù)幀都會被邏輯分區(qū)，而每個分區(qū)的數(shù)據(jù)處理都會發(fā)生在某個節(jié)點上。那么，在處理數(shù)據(jù)時，如果某個節(jié)點發(fā)生了故障，那么集群管理器會按照有向無環(huán)圖(Directed Acyclic Graph，DAG)的邏輯，分配另一個節(jié)點來執(zhí)行數(shù)據(jù)幀的相同分區(qū)，進而確保絕對的零數(shù)據(jù)丟失。
• 效率高：由于Spark支持內(nèi)存計算，因此在執(zhí)行期間，數(shù)據(jù)可以根據(jù)Hadoop的需要，被存儲在RAM中，而非磁盤上。其效率顯然要高得多。
• 靈活的負載分配：由于Spark支持分布式計算，能夠橫跨多個節(jié)點共享任務(wù)的組件，并作為一個集成單元生成輸出。Spark可以運行在 Kubernetes 管理的集群上，這使得它在云環(huán)境中更加合適。
• 成本：Spark是開源的，本身不包含任何成本。當(dāng)然，如果選擇托管服務(wù)，則需要付出一定的代價。

現(xiàn)在讓我們深入了解Spark以了解它如何幫助Batch和蒸汽處理。Spark由兩個主要組件組成：Spark核心 API 和Spark庫。核心 API 層提供對四種語言的支持：R、Python、Scala 和 Java。在核心 API 層之上，我們有以下Spark庫，每個庫都針對不同的目的。

• SparkSQL：處理(半)結(jié)構(gòu)化數(shù)據(jù)，執(zhí)行基本轉(zhuǎn)換功能并在數(shù)據(jù)集上執(zhí)行 SQL 查詢SparkStreaming：能夠處理流數(shù)據(jù);支持近實時數(shù)據(jù)處理
• SparkMLib：用于機器學(xué)習(xí);根據(jù)需要用于數(shù)據(jù)處理
• SparkGraphX：用于圖形處理;這里討論的范圍很少使用 

 

圖 3：Spark堆棧(來源：LearningSpark，O'Reilly Media, Inc.)

Batch層的Spark

在Lambda架構(gòu)轉(zhuǎn)換的Batch層，對(半)結(jié)構(gòu)化數(shù)據(jù)的計算、聚合操作由SparkSQL 庫處理。讓我們進一步討論SparkSQL 架構(gòu)。 

 

圖 4：SparkSQL 架構(gòu)

從上圖中可以明顯看出，SparkSQL 具有三個主要架構(gòu)層，如下所述：

• 數(shù)據(jù)源 API：處理不同的數(shù)據(jù)格式，如 CSV、JSON、AVRO 和 PARQUET。它還有助于連接不同的數(shù)據(jù)源，如 HDFS、HIVE、MYSQL、CASSANDRA 等。 用于加載不同格式數(shù)據(jù)的通用 API：

Dataframe.read.load(“ParquetFile | JsonFile | TextFile | CSVFile | AVROFile”)

• 數(shù)據(jù)幀API：Spark2.0 以后，Spark數(shù)據(jù)幀被大量使用。它有助于保存大型關(guān)系數(shù)據(jù)并公開多個轉(zhuǎn)換函數(shù)以對數(shù)據(jù)集進行切片和切塊。通過 Dataframe API 公開的此類轉(zhuǎn)換函數(shù)的示例是：

withColumn, select, withColumnRenamed, groupBy, filter, sort, orderBy etc.

• SQL 服務(wù)：SparkSQL 服務(wù)是幫助我們創(chuàng)建數(shù)據(jù)框和保存關(guān)系數(shù)據(jù)以進行進一步轉(zhuǎn)換的主要元素。這是我們使用SparkSQL 時Batch層轉(zhuǎn)換的入口點。在轉(zhuǎn)換過程中，可以使用python、R、Scala或Java中的不同API，也可以直接執(zhí)行SQL來轉(zhuǎn)換數(shù)據(jù)。

下面是一些Batch的代碼示例：

假設(shè)有兩個表：一個是 PRODUCT，另一個是 TRANSACTION。PRODUCT 表包含商店特定產(chǎn)品的所有信息，Transaction 表包含針對每個產(chǎn)品的所有交易。我們可以通過轉(zhuǎn)換和聚合得到以下信息。

• 產(chǎn)品明智的總銷售量
• 分部明智的總收入

通過在Spark數(shù)據(jù)幀上編寫純 SQL 或使用聚合函數(shù)可以獲得相同的結(jié)果。 

Python 
from pyspark.sql importSparkSession 
from pyspark.sql.functions import * 
Spark=SparkSession.builder.master("local").appName("Superstore").getOrCreate() 
df1 =Spark.read.csv("Product.csv") 
df2 =Spark.read.csv("Transaction.csv") 
df3 = df1.filter(df1.Segment != 'Electric') 
df4 = df2.withColumn("OrderDate",df2.OrderDate[7:10]) 
result_df1 = df3.join(df4, on= ['ProductCode'], how='inner') 
result_df2 = result_df1.groupBy('ProductName').sum('Quantity') 
result_df2.show() 
# Display segment wise revenue generated 
result_df3 = result_df1.groupBy('Segment').sum('Price') 
result_df3.show() 
Python 
from pyspark.sql importSparkSession 
from pyspark.sql.functions import * 
Spark=SparkSession.builder.master("local").appName("Superstore").getOrCreate() 
df1 =Spark.read.csv("Product.csv") 
df2 =Spark.read.csv("Transaction.csv") 
df3 = df1.filter(df1.Segment != 'Electric') 
df4 = df2.withColumn("OrderDate",df2.OrderDate[7:10]) 
result_df1 = df3.join(df4, on= ['ProductCode'], how='inner') 
result_df1.createOrReplaceTempView("SuperStore") 
# Display product wise quantity sold 
result_df2 =Spark.sql("select ProductName , Sum(Quantity) from Superstore group by ProductName") 
result_df2.show() 
# Display segment wise revenue earned 
result_df3 =Spark.sql("select Segment , Sum(Price) from Superstore group by Segment") 
result_df2.show() 

在這兩種情況下，第一個數(shù)據(jù)是從兩個不同的來源加載的，并且產(chǎn)品數(shù)據(jù)針對所有非電氣產(chǎn)品進行過濾。交易數(shù)據(jù)根據(jù)訂單日期的某種格式進行更改。然后，將兩個數(shù)據(jù)幀連接起來，并生成該超市中細分收入和產(chǎn)品銷售數(shù)量的結(jié)果。

當(dāng)然，這是加載、驗證、轉(zhuǎn)換和聚合的簡單示例。使用SparkSQL 可以進行更復(fù)雜的操作。要了解有關(guān)SparkSQL 服務(wù)的更多信息，請參閱此處的文檔。

Sparkfor Speed 層

SparkStreaming 是一個庫，用于核心Spark框架之上。它確保實時數(shù)據(jù)流處理的可擴展性、高吞吐量和容錯性。 

 

圖 5：SparkStreaming 架構(gòu)(來源：https : //spark.apache.org)

如上圖所示，Spark將輸入數(shù)據(jù)流轉(zhuǎn)換為批量輸入數(shù)據(jù)。這種離散Batch有兩種實現(xiàn)方式：a) Dstreams 或離散化流和 b) 結(jié)構(gòu)化流。前者非常受歡迎，直到后者作為更高級的版本出現(xiàn)。但是，Dstream 還沒有完全過時，為了完整起見，將其保留在本文中。

· Discretized Streams：這提供了對火花流庫的抽象。它是 RDD 的集合，代表一個連續(xù)的數(shù)據(jù)流。它將數(shù)據(jù)離散成小批量并運行小作業(yè)來處理這些小批量。任務(wù)根據(jù)數(shù)據(jù)的位置分配給工作節(jié)點。因此，通過 Dstream 的這個概念，Spark可以并行讀取數(shù)據(jù)，執(zhí)行小批量處理流并確保流處理的有效節(jié)點分配。

· 結(jié)構(gòu)化流：這是使用Spark引擎的最先進和現(xiàn)代的流處理方法。它與SparkDataframe API(在上面的Batch部分中討論)很好地集成在一起，用于對流數(shù)據(jù)的各種操作。結(jié)構(gòu)化流可以增量和連續(xù)地處理數(shù)據(jù)?；谔囟ù翱诤退〉慕鼘崟r聚合也是可能的。

Spark結(jié)構(gòu)化流可以處理不同的流處理用例，如下面的示例所示：

簡單的結(jié)構(gòu)化流媒體

簡單的結(jié)構(gòu)化流只會轉(zhuǎn)換和加載來自流的數(shù)據(jù)，并且不包括特定時間范圍內(nèi)的任何聚合。例如，系統(tǒng)從 Apache Kafka 獲取數(shù)據(jù)，并通過Spark流和SparkSQL 近乎實時地對其進行轉(zhuǎn)換(請參閱下面的代碼片段)。 

Python 
from pyspark.sql importSparkSession 
from pyspark.streaming import StreamingContext 
import pyspark.sql.functions as sf 
spark=SparkSession.builder.master('local').appName('StructuredStreamingApp').getOrCreate() 
df =Spark.readStream.format("kafka").option("kafka.bootstrap.servers,"localhost:9092") 
                                             .option("subscribe", "test_topic").load() 
df1 = df.selectExpr("CAST(value AS STRING)") 
df2 = df1.selectExpr("split(value, ',')[0] as Dept","split(value, ',')[1] as Age") 
df2.show() 

SparkSession 對象的ReadStream函數(shù)用于連接特定的 Kafka 主題。正如上面選項中的代碼片段一樣，我們需要提供 Kafka 集群代理的 IP 和 Kafka 主題名稱。此代碼的輸出是一個表，有兩列：Dept 和 Age。

結(jié)構(gòu)化流媒體聚合

可以通過 Structured Streaming 對流數(shù)據(jù)進行聚合，它能夠在新事件到達的基礎(chǔ)上計算滾動聚合結(jié)果。這是對整個數(shù)據(jù)流的運行聚合。請參考下面的代碼片段，它在整個數(shù)據(jù)流上推導(dǎo)出部門明智的平均年齡。 

Python 
from pyspark.sql importSparkSession 
from pyspark.streaming import StreamingContext 
import pyspark.sql.functions as sf 
spark=SparkSession.builder.master('local').appName('StructuredStreamingApp').getOrCreate() 
df =Spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092") 
                                     .option("subscribe", "test_topic").load() 
df1 = df.selectExpr("CAST(value AS STRING)") 
df2 = df1.selectExpr("split(value, ',')[0] as Dept","split(value, ',')[1] as Age") 
df3 = df2.groupBy("Dept").avg("Age") 
df3.show() 

窗口聚合

有時我們需要在某個時間窗口內(nèi)進行聚合，而不是運行聚合。SparkStructured Streaming 也提供了這樣的功能。假設(shè)我們要計算過去 5 分鐘內(nèi)的事件數(shù)。這個帶聚合的窗口函數(shù)將幫助我們。 

Python 
from pyspark.sql importSparkSession 
from pyspark.streaming import StreamingContext 
import pyspark.sql.functions as sf 
import datetime 
import time 
spark=SparkSession.builder.master('local').appName('StructuredStreamingApp').getOrCreate() 
df =Spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092") 
                                     .option("subscribe", "test_topic").load() 
df1 = df.selectExpr("CAST(value AS STRING)") 
df2 = df1.selectExpr("split(value, ',')[0] as Dept","split(value, ',')[1] as Age") 
df3 = df2.withColumn("Age", df2.Age.cast('int')) 
df4 = df3.withColumn("eventTime",sf.current_timestamp()) 
df_final = df4.groupBy(sf.window("eventTime", "5 minute")).count() 
df_final.show() 

重疊窗口上的聚合

在上面的例子中，每個窗口都是一個完成聚合的組。還提供了通過提及窗口長度和滑動間隔來定義重疊窗口的規(guī)定。它在窗口聚合中的后期數(shù)據(jù)處理中非常有用。下面的代碼基于 5 分鐘窗口計算事件數(shù)，滑動間隔為 10 分鐘。 

Python 
from pyspark.sql importSparkSession 
from pyspark.streaming import StreamingContext 
import pyspark.sql.functions as sf 
import datetime 
import time 
spark=SparkSession.builder.master('local').appName('StructuredStreamingApp').getOrCreate() 
df =Spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092") 
                                     .option("subscribe", "test_topic").load() 
df1 = df.selectExpr("CAST(value AS STRING)") 
df2 = df1.selectExpr("split(value, ',')[0] as Dept","split(value, ',')[1] as Age") 
df3 = df2.withColumn("Age", df2.Age.cast('int')) 
df4 = df3.withColumn("eventTime",sf.current_timestamp()) 
df_final = df4.groupBy("Dept",sf.window("eventTime","10 minutes", "5 minute")).count() 
df_final.show() 

帶水印和重疊窗口的聚合

數(shù)據(jù)遲到會在近實時系統(tǒng)的聚合中產(chǎn)生問題。我們可以使用重疊窗口來解決這個錯誤。但問題是：系統(tǒng)等待遲到的數(shù)據(jù)需要多長時間?這可以通過水印解決。通過這種方法，我們在重疊窗口之上定義了一個特定的時間段。之后，系統(tǒng)丟棄該事件。 

Python 
from pyspark.sql importSparkSession 
from pyspark.streaming import StreamingContext 
import pyspark.sql.functions as sf 
import datetime 
import time 
spark=SparkSession.builder.master('local').appName('StructuredStreamingApp').getOrCreate() 
df =Spark.readStream.format("kafka").option("kafka.bootstrap.servers", "localhost:9092") 
                                     .option("subscribe", "test_topic").load() 
df1 = df.selectExpr("CAST(value AS STRING)") 
df2 = df1.selectExpr("split(value, ',')[0] as Dept","split(value, ',')[1] as Age") 
df3 = df2.withColumn("Age", df2.Age.cast('int')) 
df4 = df3.withColumn("eventTime",sf.current_timestamp()) 
df_final = df4.withWatermark("eventTime","10 Minutes").groupBy("Dept",sf.window("eventTime","10 minutes", "5 minute")).count() 
df_final.show() 

上面的代碼表示對于延遲事件，10 分鐘后，舊窗口結(jié)果將不會更新。

Kafka + k8s - Speed層的另一種解決方案

托管在 Kubernetes 集群上的 Pod 形成了 Kafka 流的消費者組，是另一種近乎實時數(shù)據(jù)處理的方法。通過使用這種組合，我們可以輕松獲得分布式計算的優(yōu)勢。 

 

圖 6：通過 Kafka + Kubernetes 實現(xiàn)的Speed層示例

在上面例子中的事件驅(qū)動系統(tǒng)中，數(shù)據(jù)正在從 Kafka 主題加載到基于 Python 的處理單元中。如果 Kafka 集群中的分區(qū)數(shù)量與 Pod 的復(fù)制因子匹配，則 Pod 一起組成一個消費者組，消息被無縫消費。

這是構(gòu)建分布式數(shù)據(jù)處理系統(tǒng)的經(jīng)典示例，僅使用Kafka+k8s組合即可確保并行處理。

使用 Python 創(chuàng)建 Kafka 消費者的兩個非常流行的庫是：

• Python_Kafka 庫
• Confluent_Kafka 庫 

Python_Kafka 
Python 
from kafka import KafkaConsumer 
consumer = KafkaConsumer(TopicName, 
                         bootstrap_servers= <broker-list>, 
                         group_id=<GroupName>, 
                         enable_auto_commit=True, 
                         auto_offset_reset='earliest') 
consumer.poll() 
Confluent_Kafka  
Python 
from confluent_kafka import Consumer 
consumer = Consumer({'bootstrap.servers': <broker-list>, 
                     'group.id': <GroupName>, 
                     'enable.auto.commit': True, 
                     'auto.offset.reset': 'earliest' 
                     }) 
consumer.subscribe([TopicName]) 

K8.yml 文件的示例結(jié)構(gòu)如下： 

YAML 
metadata: 
  name: <app name> 
  namespace: <deployment namespace> 
  labels: 
    app: <app name> 
spec: 
  replicas: <replication-factor> 
  spec: 
      containers: 
      - name: <container name> 

如果按照上述方式開發(fā)基本組件，系統(tǒng)將獲得分布式計算的幫助，而無需進行內(nèi)存計算。一切都取決于系統(tǒng)的體積和所需速度。對于低/中等數(shù)據(jù)量，可以通過實現(xiàn)這種基于 python-k8 的架構(gòu)來確保良好的速度。

這兩種方法都可以托管在具有各種服務(wù)的云中。例如，我們在 AWS 中有 EMR 和 Glue，可以在 GCP 中通過 Dataproc 創(chuàng)建Spark集群，或者我們可以在 Azure 中使用 Databricks。另一方面，kafka-python-k8的方式可以很容易地在云端實現(xiàn)，這保證了更好的可管理性。例如在 AWS 中，我們可以將 MSK 或 Kinesis 和 EKS 的組合用于這種方法。在下一個版本中，我們將討論所有云供應(yīng)商中Batch和Speed層的實現(xiàn)，并根據(jù)不同的需求提供比較研究。

原文標(biāo)題：Next-Gen Data Pipes WithSpark, Kafka and k8s，作者：Subhendu Dey & Abhishek Sinha

【51CTO譯稿，合作站點轉(zhuǎn)載請注明原文譯者和出處為51CTO.com】