快速構建基於Lambda框架大數據業務架構的TBDS Demo指南

• 2019 年 11 月 4 日
• 筆記

1 前言

本文介紹了如何快速的構建一個基於Lambda框架的大數據業務場景的TBDS Demo，用於展示如何利用其對應的工具在騰訊大數據平台上搭建出實時計算、多維分析、離線計算的lambda框架demo。

典型基於Lambda框架的大數據業務架構，如下圖所示：

關於TBDS

騰訊大數據處理套件（Tencent Big Data Suite，TBDS）是基於騰訊多年海量數據處理經驗，對外提供的可靠、安全、易用的大數據處理平台。更多資訊請點擊鏈接。

1.1 讀者對象

本文適用的讀者對象為：

• 解決方案架構師
• 產品架構師
• 交付工程師

1.2 Demo組成

本文提供的指南，是用來快速構建一個基於Lambda架構的Demo，主要包含以下三部分：

• 基於Oceanus的實時計算Demo
• 基於Hermes的多維分析Demo
• 基於工作流的任務調度Demo

數據流轉圖如下所示：

1.3 文檔目的

本文旨在達到以下2個目的：

• 在產品售賣的售前階段，用於向客戶展示產品特性；
• 在產品使用的培訓期間，用於快速構建開發環境，標準化講解流程。

1.4 前提條件

對於使用此Demo的人員，基本的要求是：

• 理解實時計算、多維分析、任務調度的概念；
• 掌握Oceanus、Hermes、工作流的基本原理/使用方式。

1.5 工具獲取

目前Demo已實現腳本化一鍵式執行，腳本封裝於tbds_lambda_demo工具中，可在本文附件中獲取。

工具需要在TBDS集群的Portal主節點上運行，上傳並解壓工具到Portal主節點的任意目錄，例如/opt：

mv tbds_lambda_demo.zip /opt  unzip /opt/tbds_lambda_demo.zip

1.6 結構說明

工具的目錄結構說明如下

|-- tbds_lambda_demo                                             # 工具主目錄  |   |-- bin                                                      # 存放腳本的目錄  |   |   |-- check_kafka_topic_data.sh                            # 檢查kafka topic數據的腳本  |   |   |-- check_realtime_calculate_result.sh                   # 檢查實時計算結果的腳本  |   |   |-- get_access_key.py                                    # 生成Portal認證資訊的腳本  |   |   |-- hermes_demo_test.sh                                  # hermes demo腳本  |   |   |-- how_to_use_utils_example.sh                          # 解釋如何使用utils腳本的腳本  |   |   |-- init_hbase.sh                                        # 初始化hbase的腳本  |   |   |-- init_kafka.sh                                        # 初始化kafka的腳本  |   |   |-- init_mysql.sh                                        # 初始化mysql的腳本  |   |   |-- init_oceanus_mysql.sh                                # 初始化oceanus mysql的腳本  |   |   |-- init_oceanus.sh                                      # 初始化oceanus的腳本  |   |   |-- prepare_kafka.sh                                     # 準備kafka數據的腳本  |   |   |-- realtime_data_collector.sh                           # 實時數據收集的腳本  |   |   |-- realtime_data_generator.sh                           # 實時數據生成的腳本  |   |   |-- send_message.sh                                      # 發送消息的腳本  |   |   |-- utils.sh                                             # 通用函數的腳本  |   |   |-- workflow_demo_clean_all.sh                           # 工作流環境清理全局腳本  |   |   |-- workflow_demo_clean_sth.sh                           # 工作流環境清理腳本  |   |   |-- workflow_demo_functions.sh                           # 工作流相關函數的腳本  |   |   |-- workflow_demo_init_all.sh                            # 工作流環境初始化全局腳本  |   |   |-- workflow_demo_init_sth.sh                            # 工作流環境初始化腳本  |   |-- conf                                                     # 配置目錄  |   |   |-- clean_hive_wf.sql                                    # 清理工作流hive表的SQL文件  |   |   |-- clean_mysql_wf.sql                                   # 清理工作流mysql表的SQL文件  |   |   |-- client_plain.properties                              # kafka身份認證資訊文件  |   |   |-- hermes_demo_test.conf                                # hermes demo配置文件  |   |   |-- init_hbase.cmd                                       # 初始化hbase配置文件  |   |   |-- init_hive_db_wf.sql                                  # 初始化工作流hive資料庫的sql文件  |   |   |-- init_hive_tables_wf.sql                              # 初始化工作流hive數據表的sql文件  |   |   |-- init_mysql.sql                                       # 初始化mysql的sql文件  |   |   |-- init_mysql_wf.sql                                    # 初始化工作流mysql的sql文件  |   |   |-- init_oceanus_jobs.sql                                # 初始化oceanus job的sql文件  |   |   |-- init_oceanus_jobs.sql_bak                            # 初始化oceanus job的sql文件  |   |   |-- init_oceanus_jobs.sql_clean                          # 初始化oceanus job的sql文件  |   |   |-- init_oceanus_jobs_template.sql                       # 初始化oceanus job的sql文件  |   |   |-- tbds.api.json                                        # 生成Portal認證資訊的文件  |   |   |-- tools.conf                                           # 實時demo配置文件  |   |   |-- wf_demo.conf                                         # 工作流demo配置文件  |   |   |-- wf_demo_hermes_test.conf                             # hermes demo配置文件  |   |   |-- wf_demo_hive_cal.sql                                 # 工作流hive sql文件  |   |   |-- wf_demo_hive_query.sql                               # 工作流hive sql文件  |   |   |-- workflow_hermesdemo.xml                              # 工作流定義文件  |   |   |-- workflow_latest.xml                                  # 工作流定義文件  |   |-- prepare_data                                             # 示例數據文件夾  |   |   |-- age_info.dat                                         # 表age_info數據文件  |   |   |-- city_id.dat                                          # 表city_id數據文件  |   |   |-- city_info.dat                                        # 表city_info數據文件  |   |   |-- contact.dat                                          # 表contact數據文件  |   |   |-- country_info.dat                                     # 表country_info數據文件  |   |   |-- dimension_table_data.sql                             # 維表數據文件  |   |   |-- education.dat                                        # 表education數據文件  |   |   |-- gender_id.dat                                        # 表gender_id數據文件  |   |   |-- gender_info.dat                                      # 表gender_info數據文件  |   |   |-- id_name.txt                                          # id、name數據文件  |   |   |-- marriage.dat                                         # 表marriage數據文件  |   |   |-- occupation.dat                                       # 表occupation數據文件  |   |   |-- only_id.txt                                          # id數據文件  |   |   |-- only_name.txt                                        # name數據文件

2 準備工作

2.1 事項說明

在運行TBDS Lambda Demo之前，需要先準備完成下述事項。

2.1.1 創建用戶、項目、密鑰等

1. 創建Portal用戶lambda_demo_user；
2. 創建Portal項目lambda_demo_project；
3. 將用戶lambda_demo_user添加為項目lambda_demo_project的成員，角色為系統管理員、項目管理員、項目開發、項目運維；
4. 為用戶lambda_demo_user創建一個hadoop模組的密鑰。

2.1.2 創建策略

1. 創建HBase表訪問策略
   • 策略名：hbase_4_lambda_demo_user
   • 表：customers_details_hbase
   • 表的列族：*
   • 表的列：*
   • 策略動作：允許訪問
   • 用戶：lambda_demo_user
   • 許可權：read/write/create/admin
2. 創建Kafka Topic訪問策略
3. 策略名：topic_4_lambda_demo_user
4. Topic名：customers_balance_details_topic、customers_info_details_topic
5. 策略動作：允許訪問
6. 用戶：lambda_demo_user
7. 許可權：publish/consume/configure/describe/create/delete/kafka admin

2.2 初始化

準備事項的工作已完全實現腳本化，通過執行以下腳本進行初始化工作，即可一鍵完成以上步驟。

2.2.1 創建用戶、項目、密鑰等

cd tbds_lambda_demo/bin  ./workflow_demo_clean_sth.sh user_project_key  ./workflow_demo_init_sth.sh user_project_key

2.2.2 創建策略

cd tbds_lambda_demo/bin  ./workflow_demo_clean_sth.sh ranger  ./workflow_demo_init_sth.sh ranger

2.3 查看結果

以admin用戶登陸Portal頁面，查看項目、用戶、密鑰、策略等是否已正確創建和配置。

2.3.1 查看項目

確認lambda_demo_project項目已創建。

2.3.2 查看用戶

確認lambda_demo_user用戶已創建，並關聯到lambda_demo_project項目。

2.3.3 查看密鑰

確認用戶lambda_demo_user用戶的hadoop模組密鑰已創建並開啟。

2.3.4 查看策略

確保hbase_4_lambda_demo_user和topic_4_lambda_demo_user策略已正確創建和配置。

3. 基於Oceanus的實時計算Demo

3.1 設計原理

理解實時業務邏輯與設計，本Demo使用Oceanus來完成實時數據的聚合與ETL業務：

• Aggregation業務邏輯：每10秒鐘將收到的數據進行累加，以「計算」時間為準，而不是數據時間；
• ETL業務邏輯：實時的將收到的每一條數據加上id所對應的name後同時寫入到kafka的topic和hbase表中。

3.2 前提條件

請確保以下前提條件得以滿足：

• 在portal主節點上執行demo的腳本；
• portal節點上安裝了以下客戶端：
  • hbase client
  • kafka client

3.3 生成demo

實時流業務Demo初始化步驟如下：

1.進入bin目錄中初始化數據：

 cd tbds_lambda_demo/bin   ./init_oceanus.sh

2.執行如下腳本模擬實時數據生成：

./realtime_date_generator.sh <send_interval> <total_send_messages>

其中：

• send_interval：生成的每一條數據發送的間隔，例如2s代表每2秒發送一條數據；
• total_send_messages：總共發送數據的條目數量，例如1000代表總共發送1000條數據。

例如：

./realtime_data_generator.sh 2s 1000

正常運行樣例如下：

通過該介面，可直觀的觀察到實時產生的數據。

3.執行如下腳本採集實時數據至kafka topic：

./realtime_data_collector.sh start

正常運行的截圖如下：

4.以lambda_demo_user用戶登錄portal，並進入到「實時計算」模組，查看其中實時計算任務，示例如下：

任務說明：

• 910009：用SQL實現的oceanus應用，業務邏輯為每10秒鐘做一次聚合運算，並將聚合運算結果存到MYSQL資料庫中；
• 910010：用畫布實現的與910009同樣功能的應用，因此910009與910010只能啟動其中一個；
• 910011：用畫布實現的實時數據ETL邏輯，並將結果寫入到hbase表和kafka的topic中。具體業務邏輯比較簡單，點開應用可獲取到詳細資訊。

注意：測試時請啟動910011和910009（或910010）。

3.4 查看結果

查看實時計算結果腳本名稱為check_realtime_calculate_result.sh，腳本用法如下：

cd tbds_lambda_demo/bin  ./check_realtime_calculate_result.sh kafka|hbase|mysql

3.4.1 MySQL結果

查看10秒聚合一次的結果：

cd tbds_lambda_demo/bin  ./check_realtime_calculate_result.sh mysql

正常運行截圖如下：

列印規則為2s中獲取一次最新的數據，輸出最新的10條。

3.4.2 Kafka結果

查看實時處理後寫入kafka topic的結果

cd tbds_lambda_demo/bin  ./check_realtime_calculate_result.sh kafka

正常運行截圖如下：

輸出規則為實時刷新。

3.4.3 HBase結果

查看實時處理後寫入hbase中的結果：

cd tbds_lambda_demo/bin  ./check_realtime_calculate_result.sh hbase

正常運行截圖如下：

列印規則為2s中獲取一次最新的數據，輸出最新的10條。

4 基於Hermes的多維分析Demo

4.1 設計原理

後台模擬實時生成客戶的數據至kafka，然後在TBDS的實時計算任務處理後再把結果落地到kafka的另外一個topic，hermes側再把新的kafka數據採集到hermes中，最終用戶在hermes中對數據進行查詢使用。

4.2 配置文件

配置conf/tools.conf中的兩個參數，使用默認配置亦可。其中：

• detail_topic：一定要和Oceanus Demo中的910011任務最終落地到kafka新topic名一致；
• hermes_table_name：kafka中的數據會導入此表中，表名自行定義（確保表名不要和已有的表重名，後面開始測試會先刪除此表）。

4.3 生成demo

執行以下腳本：

./hermes_demo_test.sh start_test

執行後，腳本會先將測試表刪除，再在hermes中註冊topic資訊，最後創建hemers測試表，結果輸出如下，確保register和create結果是成功的。

4.4 其他功能

hermes_demo_test.sh腳本中封裝了一些其他功能，可單獨執行調用hermes的介面進行調試，腳本中涉及到的kafka topic即為tools.conf中配置的detail_topic，hermes表也是tools.conf中配置的hermes_table_name。

使用方法：

hermes_demo_test.sh <parameter>  USAGE: ./hermes_demo_test.sh start_test|del|reg|list|select|create|desc|export|check_export

其中，參數說明如下：

• del：只做刪除表操作，刪除的表為tools.conf中配置的hermes_table_name；
• reg：註冊kafka topic到hermes，註冊的topic為detail_topic；
• create：在hermes中創建表，欄位和屬性都在腳本中已硬編碼，若有變化需要修改腳本；
• list：列出hermes表分區資訊；
• select：查詢當天分區的數據；
• desc：列出表的屬性資訊；
• export：導出表數據；
• check_export：查詢導出進度。

5 構建基於工作流的任務調度Demo

5.1 設計原理

5.1.1 架構圖

本Demo的設計囊括一個離線數據處理的全流程，包括模擬數據生成、數據導入、數據離線計算、數據分析查詢、數據導出等多個環節，其架構和數據流轉如下所示：

說明如下：

1. 真實的業務場景中，數據從客戶系統中產生。本Demo利用自動化腳本模擬數據的生成，先以csv數據格式的形式存放在本地文件系統中，是為模擬數據生成；
2. 通過自動化腳本，分別將數據從本地文件系統中導入到不同的源中，模擬真實業務場景下客戶各種的數據源，比如HDFS文件系統、FTP伺服器、MySQL資料庫等；
3. 通過工作流任務調度，將HDFS、FTP、MySQL三種數據源中的數據分別導入到Hive中，以Hive表形式存儲到TBDS平台中，是為數據導入；
4. 通過工作流任務調度，將上述導入的Hive表進行關聯計算，得出結果表並重新落地到Hive中，是為離線數據計算；
5. 通過工作流任務調度，對離線計算所得結果表進行分析查詢，並將結果出庫Hive重新導入到MySQL資料庫中，是為數據分析查詢和數據導出。

5.1.2 demo流程

本Demo已完全由自動化腳本實現，但工作流任務本身的運行需手動執行，全流程如下圖所示：

具體步驟如下，

腳本化執行部分：

1. 清理工作；
2. 模擬數據生成；
3. 數據上傳到HDFS；
4. 數據上傳到FTP；
5. 數據導入到MySQL；
6. 數據導入到Hive；
7. 創建工作流伺服器配置、工作流及上傳定義文件。

手動執行部分：

1. 提交工作流任務運行的申請；
2. 管理員（本Demo設定為admin用戶）審批；
3. 運行工作流任務，查看結果。

5.1.3 表結構與關係圖

表數據的示例，以及表與表之間的關係如下：

5.2 前提條件

運行本Demo前，需具備以下前提條件：

1. 確保TBDS集群中以下服務運行正常：
   • TBDS Portal
   • HDFS
   • FTP on HDFS
   • MetaDB（或CDB）
   • Hive
   • Lhotse
   • TBDS Authorizer
2. 第2章節中的Demo準備工作已完成；
3. 基於Oceanus的實時計算Demo中的init_oceanus.sh和realtime_data_generator.sh腳本已運行，產生了實時數據並已從本地文件系統上傳到以下HDFS目錄/user/lambda_demo_user/tbdsLambdaDemo/wfDemo/hdfsData/balance；
4. 運行腳本的操作請在Portal節點上執行，且要求Portal節點安裝了以下組件
   • HDFS client
   • FTP client
   • MySQL Client

5.3 配置文件

修改配置文件conf/wf_demo.conf中以下配置項

1. portalAdminPW：admin用戶的密碼，如123456；
2. portalMySQLPW：portal節點上MySQL DB的root用戶密碼，例如[email protected]。

其餘配置項按默認值即可，均不需要修改。

5.4 執行Demo腳本

執行以下步驟即可一鍵生成Demo

1.執行workflow_demo_init_all.sh腳本

cd tbds_lambda_demo/bin  chmod +x ./*.sh  ./workflow_demo_init_all.sh  

2.正常運行的輸出結果如下：

5.5 檢查工作

執行完工作流Demo初始化的腳本後，在運行工作流調度任務前，需要確保以下幾項檢查工作。

5.5.1 HDFS數據

1.檢查以下目錄是否已成功創建：

執行命令：

xportCmd=`/data/tools/getkey.py hdfs`  $exportCmd  hdfs dfs -ls /user/lambda_demo_user/tbdsLambdaDemo/wfDemo/

2.檢查以下數據文件是否已成功上傳：

執行命令：

hdfs dfs -ls /user/lambda_demo_user/tbdsLambdaDemo/wfDemo/hdfsData

5.5.2 FTP數據

1.檢查以下目錄是否已成功創建：

執行命令：

cd lambda_demo/bin  ftpHost=`./utils.sh ftpServerList | head -n 1`  ftpPort="2222"  ftp -n <<EOF  open $ftpHost $ftpPort  user lambda_demo_user lambda_demo_user  ls /user/lambda_demo_user/tbdsLambdaDemo/wfDemo/  bye  EOF

2.檢查以下數據文件是否已成功上傳到上述已創建的目錄 ：

執行命令：

cd lambda_demo/bin  ftpHost=`./utils.sh ftpServerList | head -n 1`  ftpPort="2222"  ftp -n <<EOF  open $ftpHost $ftpPort  user lambda_demo_user lambda_demo_user  ls /user/lambda_demo_user/tbdsLambdaDemo/wfDemo/ftpData  bye  EOF

5.5.3 MySQL數據

1.連接到MetaDB/CDB MySQL伺服器，執行以下命令：

cd lambda_demo/bin  source ../conf/wf_demo.conf  metadbMaster=`./utils.sh metadbMaster`  mysql -u$metaDbMasterUser -p$metaDbMasterPW -h$metadbMaster

2.檢查以下資料庫是否已成功創建：

成功連接到MetaDB/CDB MySQL後，執行命令：

show databases;

3.若資料庫已成功創建，檢查以下數據表是否已成功創建，且對應條目數符合預期：

4.成功連接到MetaDB/CDB MySQL後，執行命令：

use lambda_demo_db;  show tables;  select count(*) from city_id;  select count(*) from city_info;  select count(*) from country_info;  select count(*) from deposit_month;  select count(*) from marriage; 

5.5.4 Hive表

1.連接到Hive Server，執行以下命令：

cd lambda_demo/bin  source ../conf/wf_demo.conf  zkList=`./utils.sh zkList`  hiveServerConnectStr="jdbc:hive2://$zkList/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2"  beeline -u "$hiveServerConnectStr" -n$portalAdmin -p$portalAdminPW

2.檢查以下資料庫是否已成功創建：

成功連接到Hive Server後，執行命令：

show databases;

3.若資料庫已成功創建，檢查以下數據表是否已成功創建，且對應條目數符合預期：

成功連接到Hive Server後，執行命令：

use lambda_demo_db;  show tables;  select count(*) from age_info;  select count(*) from city_id;  select count(*) from city_info;  select count(*) from contact;  select count(*) from country_info;  select count(*) from deposit_date;  select count(*) from education;  select count(*) from gender_id;  select count(*) from gender_info;  select count(*) from marriage;  select count(*) from occupation;  select count(*) from person;  select count(*) from trans_deposit;

5.5.5 Portal頁面

1.用戶

用戶lambda_demo_user已創建，初始密碼為lambda_demo_user，驗證能夠以該用戶登陸Portal頁面。

2.項目

進入平台管理->項目管理，驗證項目lambda_demo_project已創建，

項目成員包含兩位，即admin和lambda_demo_user，角色均為operator/dev/projectManager。

3.密鑰

進入用戶管理->密鑰管理，搜索lambda_demo_user，驗證已為lambda_demo_user創建一個hadoop模組的密鑰並已啟用。

4.工作流伺服器配置

進入工作流 -> 伺服器配置，驗證已成功創建以下4個工作流伺服器配置

確保伺服器均為可用狀態，可點擊操作->編輯進行驗證。

5.工作流文件夾與定義文件

進入工作流 -> 工作流，驗證已成功創建workflow_lambda_demo_folder工作流文件夾和文件workflow_lambda_demo，其定義能夠點擊打開並展示如下。

5.6 運行Demo任務

1.先點擊保存工作流：

2.在工作流任務定義中，從上而下依次右鍵任務，點擊運行：

註：如果運行失敗，可以嘗試點擊編輯任務，然後保存任務，再點擊運行任務進行嘗試。

5.7 查看結果

5.7.1 Hive表

1.連接到Hive Server，執行以下命令：

cd lambda_demo/bin    source ../conf/wf_demo.conf    zkList=`./utils.sh zkList`    hiveServerConnectStr="jdbc:hive2://$zkList/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2"    beeline -u "$hiveServerConnectStr" -n"$portalAdmin" -p"$portalAdminPW" 2>/dev/null

2.檢查以下數據表的數據是否已成功從各數據源導入Hive和通過Hive SQL生成，且對應條目數符合預期：

各數據表：

表age_info

select count(*) from lambda_demo_db.age_info;  select * from lambda_demo_db.age_info limit 5;

表city_id

select count(*) from lambda_demo_db.city_id;  select * from lambda_demo_db.city_id limit 5;

表city_info

select count(*) from lambda_demo_db.city_info;  select * from lambda_demo_db.city_info limit 5;

表contact

select count(*) from lambda_demo_db.contact;  select * from lambda_demo_db.contact limit 5;

表country_info

select count(*) from lambda_demo_db.country_info;  select * from lambda_demo_db.country_info limit 5;

表deposit_date

select count(*) from lambda_demo_db.deposit_date;  select * from lambda_demo_db.deposit_date limit 5;

表education

select count(*) from lambda_demo_db.education;  select * from lambda_demo_db.education limit 5;

表gender_id

select count(*) from lambda_demo_db.gender_id;  select * from lambda_demo_db.gender_id limit 5;

表gender_info

select count(*) from lambda_demo_db.gender_info;  select * from lambda_demo_db.gender_info limit 5;

表marriage

select count(*) from lambda_demo_db.marriage;  select * from lambda_demo_db.marriage limit 5;

表occupation

select count(*) from lambda_demo_db.occupation;  select * from lambda_demo_db.occupation limit 5;

表person

select count(*) from lambda_demo_db.person;  select * from lambda_demo_db.person limit 5;

表trans_deposit

select count(*) from lambda_demo_db.trans_deposit;    select * from lambda_demo_db.trans_deposit limit 5;

5.7.2 MySQL表

1.連接到MetaDB/CDB MySQL伺服器，執行以下命令：

cd tbds_lambda_demo/bin  source ../conf/wf_demo.conf  metadbMaster=`./utils.sh metadbMaster`  mysql -u$metaDbMasterUser -p$metaDbMasterPW -h$metadbMaster

2.檢查以下數據表的數據是否已導入成功，且對應條目數符合預期：

表deposit_month

select count(*) from lambda_demo_db.deposit_month;  select * from lambda_demo_db.deposit_month limit 5;

附錄

工具包下載