第第頁基于電商大數據的數據倉庫的設計與實現摘要：近年來隨著國內外電子商務的飛速發展，電商用戶迎來了井噴式的增長。海量用戶所帶來的是海量數據。無論是是用戶在電商系統中點擊和瀏覽等事件埋點數據，還是用戶進行加入購物、收藏、下單和支付等實際業務數據都達到了PB乃至ZB級。面對海量數據，各電商企業難以避免的要進行數據挖掘、分析和實時計算，從而使數據產生更大的價值。但是面對數量如此龐大的歷史數據，如若在傳統的關系型數據庫上進行存儲、分析和計算并非明智之舉，關系型數據庫由于其低冗余和紛繁復雜的關系映射在面對海量數據時將大幅度降低數據查詢的效率。因此，建立數據倉庫已成為了大數據業務的基礎。在數據倉庫中可以進行數據分層，在不同層級中可以進行數據清洗、數據整合，將數據整合匯總成分析某一個主題域的服務數據寬表，為數據挖掘、分析和可視化打下基礎。當然，如何保質保量的將數據從業務數據服務器和日志服務器導入數據倉庫也是倉庫搭建前需要解決的重要問題，對于不同類型和不同來源的數據，都需要提供最為高效和安全的解決方案。數據倉庫搭建完成之后，對數據進行可視化和分析也就有了便捷和高質量的數據來源。綜上所述，本文將詳述如何將海量數據采集至數據倉庫，并且進行數據倉分層開發。關鍵詞：電子商務；大數據；數據倉庫；數據可視化目錄TOC\o"1-2"\h\u67761、緒論 1254181.1、研究背景與意義 160251.2、國內外研究現狀 238671.3、主要研究內容 2271721.4、本文的組織結構 3224582、相關技術分析 418182.1、Hadoop框架及其組件 4297002.2、數據采集技術 6209532.3、Hive 8188313、電商數據生成與采集 9291783.1、環境搭建 9165203.2、數據生成腳本編寫 13230703.3、數據采集 20101194、數據倉庫設計 25130534.1、環境搭建 25120284.2、數據倉建模方法介紹及選擇 26131954.3、數據分層 272985、壓力測試 34306205.1、HDFS測試 35252855.2、Kafka測試 36124126、期待與展望 3627053參考文獻 37隨著電子商務的飛速發展，隨著大數據和云計算相關技術的誕生，隨著微服務和SaaS等新興架構的出現，電商大數據已成為了一個炙手可熱的話題。由于龐大的用戶量以至于各大電商系統中每天都能產生海量數據。對于這些數據進行采集、存儲、加工和分析并為企業決策提供支撐已經成為了至關重要的問題。然而在這一系列過程中如何高效的采集數據，清洗數據并為可視化提供支撐等一系列問題，亟需改善!1、緒論1.1、研究背景與意義隨著電子商務的飛速發展，電商用戶不斷增長，電商業務不斷拓寬，近年來其產生的數據以指數級增長。截至2020年6月，中國網絡購物用戶高達7.49億，占網民整體的79.7%。數據顯示，截至2020年10月，我國在業/存續的電商主體265.1萬，其中企業196.0萬，個體戶69.1萬，電商個體戶占比26.1%，超過總量的1/4。顯而易見的是新冠疫情大流行對實體店的沖擊巨大，大量實體銷售份額流入電商。企查查數據顯示，2020年一季度電商市場主體注冊量12.6萬，同比下降31.8%，二、三季度電商市場主體注冊量26.6萬、24.4萬，同比增長26.2%、17.4%。而“新零售”給眾多電商巨頭公司帶來了前所未有的機遇，究其本質是電商巨頭在數據密集型應用上具有傳統行業和企業無法比擬的數據優勢，他們擁有著全方位、寬領域和多層次的用戶數據，特是這其中還包括了用戶身份信息、用戶網絡狀態、用戶使用的終端信息、用戶業務、用戶地理位置信息、用戶社交關系和用戶信用等具有極高商業價值的數據。[1]賈翼.基于Hive的電商多維分析系統的設計與實現[D].浙江工業大學,2020.這些數據真實反映了當今世界和全社會的運行演化進程，但常常在冗雜的數據海洋之中被忽視。[1]賈翼.基于Hive的電商多維分析系統的設計與實現[D].浙江工業大學,2020.[2]崔迪,郭小燕,陳為.大數據可視化的挑戰與最新進展[J].計算機應用,2017,37(07):2044-2049+2056.由此可見電商大數據時代，或者說DT時代已然到來。海量的商品和海量用戶必然導致電商系統每日產生TB級別的數據，這些數據具有數據量大、增長快速、非結構化等特點，可能隱藏著大量的潛在信息[3]侯曉芳,王歡,李瑛.一種基于HIVE和分布式集群的大量數據高效處理方法研究[J].中國電子科學研究院學報,2018,13(03):315-320.[3]侯曉芳,王歡,李瑛.一種基于HIVE和分布式集群的大量數據高效處理方法研究[J].中國電子科學研究院學報,2018,13(03):315-320.1.2、國內外研究現狀 當前的主流離線數據倉庫是將數據通過傳輸和采集框架持久化到HDFS等分布式文件系統中，并實現壓縮（LZO壓縮等）和小文件歸檔。最終將原始數據匯總到hive等數據倉工具中，并在倉庫中實現數據分層，進而進行數據清洗、歸類、拓展及分析。最后進行可視化展示。在電商產業界，國內的知名互聯網巨頭都爭先投入資金成立大數據研發團隊。特別是阿里以及其旗下的淘寶、天貓和螞蟻金服的成就最為引人矚目。[4][4]謝路倫.基于Spark的電商用戶行為分析與研究[D].北方民族大學,2019.但隨著數據量的不斷膨脹，在一些業務場景下數據倉庫已經難以滿足需求，因此數據湖的概念應運而生。早期數據湖相關理論是由大數據應用公司提出的，雖然數據載體都是在拓展性極強的HDFS廉價存儲硬件之上的。隨著數據量的急劇增長，數據結構的多樣化，多元化的存儲方式已成為必然的需求。隨著互聯網的演進，大部分企業的業務數據都是處在大數據場景之下，但并非所有行業的數據都是適合存放在Hadoop集群之上的。數據湖的部分功能是把不同種類的數據連成網、織成片，其次是不需要預定義的模型就能進行數據分析。當年大部分的大數據架構都是極具擴展性的，并且可以為用戶提供大量的云計算服務。在傳統數據倉任然在廣泛使用的當下，大數據技術和數據湖技術不斷向更多領域的實時智能服務拓展，這些實時的智能服務可以為實時的決策制定提供支持。1.3、主要研究內容本文主要研究的內容有三個，一是設計與實現電商數據采集系統，二是設計與實現分層數據倉庫，三是設計數據可視化界面。本文將利用java代碼和shell腳本生成電商平臺測試數據，并利用數據采集框架將其持久化至分布式文件系統，并在數據倉庫中實現數據分層。在可視化系統中利用數據應用層的數據生成數據報表。在設計完成過程中還需完成相應的測試工作。系統整體架構如圖1.1所示：圖1.1系統體系結構圖1.4、本文的組織結構本文按照軟件工程方法學對數據倉庫及其相關輔助系統進行了問題定義（緒論及摘要）、可行性研究（著重技術可行性，這是最為關鍵的問題）、需求分析、軟件設計、實現和測試。緒論。該章節介紹了電商大數據的背景以及當前面臨的問題，并且對前沿技術發展進行了探討，對數據倉庫及其相關系統的設計做了簡要的介紹。相關技術分析。該章節介紹了大數據采集、存儲和可視化等相關技術，也是數據倉庫搭建中不必可少的技術。該章節主要介紹在企業中如何將業務服務器和日志服務器中累積的數據采集至分布式文件系統中。將數據采集至數據倉庫，并在數據倉庫中進行數據分層，對數據進行清洗以及粒度劃分，為數據可視化和大數據分析提供數據源。集群組建進行壓力測試，了解集群性能。對數據倉庫搭建完成之后的進一步延伸進行展望，包括可視化技術、數據分析和業務的更深層次的分析。02、相關技術分析2.1、Hadoop框架及其組件hadooop框架是由Aphache基金會所提供的分布式系統基礎架構。主要用于解決大數據場景下的數據存儲和數據的分析計算問題。Hadoop采用分布式存儲和并行處理機制充分利用集群的硬件性能進行高速運算和存儲。[5]LifengJiang.BigDataStatisticalAnalysisModelbasedonHadoopSystem[J].InternationalJournalofIntelligentInformation,2020,9(5).Hadoop實現了一個分布式文件系統(HadoopDistributedFileSystem)，即HDFS。HDFS極高的容錯性，并且部署在低廉的硬件上;而且它具有極高的讀寫能力來訪問服務器的數據，適合那些有著超大數據集(largedataset)的應用程序。HDFS通常以流的方式訪問分布式文件系統中的數據。從宏觀上來看，Hadoop可以說是一個更為龐大的概念即Hadoop生態圈，因為其經常與HBASE、Zookeeper等框架同時使用。[5]LifengJiang.BigDataStatisticalAnalysisModelbasedonHadoopSystem[J].InternationalJournalofIntelligentInformation,2020,9(5).總而言之，Hadoop的出現解決了大數據開發過程中并行計算、存儲和資源調度等至關重要的問題，用戶可以在這封裝好的一系列框架之上開發分布式程序。[6]夏靖波,韋澤鯤,付凱,陳珍.云計算中Hadoop技術研究與應用綜述[J].計算機科學,2016,43(11):6-11+48.Hadoop的運行速度極快因為它以并行的方式工作，通過并行處理加快處理速度。[7][6]夏靖波,韋澤鯤,付凱,陳珍.云計算中Hadoop技術研究與應用綜述[J].計算機科學,2016,43(11):6-11+48.[7]葉力銘.基于Spark電商用戶行為數據的分析與研究[D].沈陽師范大學,20、MapReduce架構概述MapReduce是一個能夠進行分布式并行運算的框架，適用于規模較大的數據集(TB級別)的并行運算。概念"Map(映射)"和"Reduce(歸約)"，和它們的主要思想，都是從面向過程編程語言里借鑒的，也有從矢量編程語言里借鑒的特性。對于程序員來說，無需分布式并行編程就可以在分布式系統上運行他們的程序是非常方便的。當前的軟件實現是指定一個Map函數將一組鍵值對映射為一組新的鍵值對，并指定一個并發的Reduce函數以確保所有映射的鍵值對共享同一個密鑰組。MapReduce的核心功能是將用戶自己編寫的業務邏輯代碼和默認構建集成到一個完整的分布式計算程序中，并發運行在Hadoop集群上。MapReduce操作程序一般需要分為兩個階段，即Map階段和Reduce階段。MapReduce作為早期的大數據分析框架，其有著諸多優點：MapReduce易于編程。它可以通過簡單地實現一些接口來完成一個分布式程序。這種分布式程序可以分布到龐大的服務器集群上運行。MapReduce具有良好的擴展性。當系統資源不能得到滿足的時候，可以通過簡單的增加物理機器來擴展其計算能力。MapReduce具有高容錯性。MapReduce最初的設計目的是使程序能夠部署在需要高容錯性的廉價PC上。若集群中某臺服務器宕機，它可以把計算任務轉到另一個節點執行，防止任務失敗。MapReduce適用于PB級以上海量數據的離線處理，可以實現數千個服務器集群的并發工作。MapReduce編程模型將應用程序分成許多小部分，以便對大數據進行算術處理。[8][8]Jin-BangHsu,Chi-FangLin,Yang-ChengChang,Ren-HaoPan.UsingindependentresourceallocationstrategiestosolveconflictsofHadoopdistributedarchitectureinvirtualization[J].ClusterComputing,2020(prepublish).但MapReduce隨著大數據業務需求的不斷變化，MapReduce由于不擅長實時計算、流式計算和DAG計算等原因，逐漸被Spark等主流框架替代。2.1.2、Yarn架構概述Yarn是Hadoop的資源管組件，它是一個被廣泛使用的資源管理系統，可以為上層應用提供統一的資源管理和調度。Yarn的廣泛使用給集群利用率的提升、統一的資源管理和數據同步共享帶來了便捷。Yarn層次結構的本質是ResourceManager。該實體控制整個集群，并管理應用程序對基本計算資源的分配。ResourceManager將各種資源部分(計算量、內存、帶寬等)精心安排到基本的節點管理器(Yarn的每個節點代理)。ResourceManager和ApplicationMaster一起分配資源，和NodeManager一起啟動和監控它們的基本應用程序。硬盤資源的分配交給NameNode，SecondaryNameNode是NameNode的備份，但不是NameNode的熱備份，不能對其進行代替。2.1.3、HDFS架構概述HDFS是Hadoop框架中最為核心的部分，直到今天依然被廣泛使用。隨著數據量越來越大，不可能將所有數據都存儲在一個操作系統上，因此數據會被分發到更多由操作系統管理的磁盤上，但是這樣一來并不方便管理與維護，于是用來管理多臺機器上的文件的分布式文件管理系統便應運而生。HDFS只是眾多分布式文件管理系統中的一種，它在hadoop框架中位于架構中的最底層，其持久化文件數據到磁盤的過程中是通過分塊的處理來持久化若干小文件，不同的版本的分塊默認的存儲空間不一樣，在hadoop1.x版本中默認大小為64MB，在hadoop2.x及以上版本中默認大小為128MB。但是在數據采集的過程中，可以把默認占用的存儲空間調高或者調低。[9][9]周偉坤.基于spark的電商用戶行為大數據分析的研究[D].廣東工業大學,2019.HDFS被設計為運行在通用硬件上的分布式文件系統。它與現有的分布式文件系統有很多共同之處。但與此同時，它明顯不同于其他分布式文件系統。HDFS是一個高度容錯的系統，適合部署在便宜的機器上。HDFS可以提供高吞吐量的數據訪問，非常適合大規模的數據集。其廣泛應用于一次寫入多次讀出的場景，且不支持文件的修改。適合用于數據分析應用，但并不適用于網盤應用。此外，HDFS在POSIX約束上也比較寬松，在很大程度上有效支持流式文件及相關數據。[10][10]余文俊.電商大數據實時數據看板的設計和實現[D].上海交通大學,2017.作為使用最為廣泛的文件系統之一，HDFS有以下優點：具有高容錯性，HDFS中的數據會自動保存多個副本，通過增加副本的數量成功提高了容錯性。當副本出現丟失現象時，系統會進行自動恢復。適合處理大數據，從數據源大小上看HDFS能夠處理數據規模達到TB乃至PB級別的數據，從文件數量上看HDFS可以處理百萬規模以上的文件數量。HDFS可以構建在廉價機器上，成本較低。2.2、數據采集技術2.2.1、FlumeFlume是一個由Cloudera提供的用于收集、聚集和傳輸大量日志的高可用、可靠和分布式系統。Flume通常在日志系統中定制各種數據發送器，用于收集數據;同時，Flume提供了簡單地處理數據并將其寫入各種數據接收者的能力。在電商系統中日志數據是較為重要的分析依據，可以根據用戶的登錄日志、點擊操作等一系列行為大數據分析出用戶畫像。因此flume作為一個日志采集系統在數據倉庫構建的過程中是不可或缺的。每個flume進程包含source、channel和sink三個部分。Source是負責接收數據到FlumeAgent的組件。Source組件可以處理各種類型、各種格式的日志數據，包括avro、thrift、exec、jms、spoolingdirectory、netcat、sequencegenerator、syslog、http、legacy。Channel是位于Source和Sink之間的緩沖區。因此，Channel允許Source和Sink運作在不同的速率上。Channel是線程安全的，可以同時處理幾個Source的寫入操作和幾個Sink的讀取操作。Flume自帶兩種Channel：MemoryChannel和FileChannel。MemoryChannel是內存中的隊列。MemoryChannel在不需要關心數據丟失的情景下適用。如果需要關心數據丟失，那么MemoryChannel就不應該使用，因為程序死亡、機器宕機或者重啟都會導致數據丟失。FileChannel將所有事件寫到磁盤。因此在應用程序關閉或服務器宕機時不會出現數據丟失。Sink在執行時持續地遍歷Channel中的事件且批量地移除它們，并將這些事件批量寫入到存儲或索引系統、或者被輸送到另一個FlumeAgent。Sink組件目的地包括hdfs、logger、avro、thrift、ipc、file、HBase、solr、自定義。2.2.2、KafkaKafka是一個開源的流處理平臺，由Apache軟件基金會開發，由Scala和Java編寫。Kafka是一個高吞吐量的分布式發布-訂閱消息系統，可以處理網站中消費者的所有動作流數據。Kafka本質上是一個基于發布/訂閱模式的分布式消息隊列，主要用于大數據的實時處理。現在它已被廣泛應用于分布式集群應用中，當做多種類型的數據管道和消息系統使用。[11][11]牛牧.基于Kafka的大規模流數據分布式緩存與分析平臺[D].吉林大學,2016.Kafka由Producer、Consumer、ConsumerGroup、Broker、Topic、Partition、Replica、leader和follower等多個組件構成。Producer是消息生產者，就是向kafkabroker發消息的客戶端。Consumer是消息消費者，向kafkabroker取消息的客戶端。ConsumerGroup（CG是消費者組，由多個consumer組成。消費者組中的每個消費者負責使用不同分區中的數據，一個分區只能被一個消費者使用；消費者不會互相影響。全部的消費者都從屬于某個消費者組，可見消費者組是邏輯上的一個訂閱者。一臺kafka服務器就是一個broker。一個集群由多個broker組成。一個broker可以容納多個topic。Topic可以認為是一個隊列，生產者和消費者交互對象都是一個topic。Partition具有較高擴展性，因此一個規模龐大的topic可以分布到多個broker（即服務器）上，一個topic可以分為多個partition，每個partition是一個有序的隊列。Replica副本，用于確保集群中的某個節點出現異常時，該節點上的partition數據不丟失，且kafka仍然能夠繼續工作，kafka提供了副本機制，每個topic的每個分區都有很多個副本，一個leader和若干個follower。生產者和消費者都是向leader中生產數據和消費數據。follower實時從leader中同步數據，保持和leader數據的同步。leader發生故障時，某個follower會成為新的leader。kafka作為容錯極高的消息隊列，它的產生在一定程度上解決了flume等采集框架由于讀寫速度不匹配造成的傳輸出錯等一系列問題，這類消息隊列式的框架也成為了數據倉架構中不可或缺的一部分。2.2.3、SqoopSqoop是一個開源工具，主要用于Hadoop(Hive)和關系型數據庫之間的數據傳輸。它可以將傳統據庫中的數據導入Hadoop的HDFS，也可以將HDFS的數據導入關系數據庫。在Sqoop環境搭建完成之后，開發者便可以編寫相應的shell腳本將數據導入/導出HDFS。2.3、HiveHive是Hadoop生態圈的數據倉[12]曹軍瑤.基于Hadoop的軟包印刷設備售后數據研究與分析[D].西安理工大學,2018.，是hadoop生態中最為重要的一環。Hive可以將文件中的結構化數據映射到一個數據庫表中，并有著諸多查詢函數，可以將sql語句轉換成MapReduce任務來執行。其優點是學習成本低，簡單的MapReduce統計可以通過類似SQL的語句快速實現，而且不需要開發專門的MapReduce應用程序，非常適合數據倉庫的統計分析。Hive定義了一種簡單的類似于SQL的查詢語言叫做HQL，它允許熟悉SQL的用戶查詢數據，這也是Hive能夠快速普及的一個重要原因。現如今Hiveonspark等概念的誕生，則將Hive技術推向了頂點。HiveonSpark是指使用Spark替代傳統MapReduce作為Hive的執行引擎。Spark可以算是Hadoop中MapReduce組建的替代品，它使用高效的內存計算和處理優化機制來加快批量處理數據任務的運行速度，將Hadoop集群應用在內存上的運行速度提高100倍，在磁盤上的運行速度提高10倍。Spark使用生態系統中的流庫來處理實時流數據，以實現流處理能力。[13]趙娟,程國鐘.基于Hadoop、Storm、Samza、Spark及Flink大數據處理框架的比較研究[J].信息系統工程,2017(06):117+119.[12]曹軍瑤.基于Hadoop的軟包印刷設備售后數據研究與分析[D].西安理工大學,2018.[13]趙娟,程國鐘.基于Hadoop、Storm、Samza、Spark及Flink大數據處理框架的比較研究[J].信息系統工程,2017(06):117+119.電商數據生成與采集在數據采集過程中采用flume+kafka+flume的采集通道架構，其原因是flume不支持副本事件，當集群崩潰時可能導致數據丟失，而kafka則是一個極其可靠的消息隊列能對數據進行備份，在第二章已有詳細介紹。而kafka與HDFS可能存在數據讀寫速度不匹配的問題，所以需要使用flume采集數據至HDFS，詳見第五章壓力測試的實驗。3.1、環境搭建3.1.1、虛擬機集群搭建本項目預計在本地搭建三臺虛擬機，使用VMwareWorkstationPro作為集群搭建工具，使用Xshell5作為終端模擬軟件。虛擬機鏡像使用CentOS版本，并設置4內核驅動和50GB的磁盤空間。為了保證虛擬機與互聯網連通，并通過橋接方式接入互聯網。虛擬機創建完成之后還需對虛擬機的網絡進行配置。設置在本地局域網內的子網為，網關為，子網掩碼為。將三臺虛擬機的ip地址分別改為2、3、4，為防止虛擬機重啟之后重新動態分配地址，需將網絡相關配置文件，即/etc/sysconfig/network-scripts/ifcfg-ens33中的BOOTPROTO屬性改為static。并且在三個服務器上都需配置網關和域名解析器的ip。以上操作完成后執行systemctlrestartnetwork，配置即可生效。為了方便后續操作，也可以修改主機名并且配置相應映射。分別在三臺主機上打開/etc/hostname文件將對應的主機名分別改為hadoop1、hadoop2和hadoop3即可,并且在各虛擬機和windows中配置相應映射。項目硬件及虛擬機參數配置如表3.1所示：表3.1硬件及虛擬機參數部件參數CPUIntel(R)Core(TM)i5-7200UCPU@2.50GHz2.71GHz虛擬機內存2GB虛擬機處理器數量8虛擬機網絡適配器NAT虛擬機磁盤空間40GB3.1.2、hadoop環境配置Hadoop集群的安裝方式有三種，分別是單機模式、偽步進模式和完全步進模式。其中，單機和偽分步實質都是在單臺服務器上實現，沒有將Hadoop的各種子服務運行在多個服務器上，所以只有在完全分布式下，才是真正用于數據分析的項目環境。本課題便是將hadoop搭建在多服務器集群上。[14][14]曹軍瑤.基于Hadoop的軟包印刷設備售后數據研究與分析[D].西安理工大學,2018.由于三臺服務器集群可能會有諸多相同配置，因此先行編寫xsync腳本用于分發文件。具體如下：f[$#-lt1]Thenexit;fiforhostinhadoop1hadoop2hadoop3#遍歷整個服務器集群doforfilein$@doif[-e$file]#接收到傳入的文件路徑then#用ssh免密登錄分發文件pdir=$(cd-P$(dirname$file);pwd)#文件路徑fname=$(basename$file)#文件名ssh$host"mkdir-p$pdir"#免密登錄后創建文件夾rsync-av$pdir/$fname$host:$pdirelseecho$filedoesnotexists!#報錯fidoneDone集群各個組件在某一服務器上是否部署如下表3-2所示：組件名稱子服務服務器1服務器2服務器3HDFSNameNode是否否DataNode是是是SecondaryNameNode否否是YarnNodeManager是是是Resourcemanager否是否ZookeeperZookeeperServer是是是Flume(采集)Flume是是否KafkaKafka是是是Flume（消費）Flume否否是HiveHive是否否MySQLMySQL是否否SqoopSqoop是否否AzkabanAzkabanWebServer是否否AzkabanExecutorServer是否否HbaseHMaster是否否HRegionServer是是是配置時需在第一個主機的opt目錄下創建software文件夾，將apache-flume-1.9.0-bin.tar.gz、apache-zookeeper-3.5.7-bin.tar.gz、hadoop-3.1.3.tar.gz、hadoop-lzo-0.4.20.jar、jdk-8u212-linux-x64.tar.gz、kafka_2.11-2.4.1.tgz等相關壓縮包傳輸至該文件夾。使用tar命令將這些文件解壓至/opt/app/下，并配置相應環境變量。在解壓和配置好jdk和hadoop之后實現ssh免密登錄。具體環境變量配置如下：#JAVA_HOMEexportJAVA_HOME=/opt/app/jdk1.8.0_45exportPATH=$PATH:$JAVA_HOME/bin##HADOOP_HOMEexportHADOOP_HOME=/opt/app/hadoop-3.1.3exportPATH=$PATH:$HADOOP_HOME/binexportPATH=$PATH:$HADOOP_HOME/sbin而后使用xsync腳本將第一個主機上的jdk以及hadoop文件復制到其他主機，只需要進入jdk和hadoop所在上層目錄(即app，xsync腳本已經移動到bin文件夾下，可全局調用)，執行xsync所要復制的文件夾即可。而后還需使用scp命令將配置文件復制到其他主機，scp雖然傳輸速度比xsync慢，但是安全性更高。以上基本操作完成之后還需配置core-site.xml、hdfs-site.xml、yarn-site.xml和mapred-site.xml等配置文件。以上配置文件寫入后，還需配置worker文件（即配置統一啟動服務器集群各個主機的名稱）。再次使用xsync進行分發，分發完畢后hadoop的基本配置即可完成。結束配置后在2的hadoop文件的目錄下執行bin/hdfsnamenode-format進行格式化NameNode。因為Hadoop生態中的文件系統HDFS類似一塊磁盤，初次使用硬盤需要格式化，讓存儲空間明白該按什么方式組織存儲數據。3.1.3、Mysql與Scoop安裝與配置電商平臺的業務數據都是以存儲在結構化數據庫中，因此本案例以mysql為例模擬業務數據的相關操作。而mysql與hdfs之間傳輸數據需要依賴Sqoop框架，具體技術見上文，此處不再贅述。具體步驟如下：安裝mysql依賴、mysql-client和mysql-server更改密碼。先設置密碼最低長度和校正密碼，再更改密碼。實現任意主機都能訪問該數據庫。進入mysql庫，修改user表，把Host表內容修改為%，即執行命令updateusersethost="%"whereuser="root";。將Sqoop壓縮包解壓，重命名配置文件，并添加以下配置：#關于Hadoop、Hive和Zookeeper的路徑exportHADOOP_COMMON_HOME=/opt/app/hadoop-3.1.3exportHADOOP_MAPRED_HOME=/opt/app/hadoop-3.1.3exportHIVE_HOME=/opt/app/hiveexportZOOKEEPER_HOME=/opt/app/zookeeper-3.5.7exportZOOCFGDIR=/opt/app/zookeeper-3.5.7/conf(5)拷貝jdbc驅動至sqoop的lib目錄下。3.2、數據生成腳本編寫為保證倉庫搭建之后有合適的測試數據，以下設計生產用戶埋點數據與業務數據。數據埋點是一種很好的私有化部署數據獲取方法。所謂“埋點”，是數據收集領域(尤其是用戶行為數據收集領域)的一個術語，指的是針對特定用戶行為或事件進行捕獲、處理和發送的相關技術及其實施過程。埋點的技術本質是首先監控軟件應用程序運行過程中的功能使用情況和異常，并在事件發生時判斷和捕捉需要注意的事件，并且取必要的上下文信息，最后將特定格式的數據（json）發送至服務器集群。前端埋點的監聽事件，大部分由操作系統、web瀏覽器等平臺提供，通常也可以在開發時進行自行定義和監聽埋點事件（例如按鈕點擊事件等）。一般情況下，埋點可以通過監測分析工具提供的軟件開放工具包來進行編程實現。本案例針對用戶啟動檢測數據和行為日志進行數據腳本編寫，模擬埋點數據。而業務數據則是指在現實中實際發生的業務。例如，登錄、訂單、用戶和商品等相關數據。其以關系模型的映射方式持久化在結構化數據庫中。3.2.1、日志記錄結構在前端埋點時普遍采用異步響應以json的格式發送至日志服務器，為記錄最為詳細的日志產生時間，可以在json字符串中拼接上時間戳。以下為啟動日志字符串為例：1620181383937|{#該日志的時間戳，代表從1970至日志產生的那一刻經過了多少毫秒 "action":"1", "address":"China",#來自哪個區域 "mobilephone_brand":"Huawei",#手機品牌 "log_type":"start",#日志類型 "extend1":"",#拓展字段 "mail":"O90GZWW5@", "hw":"640*1136",#屏幕參數 "language":"es",#語言類型 "la":"47.1",#緯度s "ln":"36.4",#經度 "loading_time":"12",#加載時間 "md":"Huawei-2",#手機型號 "mid":"1",#每個設備獨有的id "network":"WIFI",#啟動時的網絡模式 "open_ad_type":"1",#啟動輪播圖推送廣告類型 "os":"8.2.3",#系統版本號 "t":"1620133941228",#時間戳 "uid":"1",#用戶的id "app_version_count":"17",#程序版本號 "app_version_name":"1.2.8"#程序版本名}3.2.2、業務數據庫設計由于電商紛繁復雜的業務邏輯必然要有大量的數據表來承載相關數據，此處只以部分業務為例，繪制ER圖并設計相關表的結構。部分實體關系圖如圖3.1所示：圖3.1業務系統ER圖以下對主要業務數據表結構進行設計，具體如表3.3、表3.4、表3.5、表3.6所示:表3.3用戶表字段類型含義USER_IDVARCHAR2(32)用戶ID，主鍵USER_NAMEVARCHAR2(100)用戶名稱USER_GENDERVARCHAR2(1)性別LOGIN_NAMEVARCHAR2(100)登錄名HOME_PHONEVARCHAR2(100)家庭電話EMAILVARCHAR2(200)郵箱USER_STATUSCHAR(1)用戶狀態CREATORVARCHAR2(16)創建者CREATE_TIMEDATE創建時間MOBILE_PHONEVARCHAR2(11)手機號ID_CARDVARCHAR2(18)身份證表3.4訂單表字段類型含義ORDER_IDVARCHAR2(32)訂單ID，主鍵CONSIGNEEVARCHAR2(32)收貨人CONSIGNEE_PHONEVARCHAR2(11)收件人電話TOTAL_AMOUNTVARCHAR2(10)總金額ORDER_STATUSCHAR(1)訂單狀態USER_IDVARCHAR2(32)用戶idADDRESSVARCHAR2(100)送貨地址ORDER_COMMENTVARCHAR2(255)訂單備注ORDER_TRADE_NUMVARCHAR2(32)訂單交易編號CREATE_TIMEDATE創建時間OPERATE_TIMEDATE操作時間EXPIRE_TIMEDATE失效時間PROVINCE_IDVARCHAR2(32)地區FEIGHT_FEEVARCHAR2(10)運費表3.5商品表字段類型含義PRODUCT_IDVARCHAR2(32)商品ID，主鍵PRICEVARCHAR2(16)價格PRODUCT_NAMEVARCHAR2(32)商品名稱PRODUCT_DESCVARCHAR2(255)商品描述WEIGTHVARCHAR2(16)商品重量TM_IDVARCHAR2(32)品牌idPRODUCT_DEFAULT_IMGVARCHAR2(100)默認圖片的路徑CREATE_TIMEDATE創建時間表3.6品牌表字段類型含義TM_IDVARCHAR2(32)品牌idTM_NAMEVARCHAR2(100)品牌名稱表3.7評論表字段類型含義COMMENT_IDVARCHAR2(32)評論IDUSER_IDVARCHAR2(32)用戶idPRODUCT_IDVARCHAR2(32)商品idORDER_IDVARCHAR2(32)訂單編號COMMENT_TYPEVARCHAR2(2)評價類型COMMENT_CONTENTVARCHAR2(255)評價內容CREATE_TIMEDATE創建時間表3.8優惠劵表字段類型含義COUPON_IDVARCHAR2(32)購物券IDCOUPON_NAMEVARCHAR2(100)購物券名稱COUPON_TYPEVARCHAR(10)購物券類型CONDITION_AMOUNTVARCHAR2(32)滿額數CONDITION_NUMVARCHAR2(32)滿件數ACTIVITY_IDVARCHAR2(32)活動編號BENEFIT_AMOUNTVARCHAR2(32)減金額BENEFIT_DISCOUNTVARCHAR2(32)折扣CREATE_TIMEDATE創建時間PRODUCT_IDVARCHAR2(32)商品idTM_IDVARCHAR2(32)品牌idOPERATE_TIMEDATE修改時間EXPIRE_TIMEDATE過期時間表3.9活動表字段類型含義ACTIVITY_IDVARCHAR2(32)活動idACTIVITY_NAMEVARCHAR2(32)活動名稱ACTIVITY_TYPEVARCHAR2(10)活動類型ACTIVITY_DESCVARCHAR2(255)活動描述START_TIMEDATE開始時間END_TIMEDATE結束時間CREATE_TIMEDATE創建時間3.2.3、埋點數據生成根據上述數據結構分析可得創建啟動日志、公共日志、廣告、用戶后臺活躍、評論、商品點擊日志、錯誤日志、收藏、商品列表、商品詳情、消息通知日志、點贊等實體類所需相關屬性。實體類設計完成之后即可編寫主類的生成腳本，該腳本中使用隨機數隨機生成各類型數據，其中應用啟動日志和事件日志產生的可能性各為50%。而在事件日志產生的過程中，具體是哪一類型的數據依舊由隨機數控制rand.nextBoolean()方法決定某一類型的數據是否發生，如此一來各類型事件發生的可能性均為50%。具體到某一事件的數據生成依然使用隨機數控制，使得整體數據呈均勻分布。將所有產生的數據以json格式寫入json數組。完成之后將該腳本打包成jar包，并上傳至服務器集群。最終需要在服務器集群中編寫shell腳本統一執行該jar包生成數據，具體腳本如下：#!/bin/bashforiinhadoop1hadoop2#只在兩個服務器采集數據do ssh$i"java-jar/opt/app/log-collector-1.0-SNAPSHOT-jar-with-dependencies.jar$1$2>/dev/null2>&1&"#執行jar包done腳本編寫完成之后還需給該文件提權，即執行chmod777lg.sh，隨后執行腳本數據即可生成。3.2.4、業務數據生成業務數據依然采用java腳本生成，其中將日期、比例和數量等參數寫在配置配置文件中，這樣將該腳本解耦合以便日后復用jar包。利用隨機數控制各類型數據生成，力求達到均勻分布。配置文件如下:#配置jdbc驅動，數據庫的url，用戶名及密碼等相關配置spring.datasource.driver-class-name=com.mysql.jdbc.Driverspring.datasource.url=jdbc:mysql://hadoop1:3306/ebusiness?characterEncoding=utf-8&useSSL=false&serverTimezone=GMT%2B8spring.datasource.username=rootspring.datasource.password=jxjjxjlogging.pattern.console=%m%nmybatis-plus.global-config.db-config.field-strategy=not_null#生成日志的時間mock.date=2021-02-15#是否重置mock.clear=1#是否生成新用戶mock.user.count=50#男性比例mock.user.male-rate=20#收藏取消比例mock.favor.cancel-rate=10#收藏數量mock.favor.count=100#購物車數量mock.cart.count=10#每個商品最多購物個數mock.cart.sku-maxcount-per-cart=3#用戶下單比例mock.order.user-rate=80#用戶從購物中購買商品比例mock.order.sku-rate=70#是否參加活動mock.order.join-activity=1#是否使用購物券mock.order.use-coupon=1#購物券領取人數mock.coupon.user-count=10#支付比例mock.payment.rate=70#支付方式支付寶：微信：銀聯mock.payment.payment-type=30:60:10#評價比例好：中：差：自動ment.appraise-rate=30:10:10:50#退款原因比例：質量問題商品描述與實際描述不一致缺貨號碼不合適拍錯不想買了其他mock.refund.reason-rate=30:10:20:5:15:5:5配置完成后用java-jar命令執行jar包即可生成數據。3.3、數據采集3.3.1、Flume采集數據在上述生成數據的過程中只在2和3執行了腳本，因此只在這兩個節點部署日志采集flume。由于需要先將數據導入Kafka，因此在選擇采集flume組件時選用TailDirSource+KafkaChannel。在Flume1.6以前需要自己自定義Source定位每次讀取到文件的某個位置，實現斷點續傳，因此常常出現數據丟失或產生一定數量臟數據的情況（數據重復），但支持斷點續傳和多目錄的TailDirSource出現之后改變了這一狀況。而采用KafkaChannel，省去了Sink，提高了效率，KafkaChannel數據存儲在Kafka里面，所以數據是存儲在磁盤中，因此數據可靠性也有了保障。在大數據時代數據恢復需求具有高效率性[15]盧學遠,錢育蓉,英昌甜.基于內存云的自適應數據恢復策略[J].東北師大學報(自然科學版),2020,52(04):82-89.[15]盧學遠,錢育蓉,英昌甜.基于內存云的自適應數據恢復策略[J].東北師大學報(自然科學版),2020,52(04):82-89.在采集日志的flume中需要配置兩個攔截器，分別是日志過濾攔截器和類型區分攔截器。前者對數據進行初步清洗，清除部分臟數據，后者對日志類型進行區分以便在Kafka中區分topic。在日志過濾攔截器中分別對日志收尾、是否符合json格式和是否匹配時間戳長度等一系列特征進行校驗，若不符合則移除該數據。在類型區分攔截器中根據之前生成數據中是否包含start和event字符串進行判斷，并給啟動日志數據與用戶事件數據添加相應的頭部，以便后續進行區分。攔截器代碼編寫完成后需創建flume相關配置文件，其主要配置如下：#攔截器配置erceptors=i1erceptors.i1.type=erceptor.LogETLInterceptor$Berceptors.i2.type=erceptor.LogTypeInterceptor$Buildera1.sources.r1.selector.type=multiplexinga1.sources.r1.selector.header=topica1.sources.r1.selector.mapping.topic_start=c1a1.sources.r1.selector.mapping.topic_event=c2配置完成后可編寫shell腳本完成數據采集。攔截器部分實現如下：//過濾啟動日志publicstaticbooleanvalidateStart(Stringlog){if(log==null){returnfalse;}//為空即返回flag過濾//每條log去重之后判斷其前后是否與日志格式匹配if(!log.trim().startsWith("{")||!log.trim().endsWith("}")){returnfalse;}returntrue;}//過濾事件日志publicstaticbooleanvalidateEvent(Stringlog){if(log==null){returnfalse;}String[]logContents=log.split("\\|");//按照分隔符切分日志if(logContents.length!=2){returnfalse;}//分隔符號大于1即不符合格式//判斷時間戳長度是否符合要求，內容是否有異常if(logContents[0].length()!=13||!NumberUtils.isDigits(logContents[0])){returnfalse;}//對分割后的內容格式再次進行判斷if(!logContents[1].trim().startsWith("{")||!logContents[1].trim().endsWith("}")){returnfalse;}returntrue;}//單event過濾方法@OverridepublicEventintercept(Evente){//取出body數據byte[]body=e.getBody();Stringlog=newString(body,Charset.forName("UTF-8"));//取出header，注意下面取出的這個headers不僅僅是個引用，底層源碼與se不同//這個headers可以直接改變header的實際值Map<String,String>headers=e.getHeaders();if(log.contains("start")){headers.put("topic","topic_start");}else{headers.put("topic","topic_event");}returne;}//多event過濾方法@OverridepublicList<Event>intercept(List<Event>events){ArrayList<Event>interceptors=newArrayList<>();for(Eventevent:events){ //遍歷list，依次調用單event方法即可Eventintercept1=intercept(event);interceptors.add(intercept1);}returninterceptors;}3.3.2、Flume消費Kafka在2和3的flume已經采集日志文件之后，即可在Kafka上創建topic分區，根據過濾器加的頭部將啟動日志和事件日志分離，并且在不同服務器上的Kafka中實現交叉備份。而Kafka中的數據最后會經由Flume消費至HDFS。Flume消費Kafka數據采用KafkaSource+FileChannel+HDFSSink的組合。FileChannel傳輸速度相對于Memory慢，但數據安全保障高，Agent進程掛掉也可以從失敗中恢復數據。FileChannel優化也可進行優化，通過配置dataDirs指向多個路徑，每個路徑對應不同的硬盤，增大Flume吞吐量。值得注意的是HDFSSink在把數據傳入HDFS時候需要實現LZO壓縮和小文件的處理。LZO是一個用ANSIC語言編寫的無損壓縮庫。它能夠提供非常快速的壓縮和解壓功能。解壓并不需要內存的支持，即使使用非常大的壓縮比例進行緩慢壓縮出的數據，依然能夠非常快速的解壓。并且LZO壓縮是線程安全的。而小文件問題是大數據場景中最為常見的問題之一，每個小文件都有一份元數據，其中包括文件路徑，文件名，所有者，所屬組，權限，創建時間等，這些信息都保存在Namenode內存中。所以過多的小文件會占用Namenode服務器大量內存，影響Namenode的性能和使用壽命。而且默認情況下MapReduce會對每個小文件啟用一個Map任務計算，非常影響計算性能。同時也影響磁盤尋址時間。實現LZO壓縮和解決小文件問題都可以在配置文件中進行配置，具體配置如下：#每二十秒刷新一次a1.sinks.k1.hdfs.rollInterval=20#達到指定的字節長度即生成一個文件a1.sinks.k1.hdfs.rollSize=134217728a1.sinks.k1.hdfs.rollCount=0a1.sinks.k2.hdfs.rollInterval=20a1.sinks.k2.hdfs.rollSize=134217728a1.sinks.k2.hdfs.rollCount=0##CompressedStream表示壓縮文件a1.sinks.k1.hdfs.fileType=CompressedStreama1.sinks.k2.hdfs.fileType=CompressedStream##壓縮類型為lzop壓縮a1.sinks.k1.hdfs.codeC=lzopa1.sinks.k2.hdfs.codeC=lzop3.3.3、業務數據導入HDFS業務數據導入HDFS基于Sqoop編寫Shell腳本，命令行只要要輸入兩個參數控制最終結果，分別是日期和是否是第一次傳入。日期即是把所需某日的輸入導入。是否第一次傳入考慮的是數據中存在一些特殊的固定數據，這種數據只需要采集一次，不需要每天都進行采集。例如，省份、地區等相關數據。在腳本中配置好mysql、sqoop和hadoop等相關參數之后將各表select語句作為參數傳入hadoopjar命令執行即可。由于腳本過于冗長此處不詳細贅述，關鍵命令如下：#do_date參數即由select語句替換hadoopjar/opt/app/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jarpression.lzo.DistributedLzoIndexer/origin_data/ebusiness/db/$1/$do_date4、數據倉庫設計4.1、環境搭建在數據采集的環境搭建中，已經能夠滿足數據倉的設計與分層，但由于spark等新興技術的崛起，現實業務中往往會引入spark框架代替MapReduce進行實時計算。由于應用程序編程接口（API）的可用性及其性能，Spark變得非常流行，甚至比MapReduce框架更受歡迎。[16]AhmedN.,BarczakAndreLC,SusnjakTeo,RashidMohammedA..AcomprehensiveperformanceanalysisofApacheHadoopandApacheSparkforlargescaledatasetsusingHiBench[J].JournalofBigData,2020,7(1).Spark是把運算的結果放到了內存中，而MapReduce是將結果放到硬盤中，顯然內存的計算速度要比磁盤快很多。[17]高月.基于大數據的電商用戶畫像的研究與應用[D].沈陽師范大學,2020.而且隨著數據量的提升，Spark的速度比MapReduce的速度提升的更快。[16]AhmedN.,BarczakAndreLC,SusnjakTeo,RashidMohammedA..AcomprehensiveperformanceanalysisofApacheHadoopandApacheSparkforlargescaledatasetsusingHiBench[J].JournalofBigData,2020,7(1).[17]高月.基于大數據的電商用戶畫像的研究與應用[D].沈陽師范大學,2020.[18]常有學.基于智能制造的數據倉庫的設計與實現[D].南昌大學,2020.具體配置步驟如下：下載spark源碼編譯成壓縮包配置spark環境變量及運行環境連接sparkjar包到hive上傳spark依賴到HDFS修改hive配置文件，具體如下：<!--Sparkjar包路徑--><property><name>spark.yarn.jars</name><value>hdfs://hadoop1:8020/spark-jars/*</value></property><!--Hive引擎--><property><name>hive.execution.engine</name><value>spark</value></property>4.2、數據倉建模方法介紹及選擇4.2.1、關系模型關系建模又叫ER建模，用實體和實體間的關系描述數據模型和企業業務架構，在范式理論上符合3NF，即表中的所有數據元素不但要能唯一地被主關鍵字所標識,而且它們之間還必須相互獨立,不存在其他的函數關系。由此可見在關系模型中數據冗余較少，在關系型數據庫中采用低延時實時寫入數據，適應于傳統信息系統的業務場景。雖然關系模型冗余較少，但在大規模數據和跨表分析、統計查詢過程中，會造成多表關聯，大大降低執行效率，并且開發者要熟悉上百個表中紛繁復雜的關系需要投入大量時間成本。所以通常我們采用維度模型建模，把相關各種表整理成兩種：事實表和維度表兩種。故關系模型不在此處贅述。4.2.2、維度模型維度是從多個不同的角度度量業務，常用來反映業務一些屬性，這些屬性的集合構成一個維度。維度可以看作是一個數據域，如地理維度(國家、省、市、州、縣等)，時間維度(年、月、日、時、分、秒等內容)。在電商業務中還有用戶維度、商品維度等。維度建模以分析決策的需求為出發點建立模型，一般有較好的大數據復雜查詢的響應效率，更直接面向業務，維度模型常用代表有星形模型和在一些特殊場景下適用的雪花模型。雪花模型用于維度外部還有進一步的維度，雪花模型比起數據冗余較小，故查詢時效率較低。因此在本案例中主要使用星型模型。維度模型包含維度表和事實表兩種基本表，事實表中的每行數據代表一個業務事件（訂單生成、付款、退款、評論等）。“事實”這個術語表示的是業務事件的度量值（頻率、數量、價格等）。維度表則是對事實信息進行描述。以下圖4-1以付款為例進行維度建模：4.3、數據分層4.3.1、數據倉分層概述數據倉分層是大數據時代處理大數據業務的重要基礎，只有進行合理的分層才能可靠、高校和安全的處理大數據業務。分層工作有著諸多優點。一是把復雜的問題簡化，分層之后每層只需要執行簡單的業務，但一系列簡單的業務組合之后便能夠完成復雜的業務。二是減少重復開發和計算，經過分層之后復雜的計算和查詢被拆開，如果有類似的業務再出行，之前開發的部分便可復用。三是保證了數據的安全性和可靠性，分層之后在原始數據層就存在了對原始數據的備份有利于防止數據丟失，分層也可以將隱私數據進行模糊處理確保后續分析不會觸及個人隱私。4.3.2、ODS層ODS層(OperationDataStore)即為原始數據層，這是數據準備區，也稱為貼源層。數據倉庫源頭系統的數據表通常會原封不動的存儲一份，這稱為ODS層，是后續數據倉加工數據的來源。ODS層也起到隔離原始數據和業務數據的作用。ODS層的表通常包括兩種類型，一種用于存儲當前要加載的數據，另一種用于存儲處理后的歷史數據。在分層前的一系列工作已經將數據采集至HDFS并實現了壓縮，在此處只需將Hive數據庫中的相關表建立，并且將數據剪切過來并建立索引即可。在建表時需要注意對除只需要導入一次的特殊表之外的數據進行分區。因為在Hive查詢中通常會掃描整個表內容，所以做不必要的工作會花費很多時間。有時，您只需要掃描表中的一部分相關數據，并在相應的分區中查找，從而減少查詢時間。為了方便后續數據自動化裝載，可編寫相應的shell腳本完成數據傳輸，以下以導入埋點數據為例：#!/bin/bash#定義數據庫名，hive目錄路徑，以及上一天的日期database=ebusinesshive=/opt/app/hive/bin/hiveselect_date=`date-d'-1day'+%F`#如果有指定日期，則導入那天的數據if[[-n"$1"]];thenselect_date=$1fi#加載數據的sql語句sql="loaddatainpath'/origin_data/ebusiness/log/topic_start/$select_date'overwriteintotable${database}.ods_start_logpartition(dt='$select_date');loaddatainpath'/origin_data/ebusiness/log/topic_event/$select_date'overwriteintotable${database}.ods_event_logpartition(dt='$select_date');"#執行$hive-e"$sql"hadoopjar/opt/app/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jarpression.lzo.DistributedLzoIndexer/warehouse/ebusiness/ods/ods_start_log/dt=$select_datehadoopjar/opt/app/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jarpression.lzo.DistributedLzoIndexer/warehouse/ebusiness/ods/ods_event_log/dt=$select_date4.3.3、DW層之DWD層數據倉庫層（即DW）是數據倉庫中最為重要的部分，其在原始數據層的基礎上對數據進行大量的加工和細分工作，是承上啟下的部分。DW層可以根據不同的業務需求分為1到n層不等，在此處又可細分為DWD層和DWS層。DWD層是在原始數據層的基礎上對行為數據進行解析并對核心數據進行過濾，保存的是粒度最細的數據。由于前端采集的日志數據都是以json格式與后端進行交互，并由前述提及的采集流程持久化至ODS層。但其在讀入ODS的過程中只是將數據按行讀取至表中其具體字段并未解析，故在導入數據時使用hive自帶的get_json_object()函數進行數據篩選和解析。而建表可使用壓縮比較大的的parquet方式。該層啟動埋點數據裝載腳本除插入語句外與ODS層類似，插入語句將需要的數據使用get_json_object()函數選擇封裝進腳本即可，此處不再贅述。而解析埋點事件日志任然是以按行讀取字符串的方式進入ODS層，由于其格式與json有一定的差異，因此需要使用自定義函數進行解析，而對于不同的事件必定要分配到不同的表中，這樣數據才有分析的價值。解析函數主要使用java實現，實現之后打成jar包傳送至服務器集群，并將hive函數與java類關聯即可，主要函數如下：publicStringevaluate(Stringline,Stringkey)throwsException{//傳入的key可能是時間戳，事件數組和公共字段//處理line，即按行讀取后的日志數據//isBlank判斷是否為null或空字符串If(StringUtils.isBlank(log[1])){return"";}String[]logLine=line.split("\\|");//將時間戳與后續數據切割//切割后數組長度不為2，即格式不對，該行數據丟棄if(logLine.length!=2){return"";}//將字符串頭尾空值去除，并創建為json對象開始解析JSONObjectbaseJson=newJSONObject(logLine[1].trim());Stringresult="";//根據傳進來的key獲取對應的值if("action".equals(key)){if(baseJson.has("action")){result=baseJson.getString("action");}}elseif("start".equals(key)){result=logLine[0].trim();}else{JSONObjectallShare=baseJson.getJSONObject("as");if(allShare.has(key)){result=cm.getString(key);}}returnresult;}//以下方法輸入參數是用戶輸入的參數類型@OverridepublicStructObjectInspectorinitialize(ObjectInspector[]argOIs)throwsUDFArgumentException{//首先進行參數類型校驗List<?extendsStructField>allStructFieldRefs=argOIs.getAllStructFieldRefs();If(!allStructFieldRefs.size()==1){thrownewUDFArgumentException("參數個數異常")；}if(!"string".equals(allStructFieldRefs.get(0).getFieldObjectInspector).getTypeName){thrownewUDFArgumentException("參數類型只能為字符串")；}ArrayList<String>fieldNames=newArrayList<String>();ArrayList<ObjectInspector>fieldOIs=newArrayList<ObjectInspector>(); //添加列名及對象檢查器fieldNames.add("event_name");fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);fieldNames.add("event_json");fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);//返回列名及對象檢查器returnObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);}//輸入1條記錄，將記錄切割成若干結果@Overridepublicvoidprocess(Object[]objects)throwsHiveException{//獲取傳入的記錄，只有一條，在上述初始化方法中已經進行了校驗Stringinput=objects[0].toString();//如果傳進來的數據為空，直接返回過濾掉該數據if(StringUtils.isBlank(input)){return;}else{try{//獲取一共有幾個事件（ad/facoriters）JSONArrayja=newJSONArray(input);if(ja==null)return;//循環遍歷每一個事件for(inti=0;i<ja.length();i++){Str