基于 Spark 的數(shù)據(jù)分析實(shí)踐是怎樣進(jìn)行的

今天就跟大家聊聊有關(guān)基于 Spark 的數(shù)據(jù)分析實(shí)踐是怎樣進(jìn)行的，可能很多人都不太了解，為了讓大家更加了解，小編給大家總結(jié)了以下內(nèi)容，希望大家根據(jù)這篇文章可以有所收獲。

引言：

val lines = List(“A”, “B”, “C”, “D” …)
val rdd:RDD = sc.parallelize(lines);

val rdd:RDD[String] = sc.textFile(“hdfs://path/filename”)

public abstract class InputFormat<K,V>{
  public abstract List<InputSplit> getSplits(JobContextcontext
                               ) throwsIOException,InterruptedException;
  
  public abstract RecordReader<K,V> createRecordReader(InputSplitsplit,
                                         TaskAttemptContextcontext
                                        )throwsIOException,InterruptedException;
}

Spark 從 1.3 版本開(kāi)始原有 SchemaRDD 的基礎(chǔ)上提供了類(lèi)似Pandas DataFrame API。新的DataFrame API不僅可以大幅度降低普通開(kāi)發(fā)者的學(xué)習(xí)門(mén)檻，同時(shí)還支持Scala、Java與Python三種語(yǔ)言。更重要的是，由于脫胎自SchemaRDD，DataFrame天然適用于分布式大數(shù)據(jù)場(chǎng)景。

一般的數(shù)據(jù)處理步驟：讀入數(shù)據(jù) -> 對(duì)數(shù)據(jù)進(jìn)行處理 -> 分析結(jié)果  -> 寫(xiě)入結(jié)果

SparkSQL 結(jié)構(gòu)化數(shù)據(jù)

• 處理結(jié)構(gòu)化數(shù)據(jù)(如 CSV，JSON，Parquet 等);

• 把已經(jīng)結(jié)構(gòu)化數(shù)據(jù)抽象成 DataFrame （HiveTable）；

• 非結(jié)構(gòu)化數(shù)據(jù)通過(guò) RDD.map.filter 轉(zhuǎn)換成結(jié)構(gòu)化進(jìn)行處理；

• 按照列式數(shù)據(jù)庫(kù)，只加載非結(jié)構(gòu)化中可結(jié)構(gòu)化的部分列（Hbase，MongoDB）；

處理非結(jié)構(gòu)化數(shù)據(jù)，不能簡(jiǎn)單的用 DataFrame 裝載。而是要用 SparkRDD 把數(shù)據(jù)讀入，在通過(guò)一系列的 Transformer Method 把非結(jié)構(gòu)化的數(shù)據(jù)加工為結(jié)構(gòu)化，或者過(guò)濾到不合法的數(shù)據(jù)。

SparkSQL DataFrame

SparkSQL 中一切都是 DataFrame，all in DataFrame. DataFrame是一種以RDD為基礎(chǔ)的分布式數(shù)據(jù)集，類(lèi)似于傳統(tǒng)數(shù)據(jù)庫(kù)中的二維表格。DataFrame與RDD的主要區(qū)別在于，前者帶有schema元信息，即DataFrame所表示的二維表數(shù)據(jù)集的每一列都帶有名稱(chēng)和類(lèi)型。如果熟悉 Python Pandas 庫(kù)中的 DataFrame 結(jié)構(gòu)，則會(huì)對(duì) SparkSQL DataFrame 概念非常熟悉。

TextFile DataFrame

import.org.apache.spark.sql._
//定義數(shù)據(jù)的列名稱(chēng)和類(lèi)型
valdt=StructType(List(id:String,name:String,gender:String,age:Int))

//導(dǎo)入user_info.csv文件并指定分隔符
vallines = sc.textFile("/path/user_info.csv").map(_.split(","))

//將表結(jié)構(gòu)和數(shù)據(jù)關(guān)聯(lián)起來(lái)，把讀入的數(shù)據(jù)user.csv映射成行,構(gòu)成數(shù)據(jù)集
valrowRDD = lines.map(x=>Row(x(0),x(1),x(2),x(3).toInt))

//通過(guò)SparkSession.createDataFrame()創(chuàng)建表,并且數(shù)據(jù)表表頭
val df= spark.createDataFrame(rowRDD, dt)

讀取規(guī)則數(shù)據(jù)文件作為DataFrame

SparkSession.Builder builder = SparkSession.builder()
Builder.setMaster("local").setAppName("TestSparkSQLApp")
SparkSession spark = builder.getOrCreate();
SQLContext sqlContext = spark.sqlContext();

# 讀取 JSON 數(shù)據(jù)，path 可為文件或者目錄
valdf=sqlContext.read().json(path);

# 讀取 HadoopParquet 文件
vardf=sqlContext.read().parquet(path);

# 讀取 HadoopORC 文件
vardf=sqlContext.read().orc(path);

JSON 文件為每行一個(gè) JSON 對(duì)象的文件類(lèi)型，行尾無(wú)須逗號(hào)。文件頭也無(wú)須[]指定為數(shù)組；SparkSQL 讀取是只是按照每行一條 JSON Record序列化；

Parquet文件

Configurationconfig = new Configuration();
ParquetFileReaderreader = ParquetFileReader.open(
        HadoopInputFile.fromPath(new Path("hdfs:///path/file.parquet"),conf));
Map<String, String>schema = reader.getFileMetaData().getKeyValueMetaData();
String allFields= schema.get("org.apache.spark.sql.parquet.row.metadata");

allFiedls 的值就是各字段的名稱(chēng)和具體的類(lèi)型，整體是一個(gè)json格式進(jìn)行展示。

讀取 Hive 表作為 DataFrame

Spark2 API 推薦通過(guò) SparkSession.Builder 的 Builder 模式創(chuàng)建 SparkContext。 Builder.getOrCreate() 用于創(chuàng)建 SparkSession，SparkSession 是 SparkContext 的封裝。

在Spark1.6中有兩個(gè)核心組件SQLcontext和HiveContext。SQLContext 用于處理在 SparkSQL 中動(dòng)態(tài)注冊(cè)的表，HiveContext 用于處理 Hive 中的表。

從Spark2.0以上的版本開(kāi)始，spark是使用全新的SparkSession接口代替Spark1.6中的SQLcontext和HiveContext。SQLContext.sql 即可執(zhí)行 Hive 中的表，也可執(zhí)行內(nèi)部注冊(cè)的表；

在需要執(zhí)行 Hive 表時(shí)，只需要在 SparkSession.Builder 中開(kāi)啟 Hive 支持即可（enableHiveSupport()）。

SparkSession.Builder builder = SparkSession.builder().enableHiveSupport();
SparkSession spark = builder.getOrCreate();
SQLContext sqlContext = spark.sqlContext();

// db 指 Hive 庫(kù)中的數(shù)據(jù)庫(kù)名，如果不寫(xiě)默認(rèn)為 default

// tableName 指 hive 庫(kù)的數(shù)據(jù)表名

sqlContext.sql(“select * from db.tableName”)

SparkSQL ThriftServer

//首先打開(kāi) Hive 的 Metastore服務(wù)

hive$bin/hive –-service metastore –p 8093

//把 Spark 的相關(guān) jar 上傳到hadoophdfs指定目錄，用于指定sparkonyarn的依賴(lài) jar

spark$hadoop fs –put jars/*.jar /lib/spark2

// 啟動(dòng) spark thriftserver 服務(wù)

spark$ sbin/start-thriftserver.sh --master yarn-client --driver-memory 1G --conf 
spark.yarn.jars=hdfs:///lib/spark2/*.jar

當(dāng)hdfs 上傳了spark 依賴(lài) jar 時(shí)，通過(guò)spark.yarn.jars 可看到日志 spark 無(wú)須每個(gè)job 都上傳jar，可節(jié)省啟動(dòng)時(shí)間

19/06/1114:08:26 INFO Client: Source and destination file systems are the same. Notcopying hdfs://localhost:9000/lib/spark2/snappy-java-1.0.5.jar
19/06/1114:08:26 INFO Client: Source and destination file systems are the same. Notcopying hdfs://localhost:9000/lib/spark2/snappy-java-1.1.7.3.jar

//通過(guò) spark bin 下的 beeline 工具，可以連接到 spark ThriftServer（SparkOnHive）

bin/beeline -u jdbc:hive2://ip:10000/default -n hadoop

-u 是指定 beeline 的執(zhí)行驅(qū)動(dòng)地址；

-n 是指定登陸到 spark Session 上的用戶(hù)名稱(chēng)；

Beeline 還支持傳入-e 可傳入一行 SQL， 

-e <query>                      query that should be executed

也可通過(guò) –f 指定一個(gè) SQL File，內(nèi)部可用逗號(hào)分隔的多個(gè) SQL（存儲(chǔ)過(guò)程）

-f <exec file>                  script file that should be executed

SparkSQL Beeline 的執(zhí)行效果展示

SparkSQL ThriftServer

對(duì)于 SparkSQL ThriftServer 服務(wù)，每個(gè)登陸的用戶(hù)都有創(chuàng)建的 SparkSession，并且執(zhí)行的對(duì)個(gè) SQL 會(huì)通過(guò)時(shí)間順序列表展示。

SparkSQL ThriftServer 服務(wù)可用于其他支持的數(shù)據(jù)庫(kù)工具創(chuàng)建查詢(xún)，也用于第三方的 BI 工具，如 tableau。

SparkSQL Flow 是以 SparkSQL 為基礎(chǔ)，開(kāi)發(fā)的統(tǒng)一的基于 XML 配置化的可執(zhí)行一連串的 SQL 操作，這一連串的 SQL 操作定義為一個(gè) Flow。下文開(kāi)始 SparkSQL Flow 的介紹：

SparkSQL Flow 是基于 SparkSQL 開(kāi)發(fā)的一種基于 XML 配置化的 SQL 數(shù)據(jù)流轉(zhuǎn)處理模型。該模型簡(jiǎn)化了 SparkSQL 、Spark RDD的開(kāi)發(fā)，并且降低開(kāi)發(fā)了難度，適合了解數(shù)據(jù)業(yè)務(wù)但無(wú)法駕馭大數(shù)據(jù)以及 Spark 技術(shù)的開(kāi)發(fā)者。

• 一個(gè)由普元技術(shù)部提供的基于 SparkSQL 的開(kāi)發(fā)模型;

• 一個(gè)可二次定制開(kāi)發(fā)的大數(shù)據(jù)開(kāi)發(fā)框架，提供了靈活的可擴(kuò)展 API；

• 一個(gè)提供了 對(duì)文件，數(shù)據(jù)庫(kù)，NoSQL 等統(tǒng)一的數(shù)據(jù)開(kāi)發(fā)視界語(yǔ)義；

• 基于 SQL 的開(kāi)發(fā)語(yǔ)言和 XML 的模板配置，支持 Spark UDF 的擴(kuò)展管理；

• 支持基于 Spark Standlone，Yarn，Mesos 資源管理平臺(tái)；

• 支持開(kāi)源、華為、星環(huán)等平臺(tái)統(tǒng)一認(rèn)證。

SparkSQL Flow 適合的場(chǎng)景：

1. 批量 ETL；

2. 非實(shí)時(shí)分析服務(wù)；

SparkSQL Flow XML 概覽

1. Properties 內(nèi)定義一組變量，可用于宏替換；

2. Methods 內(nèi)可注冊(cè) udf 和 udaf 兩種函數(shù)；

3. Prepare 內(nèi)可定義前置 SQL，用于執(zhí)行 source 前的 sql 操作；

4. Sources 內(nèi)定義一個(gè)到多個(gè)數(shù)據(jù)表視圖；

5. Transformer 內(nèi)可定義 0 到多個(gè)基于 SQL 的數(shù)據(jù)轉(zhuǎn)換操作（支持 join）；

6. Targets 用于定義 1 到多個(gè)數(shù)據(jù)輸出；

7. After 可定義 0到多個(gè)任務(wù)日志；

如你所見(jiàn)，source 的 type 參數(shù)用于區(qū)分 source 的類(lèi)型，source 支持的種類(lèi)直接決定SparkSQL Flow 的數(shù)據(jù)源加載廣度；并且，根據(jù) type 不同，source 也需要配置不同的參數(shù)，如數(shù)據(jù)庫(kù)還需要 driver，url，user和 password 參數(shù)。

Transformer 是基于 source 定的數(shù)據(jù)視圖可執(zhí)行的一組轉(zhuǎn)換 SQL，該 SQL 符合 SparkSQL 的語(yǔ)法（SQL99）。Transform 的 SQL 的執(zhí)行結(jié)果被作為中間表命名為 table_name 指定的值。

Targets 為定義輸出，table_name 的值需在 source 或者 Transformer 中定義。

SparkSQL Flow 支持的Sourse

• 支持從 Hive 獲得數(shù)據(jù)；

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

• 支持RDBMS數(shù)據(jù)庫(kù)：PostgreSQL， MySQL，Oracle
  

• 支持 NOSQL 數(shù)據(jù)庫(kù)：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"/>

1. Tablename 為該文件映射的數(shù)據(jù)表名，可理解為數(shù)據(jù)的視圖；

2. Fields 為切分后的字段，使用逗號(hào)分隔，字段后可緊跟該字段的類(lèi)型，使用冒號(hào)分隔；

3. Delimiter 為每行的分隔符；

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

5. 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ù)庫(kù)使用 JDBC讀取 數(shù)據(jù)集。支持 type 為：db、mysql、oracle、postgres、mssql；

1. tablename 為該數(shù)據(jù)表的抽象 table 名稱(chēng)(視圖）；

2. url、driver、user，password 為數(shù)據(jù)庫(kù) JDBC 驅(qū)動(dòng)信息，為必須字段；

3. SparkSQL 會(huì)加載該表的全表數(shù)據(jù)，無(wú)法使用 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 中使用能提高計(jì)算效率。但是需使用大量?jī)?nèi)存，開(kāi)發(fā)者需要評(píng)估該數(shù)據(jù)集能否放到內(nèi)存中，防止出現(xiàn) OutofMemory 的異常。

SparkSQL Flow Targets

SparkSQL Flow Targets 支持輸出數(shù)據(jù)到一個(gè)或者多個(gè)目標(biāo)。這些目標(biāo)，基本覆蓋了 Source 包含的外部系統(tǒng)。下面以 Hive 舉例說(shuō)明：

<target type="hive"
 table_name="cust_id_agmt_id_t" 
 savemode=”append”
target_table_name="cust_id_agmt_id_h"/>

1. table_name 為 source 或者 Transform 定義的表名稱(chēng)；

2. target_table_name 為 hive 中的表結(jié)果，Hive 表可不存在也可存在，sparksql 會(huì)根據(jù) DataFrame 的數(shù)據(jù)類(lèi)型自動(dòng)創(chuàng)建表；

3. savemode 默認(rèn)為 overwrite 覆蓋寫(xiě)入，當(dāng)寫(xiě)入目標(biāo)已存在時(shí)刪除源表再寫(xiě)入；支持 append 模式， 可增量寫(xiě)入。

Target 有一個(gè)特殊的 show 類(lèi)型的 target。用于直接在控制臺(tái)輸出一個(gè) DataFrame 的結(jié)果到控制臺(tái)（print），該 target 用于開(kāi)發(fā)和測(cè)試。

<target type="show" table_name="cust_id_agmt_id_t" rows=”10000”/>

Rows 用于控制輸出多少行數(shù)據(jù)。

SparkSQL Around

After 用于 Flow 在運(yùn)行結(jié)束后執(zhí)行的一個(gè)環(huán)繞，用于記錄日志和寫(xiě)入狀態(tài)。類(lèi)似 Java 的 try {} finally{ round.execute() }

多個(gè) round 一定會(huì)執(zhí)行，round 異常不會(huì)導(dǎo)致任務(wù)失敗。 

<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 任務(wù)的運(yùn)行日志。

SparkSQL Around可使用的變量

SparkSQL Around的執(zhí)行效果

Prepare round 可做插入（insert）動(dòng)作，after round 可做更新 （update）動(dòng)作，相當(dāng)于在數(shù)據(jù)庫(kù)表中從執(zhí)行開(kāi)始到結(jié)束有了完整的日志記錄。SparkSQL Flow 會(huì)保證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)證的租戶(hù)名稱(chēng)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

每個(gè)Spark Flow 任務(wù)本質(zhì)上是一連串的 SparkSQL 操作，在 SparkUI SQL tab 里可以看到 flow 中重要的數(shù)據(jù)表操作。

regiserDataFrameAsTable 是每個(gè) source 和 Transform 的數(shù)據(jù)在 SparkSQL 中的數(shù)據(jù)視圖，每個(gè)視圖都會(huì)在 SparkContex 中注冊(cè)一次。

對(duì)RegisterDataFrameAsTable的分析

通過(guò)單個(gè) regiserDataFrameAsTable 項(xiàng)進(jìn)行分析，SparkSQL 并不是把source 的數(shù)據(jù)立即計(jì)算把數(shù)據(jù)放到內(nèi)存，而是每次執(zhí)行 source 時(shí)只是生成了一個(gè) Logical Plan，只有遇到需要提交的算子（Action），SparkSQL 才會(huì)觸發(fā)前面所依賴(lài)的的 plan 執(zhí)行。

這是一個(gè)開(kāi)發(fā)框架，不是一個(gè)成熟的產(chǎn)品，也不是一種架構(gòu)。他只是基于 SparkSQL 整合了大多數(shù)的外部系統(tǒng)，能通過(guò) XML 的模板配置完成數(shù)據(jù)開(kāi)發(fā)。面向的是理解數(shù)據(jù)業(yè)務(wù)但不了解 Spark 的數(shù)據(jù)開(kāi)發(fā)人員。整個(gè)框架完成了大多數(shù)的外部系統(tǒng)對(duì)接，開(kāi)發(fā)者只需要使用 type 獲得數(shù)據(jù)，完成數(shù)據(jù)開(kāi)發(fā)后通過(guò) target 回寫(xiě)到目標(biāo)系統(tǒng)中。整個(gè)過(guò)程基本無(wú)須程序開(kāi)發(fā)，除非當(dāng)前的 SQL 函數(shù)無(wú)法滿(mǎn)足使用的情況下，需要自行開(kāi)發(fā)一下特定的 UDF。因此本框架在對(duì) SparkSQL 做了二次開(kāi)發(fā)基礎(chǔ)上，大大簡(jiǎn)化了 Spark 的開(kāi)發(fā)，可降低了開(kāi)發(fā)者使用難度。

看完上述內(nèi)容，你們對(duì)基于 Spark 的數(shù)據(jù)分析實(shí)踐是怎樣進(jìn)行的有進(jìn)一步的了解嗎？如果還想了解更多知識(shí)或者相關(guān)內(nèi)容，請(qǐng)關(guān)注億速云行業(yè)資訊頻道，感謝大家的支持。