Spark讀取kafka復(fù)雜嵌套json的最佳實踐，與其在大數(shù)據(jù)分析平臺中的應(yīng)用
2022-09-16 18:54:00
次
隨著互聯(lián)網(wǎng)的更進(jìn)一步發(fā)展，信息瀏覽、搜索以及電子商務(wù)、互聯(lián)網(wǎng)旅游生活產(chǎn)品等將生活中的流通環(huán)節(jié)在線化，對于實時性的要求進(jìn)一步提升，而信息的交互和溝通正在從點對點往信息鏈甚至信息網(wǎng)的方向發(fā)展，這樣必然帶來數(shù)據(jù)各個維度的交叉關(guān)聯(lián)，數(shù)據(jù)爆炸也不可避免，因此流式處理應(yīng)運而生，解決實時框架問題，助力大數(shù)據(jù)分析。
kafka是一個高性能的流式消息隊列，適用于大數(shù)據(jù)場景下的消息傳輸、消息處理和消息存儲，kafka可靠的傳遞能力讓它成為流式處理系統(tǒng)完美的數(shù)據(jù)來源，很多基于kafka構(gòu)建的流式處理系統(tǒng)都將kafka作為唯一可靠的數(shù)據(jù)來源。如Apache Storm、 Apache Spark Streaming 、Apache Flink 、Apache Samza 等。
json是kafka消息中比較常見的格式，對于單層json數(shù)據(jù)的讀取和解析相對簡單，但是在真實kafka流程處理的業(yè)務(wù)中，很多情況下都是json嵌套復(fù)雜格式消息。Spark1.1以后的版本存在一些實用的 SparkSQL函數(shù)，幫助解決復(fù)雜的json數(shù)據(jù)格式，實用函數(shù)包括get_json_object、from_json和explode等。
01、Spark框架中的基本概念和內(nèi)置函數(shù)
? RDD：Spark的基本計算單元，它是一個彈性可復(fù)原的分布式數(shù)據(jù)集。
? Dataframe：定義為指定到列的數(shù)據(jù)集（Dataset）。DFS類似于關(guān)系型數(shù)據(jù)庫中的表或者像R/Python 中的Dataframe ，可以說是一個具有良好優(yōu)化技術(shù)的關(guān)系表。
? Spark SQL：它是Spark的其中一個模塊，用于結(jié)構(gòu)化數(shù)據(jù)處理，Spark SQL提供的接口為Spark提供了有關(guān)數(shù)據(jù)結(jié)構(gòu)和正在執(zhí)行的計算的更多信息，Spark SQL會使用這些額外的信息來執(zhí)行額外的優(yōu)化。
? from_json：Spark SQL內(nèi)置的函數(shù)，從一個json 字符串中按照指定的schema格式抽取出來作為DataFrame的列，第一個參數(shù)為列名，以$"column_name"表示，第二個參數(shù)為定義的數(shù)據(jù)結(jié)構(gòu)
1）什么是復(fù)雜json?
json是一種輕量級的數(shù)據(jù)交換標(biāo)準(zhǔn)，具體以逗號分隔的key:value鍵值對的串形式，主要表現(xiàn)形式包括兩種：{對象},[數(shù)組]，其中key以字符串表達(dá)，value包括字符串、數(shù)值、boolean值、對象和數(shù)組（可嵌套）。在復(fù)雜的json數(shù)據(jù)格式中，通常json數(shù)據(jù)會有嵌套，每個層級的結(jié)構(gòu)不完全相同，value中不同類型進(jìn)行混合使用。
下圖為一份簡單json格式數(shù)據(jù)：
期望處理的結(jié)果為下圖的二維表，json串中的key(id，sepallength，sepalwidth，
petallength，petalwidth，label)作為二維表的列，value作為表的一行數(shù)據(jù)。
下圖為一份復(fù)雜json格式數(shù)據(jù)：
期望處理的結(jié)果為下圖的二維表，json串中單層key（id，createTime，deviceCode）和需要展開的數(shù)組trajectory中單個元素key（x，y）作為二維表的列，value是將數(shù)組trajectory中所有的元素展開成多行后，與其他列的數(shù)據(jù)進(jìn)行對齊。
2）整體思路
Kafka消費者收到復(fù)雜嵌套json消息后，一共有兩步。
第一步：首先把這批json字符消息轉(zhuǎn)換成分布式數(shù)據(jù)集RDD[String]中，再將RDD[String]轉(zhuǎn)換成列名為`json`的DataFrame，然后通過Spark SQL內(nèi)置函數(shù)get_json_object將json對象中的`id`、`createTime`、`deviceCode`、`data.trajectory`分別生成新列，并構(gòu)建一個包含這些列的新DataFrame；
第二步：獲取需要展開的列`data.trajectory`的schema（元數(shù)據(jù)信息），然后由SparkSQL內(nèi)置函數(shù)from_json將列`data.trajectory`的字符內(nèi)容轉(zhuǎn)換成數(shù)組對象，最后通過SparkSQL內(nèi)置函數(shù)explode將`data.trajectory`中的數(shù)組中每個元素展開成多行。
基于spark解析復(fù)雜json流程設(shè)計圖：
3）Spark讀取kafka復(fù)雜json消息解析核心代碼
json格式數(shù)據(jù)如果使用現(xiàn)有的工具，用戶常常需要開發(fā)出復(fù)雜的程序來讀寫分析系統(tǒng)中的json數(shù)據(jù)，Spark SQL對json數(shù)據(jù)的支持是從1.1版本開始發(fā)布，并且在Spark 1.2版本中進(jìn)行了加強(qiáng)。
下圖的代碼是通過Spark SQL內(nèi)置的json函數(shù)將復(fù)雜json轉(zhuǎn)換成一張二維表，并支持將json中數(shù)組數(shù)據(jù)進(jìn)行展開處理。
4）kafka復(fù)雜json解析在Tempo AI中的應(yīng)用
Tempo AI機(jī)器學(xué)習(xí)平臺將kafka數(shù)據(jù)作為數(shù)據(jù)挖掘分析標(biāo)準(zhǔn)數(shù)據(jù)源，既支持簡單的json解析，也支持復(fù)雜json解析，先進(jìn)行基礎(chǔ)配置讀取消息數(shù)據(jù)，查看消息內(nèi)容，然后進(jìn)行映射配置，將數(shù)據(jù)內(nèi)容與對應(yīng)元信息進(jìn)行匹配，最后可以預(yù)覽數(shù)據(jù)內(nèi)容。
基礎(chǔ)配置，包括連接配置和消息信息配置，如下圖所示：
在“消息內(nèi)容”頁面，查看提取的單條Kafka消息內(nèi)容，如下圖：
在“映射配置”頁面，根據(jù)左側(cè)預(yù)覽的消息內(nèi)容，通過點擊選擇左側(cè)的消息到右側(cè)，進(jìn)行映射配置，可以設(shè)置需要展開的數(shù)組，如下圖：
kafka輸入節(jié)點配置完成后，執(zhí)行AI流程，查看洞察信息，如下圖所示：
綜上，json是一種輕量級的數(shù)據(jù)交換格式，易于閱讀和編寫，目前是一種主流的數(shù)據(jù)格式，json字符串作為消息在kafka消息流中傳遞應(yīng)用很廣泛，通過Tempo 機(jī)器學(xué)習(xí)平臺封裝的Spark SQL解析復(fù)雜json的能力，極大簡化了使用json數(shù)據(jù)的終端的相關(guān)工作，使客戶更專注于自己的業(yè)務(wù)。
Tempo商業(yè)智能平臺
Tempo人工智能平臺
Tempo數(shù)據(jù)工廠平臺
Tempo指標(biāo)平臺
Tempo數(shù)據(jù)治理平臺
Tempo主數(shù)據(jù)管理平臺
首頁
美林?jǐn)?shù)據(jù)
行業(yè)資訊
陜公網(wǎng)安備 61019002000171號
