基于Spark的數(shù)據(jù)分析實踐
SparkSQL Flow 支持的Sourse

支持從 Hive 獲得數(shù)據(jù);
支持文件:JSON,TextFile(CSV),ParquetFile,AvroFile
支持RDBMS數(shù)據(jù)庫:PostgreSQL, MySQL,Oracle
支持 NOSQL 數(shù)據(jù)庫: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 為該文件映射的數(shù)據(jù)表名,可理解為數(shù)據(jù)的視圖;
Fields 為切分后的字段,使用逗號分隔,字段后可緊跟該字段的類型,使用冒號分隔;
Delimiter 為每行的分隔符;
Path 用于指定文件地址,可以是文件,也可是文件夾;
Path 指定地址需要使用協(xié)議,如: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 是從數(shù)據(jù)庫使用 JDBC讀取 數(shù)據(jù)集。支持 type 為:db、mysql、oracle、postgres、mssql;
tablename 為該數(shù)據(jù)表的抽象 table 名稱(視圖);
url、driver、user,password 為數(shù)據(jù)庫 JDBC 驅動信息,為必須字段;
SparkSQL 會加載該表的全表數(shù)據(jù),無法使用 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,相當于把該結果緩存到內(nèi)存中,緩存到內(nèi)存中的數(shù)據(jù)在后續(xù)其它 Transform 中使用能提高計算效率。但是需使用大量內(nèi)存,開發(fā)者需要評估該數(shù)據(jù)集能否放到內(nèi)存中,防止出現(xiàn) OutofMemory 的異常。
SparkSQL Flow Targets
SparkSQL Flow Targets 支持輸出數(shù)據(jù)到一個或者多個目標。這些目標,基本覆蓋了 Source 包含的外部系統(tǒng)。下面以 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 會根據(jù) DataFrame 的數(shù)據(jù)類型自動創(chuàng)建表;
savemode 默認為 overwrite 覆蓋寫入,當寫入目標已存在時刪除源表再寫入;支持 append 模式, 可增量寫入。
Target 有一個特殊的 show 類型的 target。用于直接在控制臺輸出一個 DataFrame 的結果到控制臺(print),該 target 用于開發(fā)和測試。
<target type="show" table_name="cust_id_agmt_id_t" rows=”10000”/>
可左右滑動查看代碼
Rows 用于控制輸出多少行數(shù)據(jù)。
SparkSQL Around
After 用于 Flow 在運行結束后執(zhí)行的一個環(huán)繞,用于記錄日志和寫入狀態(tài)。類似 Java 的 try {} finally{ round.execute() }
多個 round 一定會執(zhí)行,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的執(zhí)行效果

Prepare round 可做插入(insert)動作,after round 可做更新 (update)動作,相當于在數(shù)據(jù)庫表中從執(zhí)行開始到結束有了完整的日志記錄。SparkSQL Flow 會保證round 一定能被執(zhí)行,而且 round 的執(zhí)行不影響任務的狀態(tài)。
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
可左右滑動查看代碼

接收必須的參數(shù) –f,可選的參數(shù)為支持 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 中重要的數(shù)據(jù)表操作。
regiserDataFrameAsTable 是每個 source 和 Transform 的數(shù)據(jù)在 SparkSQL 中的數(shù)據(jù)視圖,每個視圖都會在 SparkContex 中注冊一次。
請輸入評論內(nèi)容...
請輸入評論/評論長度6~500個字
最新活動更多
- 1 AI狂歡遇上油價破百,全球股市還能漲多久? | 產(chǎn)聯(lián)看全球
- 2 OpenAI深夜王炸!ChatGPT Images 2.0實測:中文穩(wěn)、細節(jié)炸,設計師慌了
- 3 6000億美元估值錨定:字節(jié)跳動的“去單一化”突圍與估值重構
- 4 Tesla AI5芯片最新進展總結
- 5 連夜測了一波DeepSeek-V4,我發(fā)現(xiàn)它可能只剩“審美”這個短板了
- 6 熱點丨AI“瑜亮之爭”:既生OpenClaw,何生Hermes?
- 7 AI界的殺豬盤:9秒刪庫跑路,全員被封號,還繼續(xù)扣錢!
- 8 2026,人形機器人只贏了面子
- 9 DeepSeek降價90%:價格屠夫不是身份,是戰(zhàn)略
- 10 AI Infra產(chǎn)業(yè)鏈卡在哪里了?
- 高級軟件工程師 廣東省/深圳市
- 自動化高級工程師 廣東省/深圳市
- 光器件研發(fā)工程師 福建省/福州市
- 銷售總監(jiān)(光器件) 北京市/海淀區(qū)
- 激光器高級銷售經(jīng)理 上海市/虹口區(qū)
- 光器件物理工程師 北京市/海淀區(qū)
- 激光研發(fā)工程師 北京市/昌平區(qū)
- 技術專家 廣東省/江門市
- 封裝工程師 北京市/海淀區(qū)
- 結構工程師 廣東省/深圳市


分享













