掃二維碼與項(xiàng)目經(jīng)理溝通
我們?cè)谖⑿派?4小時(shí)期待你的聲音
解答本文疑問/技術(shù)咨詢/運(yùn)營(yíng)咨詢/技術(shù)建議/互聯(lián)網(wǎng)交流
譯文
作者：陳峻 2021-08-06 09:47:43
開發(fā)
前端
Kafka
Spark 本文檢查了架構(gòu)模式，并提供了一些示例代碼供讀者在自己的環(huán)境中實(shí)現(xiàn)。
創(chuàng)新互聯(lián)專注于張家界企業(yè)網(wǎng)站建設(shè),響應(yīng)式網(wǎng)站設(shè)計(jì),成都做商城網(wǎng)站。張家界網(wǎng)站建設(shè)公司,為張家界等地區(qū)提供建站服務(wù)。全流程按需網(wǎng)站設(shè)計(jì)，專業(yè)設(shè)計(jì)，全程項(xiàng)目跟蹤，創(chuàng)新互聯(lián)專業(yè)和態(tài)度為您提供的服務(wù)
【51CTO.com快譯】數(shù)據(jù)集成，通常在企業(yè)的信息架構(gòu)中扮演著重要的角色。具體而言，企業(yè)的分析流程在很大程度上會(huì)依賴于此類集成模式，以便從交易系統(tǒng)中，提取方便分析與加載的數(shù)據(jù)格式。
過去，在傳統(tǒng)的架構(gòu)范式中，由于系統(tǒng)之間缺乏互連，事務(wù)和分析經(jīng)常出現(xiàn)延遲，我們只能依賴Batch以實(shí)現(xiàn)集成。在Batch模式中，大文件(即數(shù)據(jù)的dump文件)通常是由操作系統(tǒng)生成，并且通過驗(yàn)證、清理、標(biāo)準(zhǔn)化、以及轉(zhuǎn)換等處理，進(jìn)而輸出文件以供系統(tǒng)分析。由于此類大文件的讀取會(huì)占用大量的內(nèi)存，因此數(shù)據(jù)架構(gòu)師通常會(huì)依賴一些暫存類型的數(shù)據(jù)庫(kù)，來持續(xù)存儲(chǔ)已完成處理的數(shù)據(jù)輸出。
近年來，隨著以Hadoop為代表的分布式計(jì)算的廣泛發(fā)展與應(yīng)用，MapReduce通過在商用硬件上的水平擴(kuò)展，以分布處理的方式，解決了高內(nèi)存的使用需求。如今，隨著計(jì)算技術(shù)的進(jìn)一步發(fā)展，我們已可以在內(nèi)存中運(yùn)行MapReduce，并使之成為了處理大型數(shù)據(jù)文件的標(biāo)準(zhǔn)。
就在Batch方式進(jìn)行演變的過程中，非批(non-batch)處理方式也得到了重大進(jìn)展。多年來，面向用戶的物聯(lián)網(wǎng)設(shè)備已逐漸成為數(shù)據(jù)系統(tǒng)中的重要一環(huán)。大量數(shù)據(jù)源于物聯(lián)網(wǎng)設(shè)備的采集，而事件驅(qū)動(dòng)型架構(gòu)也成為了基于微服務(wù)的云原生開發(fā)方法的流行選擇。由于數(shù)據(jù)處理頻率的成倍增加，數(shù)據(jù)流的處理能力成為了數(shù)據(jù)集成工作的主要非功能性需求。因此，曾經(jīng)是大文件數(shù)據(jù)集成問題，已演變成為了流處理需求。這就需要我們提供一個(gè)具有足夠緩沖區(qū)的數(shù)據(jù)管道，通過持久性來避免數(shù)據(jù)包的丟失。
在那些以云服務(wù)為主體的平臺(tái)上，各種組件的水平擴(kuò)展能力，相對(duì)于數(shù)據(jù)流和使用者而言，要比垂直擴(kuò)展更加重要。因此，對(duì)流的水平可伸縮性以及流的使用者有明確的關(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)模式，以及對(duì)應(yīng)的示例代碼，您可以跟著一步一步在自己的環(huán)境中搭建與實(shí)現(xiàn)。
Lambda架構(gòu)主要兩個(gè)層次：Batch和Stream。Batch能夠按照預(yù)定的批次轉(zhuǎn)換數(shù)據(jù)，而Stream負(fù)責(zé)近乎實(shí)時(shí)地處理數(shù)據(jù)。Batch層通常被使用的場(chǎng)景是：在源系統(tǒng)中批量發(fā)送的數(shù)據(jù)，需要訪問整個(gè)數(shù)據(jù)集，以進(jìn)行所需的數(shù)據(jù)處理，不過因?yàn)閿?shù)據(jù)集太大，無(wú)法執(zhí)行流式處理。相反，那些帶有小塊數(shù)據(jù)包的高速數(shù)據(jù)需要在Speed層被處理。這些數(shù)據(jù)包要么相互獨(dú)立，要么按照速度相近的方式形成了對(duì)應(yīng)的上下文。顯然，這兩種類型的數(shù)據(jù)處理方式，都屬于計(jì)算密集型，盡管Batch層的內(nèi)存需求要高于Speed層。與之對(duì)應(yīng)的架構(gòu)方案需要具備可擴(kuò)展性、容錯(cuò)性、性能優(yōu)勢(shì)、成本效益、靈活性、以及分布式。
圖 1：Lambda架構(gòu)
由上圖可知，由于Lambda需要兩個(gè)單獨(dú)的組件，來進(jìn)行Batch和Speed層面的數(shù)據(jù)處理，因此其架構(gòu)較為復(fù)雜。如果我們能夠用某個(gè)單一的技術(shù)組件，來同時(shí)滿足這兩個(gè)目的，則會(huì)大幅降低復(fù)雜性。而這正是Apache Spark大顯身手之處。
憑借著包括SparkSQL和SparkStreaming在內(nèi)的一系列庫(kù)，Apache Spark作為一種有效的方案，可通過內(nèi)存計(jì)算，來實(shí)現(xiàn)分布式Lambda架構(gòu)。其中，SparkSQL能夠支持各種Batch操作，例如：通過分布式架構(gòu)加載、驗(yàn)證、轉(zhuǎn)換、聚合、以及映射數(shù)據(jù)，進(jìn)而減少對(duì)于單臺(tái)機(jī)器的內(nèi)存需求。同樣，基于SparkStreaming的作業(yè)任務(wù)，可以近乎實(shí)時(shí)地處理來自Kafka等來源的數(shù)據(jù)流，并將分析結(jié)果提供給諸如：數(shù)據(jù)倉(cāng)庫(kù)或數(shù)據(jù)湖等更為持久的組件。
圖 2：Batch和Speed層的上下文
Kubernetes是一種云平臺(tái)集群管理器，其最新版本的Spark，可以運(yùn)行在由 Kubernetes管理的集群上?？梢哉f，基于Kubernetes的Spark是在云端實(shí)現(xiàn)Lambda架構(gòu)的絕佳組合。
雖然我們可以單獨(dú)地使用Kubernetes進(jìn)行分布式計(jì)算，但是在這種情況下我們?nèi)孕枰蕾嚩ㄖ频慕鉀Q方案。例如，在Batch層中，Spring Batch框架可以與Kubernetes集群結(jié)合使用，進(jìn)而將工作任務(wù)分發(fā)到多個(gè)集群節(jié)點(diǎn)處。類似地，Kubernetes也可以將流數(shù)據(jù)分發(fā)到多個(gè)針對(duì)Speed層，而并行運(yùn)行的Pod。Pod可以通過在其中生成容器，以實(shí)現(xiàn)輕松地水平擴(kuò)展，進(jìn)而能夠根據(jù)數(shù)據(jù)的體量和速率去調(diào)整集群。
針對(duì)Batch和Speed層的非功能性需求，Apache Spark具有如下特性：
現(xiàn)在讓我們深入了解Spark以了解它如何幫助Batch和蒸汽處理。Spark由兩個(gè)主要組件組成：Spark核心 API 和Spark庫(kù)。核心 API 層提供對(duì)四種語(yǔ)言的支持：R、Python、Scala 和 Java。在核心 API 層之上，我們有以下Spark庫(kù)，每個(gè)庫(kù)都針對(duì)不同的目的。
圖 3：Spark堆棧(來源：LearningSpark，O'Reilly Media, Inc.)
在Lambda架構(gòu)轉(zhuǎn)換的Batch層，對(duì)(半)結(jié)構(gòu)化數(shù)據(jù)的計(jì)算、聚合操作由SparkSQL 庫(kù)處理。讓我們進(jìn)一步討論SparkSQL 架構(gòu)。
圖 4：SparkSQL 架構(gòu)
從上圖中可以明顯看出，SparkSQL 具有三個(gè)主要架構(gòu)層，如下所述：
Dataframe.read.load(“ParquetFile | JsonFile | TextFile | CSVFile | AVROFile”)
withColumn, select, withColumnRenamed, groupBy, filter, sort, orderBy etc.
下面是一些Batch的代碼示例：
假設(shè)有兩個(gè)表：一個(gè)是 PRODUCT，另一個(gè)是 TRANSACTION。PRODUCT 表包含商店特定產(chǎn)品的所有信息，Transaction 表包含針對(duì)每個(gè)產(chǎn)品的所有交易。我們可以通過轉(zhuǎ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()
在這兩種情況下，第一個(gè)數(shù)據(jù)是從兩個(gè)不同的來源加載的，并且產(chǎn)品數(shù)據(jù)針對(duì)所有非電氣產(chǎn)品進(jìn)行過濾。交易數(shù)據(jù)根據(jù)訂單日期的某種格式進(jìn)行更改。然后，將兩個(gè)數(shù)據(jù)幀連接起來，并生成該超市中細(xì)分收入和產(chǎn)品銷售數(shù)量的結(jié)果。
當(dāng)然，這是加載、驗(yàn)證、轉(zhuǎn)換和聚合的簡(jiǎn)單示例。使用SparkSQL 可以進(jìn)行更復(fù)雜的操作。要了解有關(guān)SparkSQL 服務(wù)的更多信息，請(qǐng)參閱此處的文檔。
SparkStreaming 是一個(gè)庫(kù)，用于核心Spark框架之上。它確保實(shí)時(shí)數(shù)據(jù)流處理的可擴(kuò)展性、高吞吐量和容錯(cuò)性。
圖 5：SparkStreaming 架構(gòu)(來源：https : //spark.apache.org)
如上圖所示，Spark將輸入數(shù)據(jù)流轉(zhuǎn)換為批量輸入數(shù)據(jù)。這種離散Batch有兩種實(shí)現(xiàn)方式：a) Dstreams 或離散化流和 b) 結(jié)構(gòu)化流。前者非常受歡迎，直到后者作為更高級(jí)的版本出現(xiàn)。但是，Dstream 還沒有完全過時(shí)，為了完整起見，將其保留在本文中。
· Discretized Streams：這提供了對(duì)火花流庫(kù)的抽象。它是 RDD 的集合，代表一個(gè)連續(xù)的數(shù)據(jù)流。它將數(shù)據(jù)離散成小批量并運(yùn)行小作業(yè)來處理這些小批量。任務(wù)根據(jù)數(shù)據(jù)的位置分配給工作節(jié)點(diǎn)。因此，通過 Dstream 的這個(gè)概念，Spark可以并行讀取數(shù)據(jù)，執(zhí)行小批量處理流并確保流處理的有效節(jié)點(diǎn)分配。
· 結(jié)構(gòu)化流：這是使用Spark引擎的最先進(jìn)和現(xiàn)代的流處理方法。它與SparkDataframe API(在上面的Batch部分中討論)很好地集成在一起，用于對(duì)流數(shù)據(jù)的各種操作。結(jié)構(gòu)化流可以增量和連續(xù)地處理數(shù)據(jù)?；谔囟ù翱诤退〉慕鼘?shí)時(shí)聚合也是可能的。
Spark結(jié)構(gòu)化流可以處理不同的流處理用例，如下面的示例所示：
簡(jiǎn)單的結(jié)構(gòu)化流只會(huì)轉(zhuǎn)換和加載來自流的數(shù)據(jù)，并且不包括特定時(shí)間范圍內(nèi)的任何聚合。例如，系統(tǒng)從 Apache Kafka 獲取數(shù)據(jù)，并通過Spark流和SparkSQL 近乎實(shí)時(shí)地對(duì)其進(jìn)行轉(zhuǎn)換(請(qǐng)參閱下面的代碼片段)。
- 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 對(duì)象的ReadStream函數(shù)用于連接特定的 Kafka 主題。正如上面選項(xiàng)中的代碼片段一樣，我們需要提供 Kafka 集群代理的 IP 和 Kafka 主題名稱。此代碼的輸出是一個(gè)表，有兩列：Dept 和 Age。
可以通過 Structured Streaming 對(duì)流數(shù)據(jù)進(jìn)行聚合，它能夠在新事件到達(dá)的基礎(chǔ)上計(jì)算滾動(dòng)聚合結(jié)果。這是對(duì)整個(gè)數(shù)據(jù)流的運(yùn)行聚合。請(qǐng)參考下面的代碼片段，它在整個(gè)數(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()
有時(shí)我們需要在某個(gè)時(shí)間窗口內(nèi)進(jìn)行聚合，而不是運(yùn)行聚合。SparkStructured Streaming 也提供了這樣的功能。假設(shè)我們要計(jì)算過去 5 分鐘內(nèi)的事件數(shù)。這個(gè)帶聚合的窗口函數(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()
在上面的例子中，每個(gè)窗口都是一個(gè)完成聚合的組。還提供了通過提及窗口長(zhǎng)度和滑動(dòng)間隔來定義重疊窗口的規(guī)定。它在窗口聚合中的后期數(shù)據(jù)處理中非常有用。下面的代碼基于 5 分鐘窗口計(jì)算事件數(shù)，滑動(dòng)間隔為 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ù)遲到會(huì)在近實(shí)時(shí)系統(tǒng)的聚合中產(chǎn)生問題。我們可以使用重疊窗口來解決這個(gè)錯(cuò)誤。但問題是：系統(tǒng)等待遲到的數(shù)據(jù)需要多長(zhǎng)時(shí)間?這可以通過水印解決。通過這種方法，我們?cè)谥丿B窗口之上定義了一個(gè)特定的時(shí)間段。之后，系統(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()
上面的代碼表示對(duì)于延遲事件，10 分鐘后，舊窗口結(jié)果將不會(huì)更新。
托管在 Kubernetes 集群上的 Pod 形成了 Kafka 流的消費(fèi)者組，是另一種近乎實(shí)時(shí)數(shù)據(jù)處理的方法。通過使用這種組合，我們可以輕松獲得分布式計(jì)算的優(yōu)勢(shì)。
圖 6：通過 Kafka + Kubernetes 實(shí)現(xiàn)的Speed層示例
在上面例子中的事件驅(qū)動(dòng)系統(tǒng)中，數(shù)據(jù)正在從 Kafka 主題加載到基于 Python 的處理單元中。如果 Kafka 集群中的分區(qū)數(shù)量與 Pod 的復(fù)制因子匹配，則 Pod 一起組成一個(gè)消費(fèi)者組，消息被無(wú)縫消費(fèi)。
這是構(gòu)建分布式數(shù)據(jù)處理系統(tǒng)的經(jīng)典示例，僅使用Kafka+k8s組合即可確保并行處理。
使用 Python 創(chuàng)建 Kafka 消費(fèi)者的兩個(gè)非常流行的庫(kù)是：
- Python_Kafka
- Python
- from kafka import KafkaConsumer
- consumer = KafkaConsumer(TopicName,
- bootstrap_servers=
, - group_id=
, - enable_auto_commit=True,
- auto_offset_reset='earliest')
- consumer.poll()
- Confluent_Kafka
- Python
- from confluent_kafka import Consumer
- consumer = Consumer({'bootstrap.servers':
, - 'group.id':
, - 'enable.auto.commit': True,
- 'auto.offset.reset': 'earliest'
- })
- consumer.subscribe([TopicName])
K8.yml 文件的示例結(jié)構(gòu)如下：