gobblin培訓文檔

1、Gobblin培訓文檔孟云雷2016/05Gobblin IntroduceGobblin ArchitectureGobblin DeploymentGobblin ScheduleGobblin on YarnJDW京東數(shù)據(jù)倉庫業(yè)務目錄 Gobblin 是一個通用數(shù)據(jù)集成框架，從一些數(shù)據(jù)源（如：數(shù)據(jù)庫，rest APIs,FTP/SFTP服務器，文件目錄等）抽取、轉(zhuǎn)換和加載海量數(shù)據(jù)到Hadoop上。Gobblin 處理日常規(guī)劃任務需要所有數(shù)據(jù)攝取 ETLs，包括作業(yè)/任務規(guī)劃，任務分配，錯誤處理，狀態(tài)管理，數(shù)據(jù)質(zhì)量檢測，數(shù)據(jù)發(fā)布等等。 Gobblin 通過同樣的執(zhí)行框架從不同數(shù)據(jù)源攝取數(shù)據(jù)

2、，在同一個地方管理所有不同數(shù)據(jù)源的元數(shù)據(jù)。同時結(jié)合了其他特性，比如自動伸縮，容錯，數(shù)據(jù)質(zhì)量保證，可擴展和處理數(shù)據(jù)模型改革等等。Gobblin 變得更容易使用，是個高效的數(shù)據(jù)攝取框架。Gobblin 介紹Gobblin Architecture架構(gòu)圖: 架構(gòu)概述 淺綠色部分是Gobblin的結(jié)構(gòu)體，都提供了相應的接口和簡單通用實現(xiàn)，所有的結(jié)構(gòu)體通過配置添加或擴展現(xiàn)有實現(xiàn)，實現(xiàn)可擴展可插拔。 綠色部分是Gobblin的工作單元或工作任務。 橙色部分是作業(yè)和任務的執(zhí)行器和任務狀態(tài)管理器，屬于部署環(huán)境，用于作業(yè)和任務的實際執(zhí)行。 紅色部分是gobblin的部署模式，有單例模式和hadoop集群模式，單

3、例模式是在一臺機器上運行，通過線程池并行運行作業(yè)和任務；hadoop集群模式是多臺機器并行運行作業(yè)和任務；還有一種是gobblin做為本地應用運行在yarn上（與mapreduce和spark同級別），這種方式實現(xiàn)了實時數(shù)據(jù)攝?。ㄒ话闶墙邮誯afka的消息數(shù)據(jù)）。Gobblin Architecture 工作流程 首先，作業(yè)調(diào)度執(zhí)行作業(yè)之前先獲取一把鎖，防止下個相同的調(diào)度作業(yè)，類似于集成的_LOCK,調(diào)度系統(tǒng)可選擇Azkaban（可百度），Oozie。 下一步，實例化Source，作用是分割攝取數(shù)據(jù)成多個工作單元（WorkUnit），每個工作單元負責攝取數(shù)據(jù)的一部分；Source還負責為每個工

4、作單元創(chuàng)建一個提取器（Extractor）,用于實際數(shù)據(jù)的攝取。 Gobblin Architecture 工作流程 攝取組件比較： 下一步，為每個工作單元（WorkUnit）創(chuàng)建任務（hadoop map任務），也可以通過一個特殊的MultiWorkUnit（包裝了多個WorkUnit）創(chuàng)建多個任務。 下一步，啟動和運行任務，在單個節(jié)點上的獨立模式中，任務是在專用于該作業(yè)線程池運行，它的大小是在每個作業(yè)基礎(chǔ)上進行配置。在Hadoop集群上的Hadoop MapReduce的模式，任務在運行映射器（純粹用作容器運行的任務）。 下一步，作業(yè)完成，對于數(shù)據(jù)發(fā)布，由任務狀態(tài)和JobCommit政策確

5、定JobCommitPolicy.COMMIT_ON_PARTIAL_SUCCESS 部分成功就可以 JobCommitPolicy.COMMIT_ON_FULL_SUCCESS 所有任務都成功才可以 下一步，數(shù)據(jù)發(fā)布后，作業(yè)持久話作業(yè)/任務的狀態(tài)到狀態(tài)存儲，類似于集成的_STARTTIME,將這一次的最大時間或者id寫入到存儲中，gobblin中稱為水?。╳atermarks ） 下一步，清理臨時目錄；最后，刪除鎖。Gobblin Architecturegobblingobblinhadoophadoop功能功能SourceSourceInputFormat控制數(shù)據(jù)分割和讀取WorkUni

6、tWorkUnitInputSplit數(shù)據(jù)塊ExtractorExtractorRecordReader讀取器 組件詳解構(gòu)成圖Source and Extractor Source在一個數(shù)據(jù)源和gobblin之間起到適配器的作用，并且在工作流的開始階段被使用。 Extractor顧名思義，實際與數(shù)據(jù)源連接和抽取數(shù)據(jù) Gobblin外圍提供了許多內(nèi)置的Source和Extractor的實現(xiàn)，為不同類型的數(shù)據(jù)源工作，為web service服務提供了許多Rest API，數(shù)據(jù)庫支持JDBC，F(xiàn)TP/SFTP服務和配置?，F(xiàn)在， Extractor是面向記錄的， Extractor每次讀取一個數(shù)據(jù)記錄

7、，雖然在內(nèi)部它可以選擇pull和cache一批數(shù)據(jù)記錄。后續(xù)會增加選項提取器，支持面向字節(jié)和面向文件的處理。Converter 轉(zhuǎn)換器（Converter）負責架構(gòu)和數(shù)據(jù)記錄之前的轉(zhuǎn)換和它是核心構(gòu)造對于數(shù)據(jù)傳輸。Gobblin Architecture 組件詳解Converter Converters之間都是兼容的并能和其他轉(zhuǎn)換器鏈接在一起，在數(shù)據(jù)輸入結(jié)構(gòu)、輸出結(jié)構(gòu)和數(shù)據(jù)記錄類型上只要是一對轉(zhuǎn)換器（輸入和輸出）都是兼容的。注意 一個Converter轉(zhuǎn)換一個輸入架構(gòu)到一個輸出架構(gòu)。轉(zhuǎn)換一個輸入數(shù)據(jù)記錄可以到0（1:0 mapping），1（1:1 mapping），or許多（1:N mappi

8、ng）輸出數(shù)據(jù)記錄。Gobblin Architecture 組件詳解Quality Checker 質(zhì)量檢查者（QualityChecker），顧名思義，負責數(shù)據(jù)的質(zhì)量檢查。其有兩種類型：一個是檢查單個的數(shù)據(jù)記錄,決定每一個記錄應該通過到下一個階段在一個任務流程中,稱為行級質(zhì)量檢查者;另一個檢查整個任務的輸出決定數(shù)據(jù)是否應該被提交，稱為任務級質(zhì)量檢查者。Fork Operator ForkOperator是一種類型的控制操作。其允許任務流程分支為多個流，其中每個進入一個單獨配置的接收器。對于數(shù)據(jù)記錄需要被寫入到多個不同的存儲器中是非常有用的，或者該數(shù)據(jù)記錄需要寫出到相同的存儲（例如，HDFS

9、）不同的下游消費者使用。Data Writer 一個DataWriter負責寫入數(shù)據(jù)記錄到它被設(shè)定的位置上，Gobblin向外提供的Avro數(shù)據(jù)寫入器用Avro的格式將數(shù)據(jù)寫入到HDFS。用戶能有他們自己的DataWriters插件通過設(shè)定用DataWriterBuilder class在作業(yè)配置中，通過在配置文件中設(shè)置DataWriterBuilder類來創(chuàng)建屬于用戶自己的DataWriter插件。Data Publisher DataPublisher是負責發(fā)布Gobblin工作的提取數(shù)據(jù)。 Gobblin使用一個默認的DataPublisher和一個以文件為基礎(chǔ)的DataWriter工作

10、，例如AvroHdfsDataWriter，移動數(shù)據(jù)從工作各項任務的輸出目錄到最終作業(yè)輸出的目錄Gobblin Architecture 任務流程Gobblin Architecture 作業(yè)狀態(tài)管理 gobblin作業(yè)周期性的運行在某些調(diào)度中和每個增量抽取數(shù)據(jù)的作業(yè)中，抽取新的數(shù)據(jù)或者改變存在的數(shù)據(jù)。實現(xiàn)增量抽取。gobblin必須持久化狀態(tài)在完成每個作業(yè)之后和載入前一個運行的狀態(tài)作為下一次運行任務的開始。Gobblin保持狀態(tài)存儲，負責的工作狀態(tài)持久性。Gobblin作業(yè)每次運行讀取上次運行狀態(tài)的狀態(tài)存儲和完成后寫入自己的狀態(tài)到存儲。gobblin作業(yè)的運行狀態(tài)由作業(yè)配置，一些屬性設(shè)置在運

11、行的作業(yè)上或任務級別上。 每次作業(yè)運行，gobblin使用一個實現(xiàn)的狀態(tài)存儲在hadoop SequenceFiles上序列作業(yè)和任務狀態(tài)。每個作業(yè)都有一個單獨的目錄，它的工作和任務狀態(tài)存儲在SequenceFiles 上。在其基于文件序列狀態(tài)存儲所在的文件系統(tǒng)是可配置的Gobblin Architecture 失敗處理 作為一個容錯數(shù)據(jù)攝取框架，gobblin部署了多個防御級別對作業(yè)和任務的失敗。對于作業(yè)失敗，gobblin跟蹤工作連續(xù)失敗的數(shù)量和可選的,如果數(shù)量超過定義的閥值會發(fā)送警告郵件，以便作業(yè)的所有者可以跳過和檢查故障發(fā)出的報警郵件。對于任務失敗，gobblin會嘗試失敗任務在作業(yè)配

12、置的最大次數(shù)。除此之外，Gobblin還提供了一個選項，以使相應的跨作業(yè)運行失敗任務的workUnit重試。如果一個任務在所有重試都失敗，則WorkUnit在此基礎(chǔ)上被自動創(chuàng)建任務包括下次運行的作業(yè)如果這個重啟類型開啟的話，這種重啟類型是非常有用的在處理間歇性故障例如臨時數(shù)據(jù)源的宕機。Gobblin Architecture 任務調(diào)度 Gobblin集成使用作業(yè)調(diào)度例如Azkaban,Oozie,或者quartz。 Gobblin的一個重要特征是，它分離作業(yè)調(diào)度器和作業(yè)被調(diào)度器調(diào)度，使得不同的jobs可以以不同的部署設(shè)置運行預定的作業(yè)。這是通過使用具有用于不同部署設(shè)置不同的實現(xiàn)抽象JobLau

13、ncher實現(xiàn)。例如：一個程序可能有預定晚五的工作：其中2個在同一主機使用本地JobLauncher調(diào)度上本地運行，而使用MRJob啟動某個Hadoop集群的其余3運行。JobLauncher使用可以使用屬性launcher.type簡單地配置。多個調(diào)度器是互不影響的。Gobblin Architecture 單例模式部署使用場景 單例環(huán)境下使用gobblin最好是用于輕量級的數(shù)據(jù)庫抽取運行架構(gòu)圖描述 JobScheduler 作為一個守護線程運行，本身包含一個線程池，其作用是實例化jobLauncher運行每一個job作業(yè)，jobLauncher可以在每個job作業(yè)基礎(chǔ)上配置，故jobSch

14、eduler可以運行不同模式的jobLauncher（包括LocalJobLauncher和MRJobLauncher）。Gobblin Deployment 單例模式部署描述 每個LocalJobLauncher運行和管理一些組件對于每個gobblin作業(yè)抽取任務。TaskExecutor負責抽取任務在一個線程池中，其大小由每個job的配置決定。LocalTaskStateTracker負責記錄運行任務狀態(tài)的軌跡和更新任務指標，LocalJobLauncher運行組件如下： 首先：啟動TaskExecutor和LocalTaskStateTracker；其次：在作業(yè)配置中創(chuàng)建Source實例

15、并得到一系列的WorkUnits;其次：對于每個workunit創(chuàng)建一個任務，并且這些任務注冊在LocalTaskStateTracker上，提交任務在TaskExecutor運行；其次：等待所有提交的任務執(zhí)行完；最后：收集狀態(tài)并持久化到狀態(tài)存儲里，發(fā)布抽取的數(shù)據(jù)。部署 workdir:gobblin作業(yè)的工作目錄，主要存儲臨時數(shù)據(jù)、正式數(shù)據(jù)、狀態(tài)和任務指標。如果不設(shè)置，默認是環(huán)境變量配置的GOBBLIN_WORK_DIR目錄。Gobblin Deployment 單例模式部署部署 全局配置可以定義到 perties中，對于每個job作業(yè)需要單獨再定

16、義一個配置*.pull文件或者*.job文件。 conf :配置每一個job運行的基本配置，配置文件以.pull或.job后綴，如果沒指定，默認獲取環(huán)境變量中GOBBLIN_JOB_CONFIG_DIR的目錄。 .pull或.job配置屬性有： 作業(yè)名稱job.group 作業(yè)組source.class 源類converter.classes 轉(zhuǎn)換類最外層配置會被最里層的相同的配置覆蓋root_job_config_dir/ perties foo/ foo1.job foo2.job perties bar/ bar1.job bar2.jo

17、b perties baz/ baz1.pull baz2.pull pertiesGobblin Deployment Hadoop模式使用場景 TB級數(shù)據(jù)抽取運行架構(gòu)圖Gobblin Deployment Hadoop模式描述 MRJobLauncher用于啟動和運行g(shù)obblin作業(yè)在hadoop Mapreduce中。 首先：在作業(yè)配置中創(chuàng)建Source實例并得到一系列的WorkUnits；其次：序列化每一個workunit文件在hdfs上，后續(xù)被mapper讀取；其次：創(chuàng)建一個文件目錄存儲序列工作單元中的文件的路徑；其次：創(chuàng)建和配置一個僅有mapper任務

18、的mapreduce作業(yè)讀取第三步創(chuàng)建的文件作為輸入；啟動MapReduce工作到選擇的集群上運行，并等待它完成；一旦MapReduce工作完成后，收集任務狀態(tài)和持久化他們的狀態(tài)存儲，發(fā)布提取的數(shù)據(jù)。 一個mapper任務可以運行一個或者多個gobblin任務，這主要取決于設(shè)定的最大的mapper數(shù)量，每一個任務對應于一個workunit，當一個mapper運行多個gobblin任務的時候，workunit需要設(shè)置成MultiWorkUnit。 一個mapper運行g(shù)obblin任務的步驟： 1、開始一個TaskExecutor（負責執(zhí)行任務在一個可配置的線程池里）和MRTaskStateTr

19、acker（負責記錄一個gobblin任務的狀態(tài)運行軌跡在mapper中）。 2、讀取下一個輸入記錄是存儲序列化的工作單位的文件路徑。 3、反序列化工作單元并將其添加到工作單元做的名單。如果輸入的是一個MultiWorkUnit，它包裝的工作單元都添加到列表中。步驟2和3重復進行，直到所有的分配任務包被解串，并添加到列表中。 4、對于在workunit列表中的每一個workunit，為每一個workunit創(chuàng)建一個任務，在MRTaskStateTracker注冊任務，TaskExecutor運行提交的任務，注意，任務可并行運行，如果所述的TaskExecutor被構(gòu)造以在其線程池多個線程。Go

20、bblin Deployment Hadoop模式描述 5、等待所有提交的任務完成。 6、在所有的提交的任務完成后，當MRJobLauncher收集任務狀態(tài)的時候，其會讀取寫出每個任務的狀態(tài)的文件. 7、讓我們再回到步驟2并讀取下一個輸入記錄（如果可用）。部署 公共配置可放在perties中, fs.uri在該文件中配置。 jt配置集群的jobTracker地址 fs配置集群的namenode地址 HADOOP_BIN_DIR需要在環(huán)境變量中設(shè)置Gobblin Deployment Quartz gobblin本身集成了Quartz scheduler調(diào)度做為JobScheduler的一部分，JobScheduler組合在了SchedulerDaemon（ A class that runs the JobScheduler in a daemon process for standalone deploymen