前言
目前公司的批開發框架基於spark, 流式處理基於structure streaming和spark sql。
目前整體開發流程與大部分流式開發平台想法一致，將數據的ETL操作分割成獨立功能的operator，其中采坑的部分與SqlOperator有關。
顧名思義，此operator的作用是將流數據映射成動態表，使用sql的方式進行操作。
主要技術框架背景介紹
spark ： 2.4.3
scala ： 2.11.12
問題演示過程
mock 數據
{"id":10000000,"json_str":"[{\"uid\":\"1000001\",\"name\":\"test-name-1\",\"type\":\"test-type-1\",\"kw_idxs\":[[[0, 0]]],\"text_idxs\":[[0, 0]]},{\"id\":\"1000002\",\"name\":\"test-name-2\",\"type\":\"test-type-2\",\"kw_idxs\":[[[0, 0]]],\"text_idxs\":[[0, 0]]}]"}
操作過程
structure streaming 消費kafka topic的數據，配合schema register，讀取數據，上文為mock的測試數據。
可以看出，json_str 數據類型為json array格式的string類型。
目標是抽取json_str的id字段，組成所有id字段組成的數組。
所以需要先將json字符串轉化成json和struct結構類型方便下一步操作。
查看spark sql 對應版本json相關函數，注意到schema_of_json函數。
https://spark.apache.org/docs/2.4.3/api/sql/
查看文檔突然眼前一亮，看起來這個函數可以完美解決我的需求。動手測試！
將數據放入json文件中，使用 spark.read.json的方式進行讀取，結果如下：
這個就很奇怪了，查閱了很多文檔，大多數都是在將DF操作數據流的，沒有針對於spark sql中此函數更好的解釋。
所以另外做了一個測試，將mock數據直接放入函數中，不在文件中讀取。
結果居然可以正常解析？？數據是相同的，為什么會這樣呢？
查閱相關文檔發現，其實這個是有人提過相關問題：
https://github.com/apache/spark/pull/22775
然而2.4.3的版本看來並沒有修復此問題。
目前還未驗證3.0是否也存在相同問題。
解決方法
使用from_json方法將json格式數據轉化為結構化數據類型。
可以使用schema_of_json方法先解析靜態數據，可以在console中看到相應的數據結構類型，在此示例中為：
array<struct<id:string,ks:array<array<array<bigint>>>,name:string,ts:array<array<bigint>>,type:string,uid:string>>
然后使用from_json函數進行解析：
1 select 2 from_json(json_str,'array<struct<id:string,ks:array<array<array<bigint>>>,name:string,ts:array<array<bigint>>,type:string,uid:string>>') as json_obj 3 from 4 test
這樣就可以將任意json復雜結構轉化為結構化數據結構。
總結
不得不吐槽下spark sql的官方文檔，過於簡潔，沒有更多的test case，導致花了很多時間在踩坑。
例如from_json example ：
> SELECT from_json('{"a":1, "b":0.8}', 'a INT, b DOUBLE');
{"a":1,"b":0.8}
> SELECT from_json('{"time":"26/08/2015"}', 'time Timestamp', map('timestampFormat', 'dd/MM/yyyy'));
{"time":2015-08-26 00:00:00}
其實在復雜數據結構中，數據類型定義會存在有冒號的情況，但是文檔中並沒有給出可能存在的情況，使用schema_of_json的方式獲取結構也是沒辦法確定正確格式到底是什么才使用的方法，在新上手時，如果不使用schema_of_json方法獲取的話，手寫很難寫對。