SparkSQL Flow 支持的Sourse

支持從 Hive 獲得數據；

支持文件：JSON，TextFile（CSV），ParquetFile，AvroFile

支持RDBMS數據庫：PostgreSQL， MySQL，Oracle

支持 NOSQL 數據庫：Hbase，MongoDB

SparkSQL Flow TextFile Source

textfile 為讀取文本文件，把文本文件每行按照 delimiter 指定的字符進行切分，切分不夠的列使用 null 填充。

＜source type＝＂textfile＂ table＿name＝＂et＿rel＿pty＿cong＂              fields＝＂cust＿id，name1，gender1，age1：int＂               delimiter＝＂，＂              path＝＂file：／／／Users／zhenqin／software／hive／user．txt＂／＞

可左右滑動查看代碼

Tablename 為該文件映射的數據表名，可理解為數據的視圖；

Fields 為切分后的字段，使用逗號分隔，字段后可緊跟該字段的類型，使用冒號分隔；

Delimiter 為每行的分隔符；

Path 用于指定文件地址，可以是文件，也可是文件夾；

Path 指定地址需要使用協議，如：file：／／ 、 hdfs：／／，否則跟 core－site．xml 配置密切相關；

SparkSQL Flow DB Source

＜source type＝＂mysql＂ table＿name＝＂et＿rel＿pty＿cong＂                table＝＂user＂                url＝＂jdbc：mysql：／／localhost：3306／tdb？characterEncoding＝UTF－8＂                driver＝＂com．mysql．jdbc．Driver＂                user＝＂root＂ password＝＂123456＂／＞

可左右滑動查看代碼

RDBMS 是從數據庫使用 JDBC讀取 數據集。支持 type 為：db、mysql、oracle、postgres、mssql；

tablename 為該數據表的抽象 table 名稱（視圖）；

url、driver、user，password 為數據庫 JDBC 驅動信息，為必須字段；

SparkSQL 會加載該表的全表數據，無法使用 where 條件。

SparkSQL Flow Transformer

＜transform type＝＂sql＂ table＿name＝＂cust＿id＿agmt＿id＿t＂ cached＝＂true＂＞            SELECT c＿phone，c＿type，c＿num， CONCAT＿VAL（cust＿id） as cust＿ids            FROM user＿concat＿testx            group by c＿phone，c＿type，c＿num＜／transform＞

可左右滑動查看代碼

Transform 支持 cached 屬性，默認為 false；如果設置為 true，相當于把該結果緩存到內存中，緩存到內存中的數據在后續其它 Transform 中使用能提高計算效率。但是需使用大量內存，開發者需要評估該數據集能否放到內存中，防止出現 OutofMemory 的異常。

SparkSQL Flow Targets

SparkSQL Flow Targets 支持輸出數據到一個或者多個目標。這些目標，基本覆蓋了 Source 包含的外部系統。下面以 Hive 舉例說明：

＜target type＝＂hive＂ table＿name＝＂cust＿id＿agmt＿id＿t＂  savemode＝”append”target＿table＿name＝＂cust＿id＿agmt＿id＿h＂／＞

可左右滑動查看代碼

table＿name 為 source 或者 Transform 定義的表名稱；

target＿table＿name 為 hive 中的表結果，Hive 表可不存在也可存在，sparksql 會根據 DataFrame 的數據類型自動創建表；

savemode 默認為 overwrite 覆蓋寫入，當寫入目標已存在時刪除源表再寫入；支持 append 模式， 可增量寫入。

Target 有一個特殊的 show 類型的 target。用于直接在控制臺輸出一個 DataFrame 的結果到控制臺（print），該 target 用于開發和測試。

＜target type＝＂show＂ table＿name＝＂cust＿id＿agmt＿id＿t＂ rows＝”10000”／＞

可左右滑動查看代碼

Rows 用于控制輸出多少行數據。

SparkSQL Around

After 用于 Flow 在運行結束后執行的一個環繞，用于記錄日志和寫入狀態。類似 Java 的 try ｛｝ finally｛ round．execute（） ｝

多個 round 一定會執行，round 異常不會導致任務失敗。

＜prepare＞        ＜round type＝＂mysql＂               sql＝＂insert into cpic＿task＿history（id， task＿type， catalog＿model， start＿time， retry＿count， final＿status， created＿at）               values（＄｛uuid｝， ＄｛task．type｝， ＄｛catalog．model｝， ＄｛starttime｝， 0， ＄｛status｝， now（））＂               url＝＂＄｛jdbc．url｝＂ ．．．／＞＜／prepare＞＜after＞        ＜round type＝＂mysql＂               sql＝＂update cpic＿task＿history set               end＿time ＝ ＄｛endtime｝， final＿status ＝ ＄｛status｝， error＿text ＝ ＄｛error｝ where id ＝ ＄｛uuid｝＂               url＝＂＄｛jdbc．url｝”…／＞＜／after＞

可左右滑動查看代碼

Prepare round 和 after round 配合使用可用于記錄 SparkSQL Flow 任務的運行日志。

SparkSQL Around可使用的變量

SparkSQL Around的執行效果

Prepare round 可做插入（insert）動作，after round 可做更新 （update）動作，相當于在數據庫表中從執行開始到結束有了完整的日志記錄。SparkSQL Flow 會保證round 一定能被執行，而且 round 的執行不影響任務的狀態。

SparkSQL Flow 提交

bin／spark－submit －－master yarn－client －－driver－memory 1G －－num－executors 10 －－executor－memory 2G －－jars ／lib／jsoup－1．11．3．jarlib／jsqlparser－0．9．6．jar，／lib／mysql－connector－java－5．1．46．jar －－conf spark．yarn．jars＝hdfs：／／／lib／spark2／＊．jar －－queue default －－name FlowTest etl－flow－0．2．0．jar －f hive－flow－test．xml

可左右滑動查看代碼

接收必須的參數 –f，可選的參數為支持 Kerberos 認證的租戶名稱principal，和其認證需要的密鑰文件。

usage： spark－submit －－jars etl－flow．jar －－class                    com．yiidata．etl．flow．source．FlowRunner －f，－－xml－file ＜arg＞     Flow XML File Path    －－keytabFile ＜arg＞   keytab File Path（Huawei）    －－krb5File ＜arg＞     krb5 File Path（Huawei）    －－principal ＜arg＞    principal for hadoop（Huawei）

可左右滑動查看代碼

SparkSQL Execution Plan

每個Spark Flow 任務本質上是一連串的 SparkSQL 操作，在 SparkUI SQL tab 里可以看到 flow 中重要的數據表操作。

regiserDataFrameAsTable 是每個 source 和 Transform 的數據在 SparkSQL 中的數據視圖，每個視圖都會在 SparkContex 中注冊一次。