使用Spark Streaming轉換不同的JSON有效負載
譯文【51CTO.com快譯】Spark Streaming 是底層基于 Spark Core 的對大數據進行實時計算的框架，可以流方式從源讀取數據。只需要從數據源創建一個讀取流，然后我們可以創建寫入流以將數據加載到目標數據源中。
接下來的演示，將假設我們有不同的 JSON 有效負載進入一個 kafka 主題，我們需要將其轉換并寫入另一個 kafka 主題。
創建一個ReadStream
為了能連續接收JSON有效負載作為消息。我們需要首先讀取消息并使用spark的readstream創建數據幀。Spark 中提供了 readStream 函數，我們可以使用這個函數基本上創建一個 readStream。這將從 kafka 主題中讀取流負載。
val df = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "topic1")
.load()
我們可以創建一個 case-class(例如CustomerUnion)，它將包含JSON有效負載的所有可能字段。這樣，我們就能在數據幀上運行select查詢而不會失敗。
val rawDfValue = rawData.selectExpr("CAST(value AS STRING)").as[String]
val schema = ScalaReflection.schemaFor[CustomerUnion].dataType.asInstanceOf[StructType]
val extractedDFWithSchema = rawDfValue.select(from_json(col("value"), schema).as("data")).select("data.*")
extractedDFWithSchema.createOrReplaceTempView(“tempView”)這將為我們提供一個數據幀提取的 DFWithSchema，其中包含作為有效負載字段的列。
示例輸入負載
這是兩個樣本輸入有效負載，但也可以有更多的有效負載，有些字段不存在(變量)。
{
“id”: 1234,
“firstName”:”Jon”,
“lastName”:”Butler”,
“City”:”Newyork”,
“Email”:abc@gmail.com,
“Phone”:”2323123”
}{
“firstName”:”Jon”,
“lastName”:”Butler”,
“City”:”Newyork”,
“Email”:abc@gmail.com,
“Phone”:”2323123”
}樣例輸出負載
根據id字段，我們將決定輸出有效負載。如果存在一個 id 字段，我們將把它視為一個用戶更新案例，并且在輸出有效負載中只發送“Email”和“Phone”。我們可以根據某些條件配置任何字段。這只是一個例子。
如果 id 不存在，我們將發送所有字段。下面是兩個輸出載荷的示例：
{
“userid”: 1234,
“Email”:abc@gmail.com,
“Phone”:”2323123”
}{
“fullname”:”Jon Butler”,
“City”:”Newyork”,
“Email”:abc@gmail.com,
“Phone”:”2323123”
}開始WriteStreams
一旦我們有了數據幀，我們就可以運行盡可能多的sql查詢，并根據所需的有效負載寫入 kafka 主題。因此，我們可以創建一個包含所有sql查詢的列表，并通過該列表進行循環，并調用writeStream函數。讓我們假設，我們有一個名為 queryList 的列表，它只包含字符串(即sql查詢)。
下面為寫入流定義的一個函數：
def startWriteStream(query: String): Unit = {
val transformedDf = spark.sql(query)
transformedDf
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "topic1")
.start()
}這將啟動列表中每個查詢的寫入流。
queryList.foreach(startWriteStream)
spark.streams.awaitAnyTermination()
如果我們知道輸入有效負載的所有可能字段，那么即使有一些字段不存在，我們的sql查詢也不會失敗。我們已經將有效負載的模式指定為case-class，它將為缺席字段創建指定 NULL 的數據幀。
通過這種方式，我們可以使用 spark-streaming 在所需的轉換/過濾器之后將多個有效負載從同一主題寫入不同的主題。
【51CTO譯稿，合作站點轉載請注明原文譯者和出處為51CTO.com】
