基于Spark的數(shù)據(jù)分析實(shí)踐
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 指定的字符進(jìn)行切分,切分不夠的列使用 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 配置密切相關(guān);
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 驅(qū)動信息,為必須字段;
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 屬性,默認(rèn)為 false;如果設(shè)置為 true,相當(dāng)于把該結(jié)果緩存到內(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ù)到一個或者多個目標(biāo)。這些目標(biāo),基本覆蓋了 Source 包含的外部系統(tǒng)。下面以 Hive 舉例說明:
<target type="hive" table_name="cust_id_agmt_id_t" savemode=”append”target_table_name="cust_id_agmt_id_h(yuǎn)"/>
可左右滑動查看代碼
table_name 為 source 或者 Transform 定義的表名稱;
target_table_name 為 hive 中的表結(jié)果,Hive 表可不存在也可存在,sparksql 會根據(jù) DataFrame 的數(shù)據(jù)類型自動創(chuàng)建表;
savemode 默認(rèn)為 overwrite 覆蓋寫入,當(dāng)寫入目標(biāo)已存在時刪除源表再寫入;支持 append 模式, 可增量寫入。
Target 有一個特殊的 show 類型的 target。用于直接在控制臺輸出一個 DataFrame 的結(jié)果到控制臺(print),該 target 用于開發(fā)和測試。
<target type="show" table_name="cust_id_agmt_id_t" rows=”10000”/>
可左右滑動查看代碼
Rows 用于控制輸出多少行數(shù)據(jù)。
SparkSQL Around
After 用于 Flow 在運(yùn)行結(jié)束后執(zhí)行的一個環(huán)繞,用于記錄日志和寫入狀態(tài)。類似 Java 的 try {} finally{ round.execute() }
多個 round 一定會執(zhí)行,round 異常不會導(dǎo)致任務(wù)失敗。
<prepare> <round type="mysql" sql="insert into cpic_task_h(yuǎn)istory(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_h(yuǎn)istory set end_time = ${endtime}, final_status = ${status}, error_text = ${error} where id = ${uuid}" url="${jdbc.url}”…/></after>
可左右滑動查看代碼
Prepare round 和 after round 配合使用可用于記錄 SparkSQL Flow 任務(wù)的運(yùn)行日志。
SparkSQL Around可使用的變量
SparkSQL Around的執(zhí)行效果
Prepare round 可做插入(insert)動作,after round 可做更新 (update)動作,相當(dāng)于在數(shù)據(jù)庫表中從執(zhí)行開始到結(jié)束有了完整的日志記錄。SparkSQL Flow 會保證round 一定能被執(zhí)行,而且 round 的執(zhí)行不影響任務(wù)的狀態(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 認(rèn)證的租戶名稱principal,和其認(rèn)證需要的密鑰文件。
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 任務(wù)本質(zhì)上是一連串的 SparkSQL 操作,在 SparkUI SQL tab 里可以看到 flow 中重要的數(shù)據(jù)表操作。
regiserDataFrameAsTable 是每個 source 和 Transform 的數(shù)據(jù)在 SparkSQL 中的數(shù)據(jù)視圖,每個視圖都會在 SparkContex 中注冊一次。

發(fā)表評論
請輸入評論內(nèi)容...
請輸入評論/評論長度6~500個字
最新活動更多
-
7月22-29日立即報名>> 【線下論壇】第三屆安富利汽車生態(tài)圈峰會
-
7.30-8.1火熱報名中>> 全數(shù)會2025(第六屆)機(jī)器人及智能工廠展
-
7月31日免費(fèi)預(yù)約>> OFweek 2025具身智能機(jī)器人產(chǎn)業(yè)技術(shù)創(chuàng)新應(yīng)用論壇
-
免費(fèi)參會立即報名>> 7月30日- 8月1日 2025全數(shù)會工業(yè)芯片與傳感儀表展
-
即日-2025.8.1立即下載>> 《2024智能制造產(chǎn)業(yè)高端化、智能化、綠色化發(fā)展藍(lán)皮書》
-
8月5日立即報名>> 【在線會議】CAE優(yōu)化設(shè)計:醫(yī)療器械設(shè)計的應(yīng)用案例與方案解析
推薦專題
- 1 AI產(chǎn)業(yè)的新高度!英偉達(dá)成為全球首家市值破4萬億美元的公司
- 2 一文讀懂:到底什么是 “具身智能” ?
- 3 豆包前負(fù)責(zé)人喬木出軌BP后續(xù):均被辭退
- 4 “支付+”時代,支付即生態(tài) | 2025中國跨境支付十大趨勢
- 5 中國最具實(shí)力AI公司TOP10
- 6 深圳跑出40億超級隱形冠軍:賣機(jī)器人年入6.1億,港股上市
- 7 特斯拉Robotaxi上路,馬斯克端上畫了十年的餅
- 8 “AI六小虎”到了下一個賽點(diǎn)
- 9 北京,跑出150億超級獨(dú)角獸:物流機(jī)器人全球第一,揭示3大機(jī)會
- 10 自動駕駛中常提的VLA是個啥?