2. 程式人生 > >基於Hadoop生態圈的資料倉庫實踐 —— ETL（三）

基於Hadoop生態圈的資料倉庫實踐 —— ETL（三）

阿新 • • 發佈：2019-01-29
三、使用Oozie定期自動執行ETL
1. Oozie簡介
（1）Oozie是什麼
        Oozie是一個管理Hadoop作業、可伸縮、可擴充套件、可靠的工作流排程系統，其工作流作業是由一系列動作構成的有向無環圖（DAGs），協調器作業是按時間頻率週期性觸發的Oozie工作流作業。Oozie支援的作業型別有Java map-reduce、Streaming map-reduce、Pig、 Hive、Sqoop和Distcp，及其Java程式和shell指令碼等特定的系統作業。
        第一版Oozie是一個基於工作流引擎的伺服器，通過執行Hadoop Map/Reduce和Pig作業的動作執行工作流作業。第二版Oozie是一個基於協調器引擎的伺服器，按時間和資料觸發工作流執行。它可以基於時間（如每小時執行一次）或資料可用性（如等待輸入資料完成後再執行）連續執行工作流。第三版Oozie是一個基於Bundle引擎的伺服器。它提供更高級別的抽象，批量處理一系列協調器應用。使用者可以在bundle級別啟動、停止、掛起、繼續、重做協調器作業，這樣可以更好地簡化操作控制。
（2）為什麼需要Oozie
  • 在Hadoop中執行的任務有時候需要把多個Map/Reduce作業連線到一起執行，或者需要多個作業並行處理。Oozie可以把多個Map/Reduce作業組合到一個邏輯工作單元中，從而完成更大型的任務。
  • 從排程的角度看，如果使用crontab的方式呼叫多個工作流作業，可能需要編寫大量的指令碼，還要通過指令碼來控制好各個工作流作業的執行時序問題，不但指令碼不好維護，而且監控也不方便。基於這樣的背景，Oozie提出了Coordinator的概念，它能夠將每個工作流作業作為一個動作來執行，相當於工作流定義中的一個執行節點，這樣就能夠將多個工作流作業組成一個稱為Coordinator Job的作業，並指定觸發時間和頻率，還可以配置資料集、併發數等。
（3）Oozie的體系結構（摘自http://www.infoq.com/cn/articles/introductionOozie/）
        Oozie的體系結構如下圖所示。


        Oozie是一種Java Web應用程式，它執行在Java servlet容器——即Tomcat——中，並使用資料庫來儲存以下內容：

  • 工作流定義
  • 當前執行的工作流例項，包括例項的狀態和變數
        Oozie工作流是放置在控制依賴DAG（有向無環圖 Direct Acyclic Graph）中的一組動作（例如，Hadoop的Map/Reduce作業、Pig作業等），其中指定了動作執行的順序。我們會使用hPDL（一種XML流程定義語言）來描述這個圖。
        hPDL是一種很簡潔的語言，只會使用少數流程控制和動作節點。控制節點會定義執行的流程，幷包含工作流的起點和終點（start、end和fail節點）以及控制工作流執行路徑的機制（decision、fork和join節點）。動作節點是一些機制，通過它們工作流會觸發執行計算或者處理任務。Oozie為以下型別的動作提供支援： Hadoop map-reduce、Hadoop檔案系統、Pig、Java和Oozie的子工作流（SSH動作已經從Oozie schema 0.2之後的版本中移除了）。
        所有由動作節點觸發的計算和處理任務都不在Oozie之中——它們是由Hadoop的Map/Reduce框架執行的。這種方法讓Oozie可以支援現存的Hadoop用於負載平衡、災難恢復的機制。這些任務主要是非同步執行的（只有檔案系統動作例外，它是同步處理的）。這意味著對於大多數工作流動作觸發的計算或處理任務的型別來說，在工作流操作轉換到工作流的下一個節點之前都需要等待，直到計算或處理任務結束了之後才能夠繼續。Oozie可以通過兩種不同的方式來檢測計算或處理任務是否完成，也就是回撥和輪詢。當Oozie啟動了計算或處理任務的時候，它會為任務提供唯一的回撥URL，然後任務會在完成的時候傳送通知給特定的URL。在任務無法觸發回撥URL的情況下（可能是因為任何原因，比方說網路閃斷），或者當任務的型別無法在完成時觸發回撥URL的時候，Oozie有一種機制，可以對計算或處理任務進行輪詢，從而保證能夠完成任務。
        Oozie工作流可以引數化（在工作流定義中使用像${inputDir}之類的變數）。在提交工作流操作的時候，我們必須提供引數值。如果經過合適地引數化（比方說，使用不同的輸出目錄），那麼多個同樣的工作流操作可以併發。
        一些工作流是根據需要觸發的，但是大多數情況下，我們有必要基於一定的時間段和（或）資料可用性和（或）外部事件來執行它們。Oozie協調系統（Coordinator system）讓使用者可以基於這些引數來定義工作流執行計劃。Oozie協調程式讓我們可以以謂詞的方式對工作流執行觸發器進行建模，那可以指向資料、事件和（或）外部事件。工作流作業會在謂詞得到滿足的時候啟動。
        經常我們還需要連線定時執行、但時間間隔不同的工作流操作。多個隨後執行的工作流的輸出會成為下一個工作流的輸入。把這些工作流連線在一起，會讓系統把它作為資料應用的管道來引用。Oozie協調程式支援建立這樣的資料應用管道。

（4）CDH 5.7.0中的Oozie

        CDH 5.7.0中，Oozie的版本是4.1.0，元資料儲存使用MySQL。關於CDH 5.7.0中Oozie的屬性，參考以下連結：
https://www.cloudera.com/documentation/enterprise/latest/topics/cm_props_cdh570_oozie.html

2. 建立定期裝載工作流
（1）修改資源配置
        需要將以下兩個引數的值調高：
yarn.nodemanager.resource.memory-mb = 2000
yarn.scheduler.maximum-allocation-mb = 2000
        否則會在執行工作流作業時報類似下面的錯誤：
org.apache.oozie.action.ActionExecutorException: JA009: org.apache.hadoop.yarn.exceptions.InvalidResourceRequestException: Invalid resource request, requested memory < 0, or requested memory > max configured, requestedMemory=1536, maxMemory=1500
        具體做法是，從CDH Web控制檯修改相關引數，儲存更改並重啟叢集。
        yarn.nodemanager.resource.memory-mb引數在YARN服務的NodeManager範圍裡，如下圖所示。
        yarn.scheduler.maximum-allocation-mb引數在YARN服務的ResourceManager範圍裡，如下圖所示。
        從Web控制檯重啟叢集的介面如下圖所示。
（2）啟用Oozie Web Console
        預設配置時，Oozie Web Console是禁用的，為了後面方便監控Oozie作業的執行，需要將其改為啟用。“啟用 Oozie 伺服器 Web 控制檯”引數在Oozie服務的主要範圍裡，如下圖所示。


        具體的做法是：

  • 下載安裝ext-2.2。
  • 從CDH Web控制檯修改相關引數，儲存更改並重啟Oozie服務。
        詳細步驟參考以下連結：http://www.cloudera.com/documentation/enterprise/5-5-x/topics/admin_oozie_console.html

（3）啟動sqoop的share metastore service
        定期裝載工作流需要用Oozie呼叫Sqoop執行，這需要開啟Sqoop元資料共享儲存，命令如下：
sqoop metastore > /tmp/sqoop_metastore.log 2>&1 &
        關於Oozie無法執行Sqoop Job的問題，參考以下連結：http://www.lamborryan.com/oozie-sqoop-fail/

（4）連線metastore重建sqoop job
        前面建立的sqoop job，其元資料並沒有儲存在share metastore裡，所以需要使用以下的命令重建。
last_value=`sqoop job --show myjob_incremental_import | grep incremental.last.value | awk '{print $3}'`
sqoop job --delete myjob_incremental_import
sqoop job \
--meta-connect jdbc:hsqldb:hsql://cdh2:16000/sqoop \
--create myjob_incremental_import \
-- \
import \
--connect "jdbc:mysql://cdh1:3306/source?useSSL=false&user=root&password=mypassword" \
--table sales_order \
--columns "order_number, customer_number, product_code, order_date, entry_date, order_amount" \
--hive-import \
--hive-table rds.sales_order \
--incremental append \
--check-column order_number \
--last-value $last_value
        其中$last-value是上次ETL執行後的值。

（5）定義工作流
        建立內容如下的workflow.xml檔案：
<?xml version="1.0" encoding="UTF-8"?>
<workflow-app xmlns="uri:oozie:workflow:0.1" name="regular_etl">
    <start to="fork-node"/>
    <fork name="fork-node">
        <path start="sqoop-customer" />
        <path start="sqoop-product" />
        <path start="sqoop-sales_order" />
    </fork>
    <action name="sqoop-customer">
        <sqoop xmlns="uri:oozie:sqoop-action:0.2">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <arg>import</arg>
            <arg>--connect</arg>
            <arg>jdbc:mysql://cdh1:3306/source?useSSL=false</arg>
            <arg>--username</arg>
            <arg>root</arg>
            <arg>--password</arg>
            <arg>mypassword</arg>
            <arg>--table</arg>
            <arg>customer</arg>
            <arg>--hive-import</arg>
            <arg>--hive-table</arg>
            <arg>rds.customer</arg>
            <arg>--hive-overwrite</arg>            
            <file>/tmp/hive-site.xml#hive-site.xml</file>
            <archive>/tmp/mysql-connector-java-5.1.38-bin.jar#mysql-connector-java-5.1.38-bin.jar</archive>
        </sqoop>
        <ok to="joining"/>
        <error to="fail"/>
    </action>
	<action name="sqoop-product">
        <sqoop xmlns="uri:oozie:sqoop-action:0.2">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <arg>import</arg>
            <arg>--connect</arg>
            <arg>jdbc:mysql://cdh1:3306/source?useSSL=false</arg>
            <arg>--username</arg>
            <arg>root</arg>
            <arg>--password</arg>
            <arg>mypassword</arg>
            <arg>--table</arg>
            <arg>product</arg>
            <arg>--hive-import</arg>
            <arg>--hive-table</arg>
            <arg>rds.product</arg>
            <arg>--hive-overwrite</arg>            
            <file>/tmp/hive-site.xml#hive-site.xml</file>
            <archive>/tmp/mysql-connector-java-5.1.38-bin.jar#mysql-connector-java-5.1.38-bin.jar</archive>
        </sqoop>
        <ok to="joining"/>
        <error to="fail"/>
    </action>
    <action name="sqoop-sales_order">
        <sqoop xmlns="uri:oozie:sqoop-action:0.2">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <command>job --exec myjob_incremental_import --meta-connect jdbc:hsqldb:hsql://cdh2:16000/sqoop</command>
            <file>/tmp/hive-site.xml#hive-site.xml</file>
            <archive>/tmp/mysql-connector-java-5.1.38-bin.jar#mysql-connector-java-5.1.38-bin.jar</archive>
        </sqoop>
        <ok to="joining"/>
        <error to="fail"/>
    </action>
    <join name="joining" to="hive-node"/>
    <action name="hive-node">
        <hive xmlns="uri:oozie:hive-action:0.2">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node> 
            <job-xml>/tmp/hive-site.xml</job-xml>
            <script>/tmp/regular_etl.sql</script>
        </hive>
        <ok to="end"/>
        <error to="fail"/>
    </action>
    <kill name="fail">
        <message>Sqoop failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>
    <end name="end"/>
</workflow-app>
        其DAG如下圖所示。

        該工作流包括9個節點，其中有5個控制節點，4個動作節點：工作流的起點（start）、終點（end）、失敗處理節點（fail，DAG圖中未顯示），兩個執行路徑控制節點（fork-node和joining，fork與join節點必須成對出現），三個並行處理的Sqoop行動節點（sqoop-customer、sqoop-product、sqoop-sales_order）用作資料抽取，一個Hive行動節點（hive-node）用作資料轉換與裝載。
（6）部署工作流
hdfs dfs -put -f workflow.xml /user/root/
hdfs dfs -put /etc/hive/conf.cloudera.hive/hive-site.xml /tmp/
hdfs dfs -put /root/mysql-connector-java-5.1.38/mysql-connector-java-5.1.38-bin.jar /tmp/
hdfs dfs -put /root/regular_etl.sql /tmp/
（7）建立作業屬性檔案
        建立內容如下的job.properties檔案：
nameNode=hdfs://cdh2:8020
jobTracker=cdh2:8032
queueName=default
oozie.use.system.libpath=true
oozie.wf.application.path=${nameNode}/user/${user.name}
（8）執行工作流
oozie job -oozie http://cdh2:11000/oozie -config /root/job.properties -run
        此時從Oozie Web Console可以看到正在執行的作業，如下圖所示。
        點選作業所在行，可以開啟作業的詳細資訊視窗，如下圖所示。
        點選動作所在行，可以開啟動作的詳細資訊視窗，如下圖所示。
        可以點選Console URL右側的圖示，可以開啟Map/Reduce作業的跟蹤視窗，如下圖所示。
        當Oozie作業執行完，可以在“All Jobs”標籤頁看到，Status列已經從RUNNING變為SUCCEEDED，如下圖所示。
        此時檢視cdc_time表的資料，可以看到日期已經改為當前日期，如下圖所示。

3. 建立協調作業定期自動執行工作流
（1）建立協調作業屬性檔案
        建立內容如下的job-coord.properties檔案：
nameNode=hdfs://cdh2:8020
jobTracker=cdh2:8032
queueName=default
oozie.use.system.libpath=true
oozie.coord.application.path=${nameNode}/user/${user.name}
timezone=UTC
start=2016-07-11T06:00Z
end=2020-12-31T07:15Z
workflowAppUri=${nameNode}/user/${user.name}
（2）建立協調作業配置檔案
        建立內容如下的coordinator.xml檔案：
<coordinator-app name="regular_etl-coord" frequency="${coord:days(1)}" start="${start}" end="${end}" timezone="${timezone}" xmlns="uri:oozie:coordinator:0.1">
    <action>
        <workflow>
            <app-path>${workflowAppUri}</app-path>
            <configuration>
                <property>
                    <name>jobTracker</name>
                    <value>${jobTracker}</value>
                </property>
                <property>
                    <name>nameNode</name>
                    <value>${nameNode}</value>
                </property>
                <property>
                    <name>queueName</name>
                    <value>${queueName}</value>
                </property>
            </configuration>
        </workflow>
    </action>
</coordinator-app>
（3）部署協調作業
hdfs dfs -put -f coordinator.xml /user/root/
（4）執行協調作業
oozie job -oozie http://cdh2:11000/oozie -config /root/job-coord.properties -run
        此時從Oozie Web Console可以看到準備執行的協調作業，作業的狀態為PREP，如下圖所示。
        此協調作業自2016年7月11日開始，每天14點執行一次。結束日期非常晚，這裡設定的是2020年12月31日。需要注意一下時區的設定。Oozie預設的時區是UTC，而且即便在屬性檔案中設定了timezone=GMT+0800也不起作用，所以start屬性設定的是06:00，實際就是北京時間14:00。
        當時間到達14:00時，協調作業開始執行，狀態由PREP變為RUNNING，如下圖所示。
        點選作業所在行，可以開啟協調作業的詳細資訊視窗，如下圖所示。
        點選協調作業所在行，可以開啟工作流作業的詳細資訊視窗，如下圖所示。
        點選動作所在行，可以開啟動作的詳細資訊視窗，如下圖所示。
        可以點選Console URL右側的圖示，可以開啟Map/Reduce作業的跟蹤視窗，如下圖所示。
        至此介紹了使用Oozie定期自動執行ETL的一般方法。Oozie 4.1.0的官方文件連結地址如下：http://oozie.apache.org/docs/4.1.0/index.html

相關推薦

基於Hadoop生態圈的資料倉庫實踐 —— ETL（三）

三、使用Oozie定期自動執行ETL1. Oozie簡介（1）Oozie是什麼        Oozie是一個管理Hadoop作業、可伸縮、可擴充套件、可靠的工作流排程系統，其工作流作業是由一系列動作構成的有向無環圖（DAGs），協調器作業是按時間頻率週期性觸發的Oozie工

Hadoop生態圈-phoenix的視圖（view）管理

oop 視圖 聲明 版權 作品 管理 原創 view 創作 　　　　　　　　　　　　　　　　　　　　　　Hadoop生態圈-phoenix的視圖（view）管理 　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　作者：尹正傑 版權

專欄筆記：基於hadoop生態圈的資料倉庫實踐

概述 資料倉庫是一個面向主題的、整合的、隨時間變化的、非易失的資料集合，用於支援決策。它主要的目標是分析和處理資料，和傳統的操作型事務處理有很大區別。 之所以不直接在操作型系統上執行分析查詢，而是從操作型系統抽取資料，最主要有以下兩個原因：（1）在操作型系統上直接執行

基於Hadoop生態圈的資料倉庫實踐 —— 環境搭建（三）

三、建立資料倉庫示例模型        Hadoop及其相關服務安裝配置好後，下面用一個小而完整的示例說明多維模型及其相關ETL技術在Hadoop上的具體實現。1. 設計ERD        操作型系統是一個銷售訂單系統，初始時只有產品、客戶、訂單三個表，ERD如下圖所示。 

基於Hadoop生態圈的資料倉庫實踐 —— 環境搭建（二）

二、安裝Hadoop及其所需的服務 1. CDH安裝概述 CDH的全稱是Cloudera's Distribution Including Apache Hadoop，是Cloudera公司的Hadoop分發版本。有三種方式安裝CDH： . Path A - 通過Cloud

hive(01)、基於hadoop叢集的資料倉庫Hive搭建實踐

在前面hadoop的一系列文中，我們對hadoop有了初步的認識和使用，以及可以搭建完整的叢集和開發簡單的MapReduce專案，下面我們開始學習基於Hadoop的資料倉庫Apache Hive，將結構化的資料檔案對映為一張資料庫表，將sql語句轉換為MapReduce任務

ElasticSearch實踐系列（三）：探索資料

前言 經過前兩篇文章得實踐，我們已經瞭解了ElasticSearch的基礎知識，本篇文章讓我來操作一些更真實的資料集。我們可以利用www.json-generator.com/生成如下的文件結構： { "account_number": 1, "balance": 39225,

Linux入門實踐筆記（三）——資料盤格式化和和多分割槽掛載

[[email protected] ~]$ sudo fdisk /dev/vdb Welcome to fdisk (util-linux 2.23.2). ​ Changes will remain in memory only, until you decide to write them.

hadoop spark 大資料叢集環境搭建（一）

大資料雲端計算現在比較熱門，未來的一個發展方向，在此分享下技術,有不對的地方歡迎指出 1、軟體環境(會分享到網盤) centos6.5 jdk1.7 hadoop2.4.1(這裡只用到hdfs,namenode不走ha) zookeeper3.4.5 spark1.3.0

大型資料倉庫的治理（1）- 資料需求響應慢

文/通貫 【導讀】這是共四篇的資料倉庫治理系列，本文是第一篇。作者從實際經驗中，總結出了一些大型資料倉庫治理中，可能會遇到的問題。本文談到了“資料需求響應慢”的問題，大資料時代，你值得關注。      資料倉庫是繼ERP之後失敗率最高的IT專案。在專案還沒立項的時候，會給老闆呈現各種美好。而實施到

【day 12】python程式設計：從入門到實踐學習筆記-基於Django框架的Web開發-Django入門（三）

學習筆記目錄 第十八章 Django入門（三） 建立網頁：學習筆記主頁 使用django建立網頁通常分三個階段：定義URL、編寫檢視和編寫模板。 首先必須定義URL模式，其描述了URL是

一次資料倉庫報表測試（2）

1.背景       最近終於將這個專案測試結束了，之前寫過一篇文章，寫的是測試過程中遇到的問題，感興趣的同學可有先去看看上一篇文章。 2.目的       專案結束後問題也沒有得到根本解決。寶路由此引發了一些

利用Data Vault對資料倉庫進行建模（二）

寫在前面 本篇先不討論Data Vault其本身，因為不見得所有人都接受這個。但是裡邊有一些很不錯的東西跟主流的資料倉庫方法是有共同點的，所以這裡主要討論這些共同的方法，在筆者看來，無論是Kimball還是DV，這些方法都是很有用的。這個系列為作者本人哥本哈士奇的個人理解和總結，可能會有理解上的偏差，也歡迎大

基於centos6的mysql5.7.13主從部署（三）

linux mysql主從 部署基於centos6的mysql5.7.13主從部署（三） 主從測試登入（主）mysql1[[email protected]/* */ ~]# mysql -uroot -p123456mysql> unlock tables; #解鎖，對應上一章節的表的

基於Html5 Plus + Vue + Mui 移動App開發（三）-文件操作（讀取、保存、更新數據）

mes das parse 移動 分享 gif ont 包括 ole 　　實全資訊采用基於Html5 Plus + Vue + Mui 移動App。主要實現功能包括： 實現搜索站點設置 實現搜索關鍵字定義 實現搜索資訊保存、刪除功能。 主界面實現關鍵字搜

基於深度學習的目標檢測演算法綜述（三）（截止20180821）

參考：https://zhuanlan.zhihu.com/p/40102001 基於深度學習的目標檢測演算法綜述分為三部分： 1. Two/One stage演算法改進。這部分將主要總結在two/one stage經典網路上改進的系列論文，包括Faster R-CNN、YOLO、SSD等經

依賴注入的兩種常用方式(構造器和Setter)與注入內容(裝配資料)——Spring IOC/DI（三）

本章主要講解一下Spring中依賴注入方式，接上一章依賴注入原理和方式： https://blog.csdn.net/qq_34598667/article/details/83315669 依賴注入常用的兩種方式以及注入的內容(裝配資料) Spring通過DI（依賴注入）實現I

React最佳實踐嘗試（三）

github地址：github.com/bbwlfx/ts-b… 第一個頁面 配置完畢之後，接下來就開始開發一個簡單的Demo頁面吧~ 首先要定義好Demo的model模型： models/demo.ts import { demoModalState } from "typings"; i

基於中臺思想的物流系統設計（三）：構建物流地址能力

一、引言 在電商物流領域我們會涉及到地址，其中包括了基礎的四級地址和使用者填寫的地址。四級地址在整個從下單到收貨的業務流程中都會用到，因此設計的時候要考慮如何最大限度地提高QPS。使用者地址在下單的時候讓使用者填寫或者選擇，然後存在交易訂單和物流訂單上，後續的流程一般不會變，如果使用者需要修改地址，直接變

.net core 實踐筆記（三）--封裝底層

初始 tex 簡單的 github 個數 cor net ati .json 前言： 　　有了前面的工作，簡單的架子基本搭建起來了，因為條件有限，只能先測試SqlServer的了，源碼放出來，也希望有興趣的夥伴可以一起改善，相信可以成為未來進階架構師的第一步，自己有小項目的