系列文章目錄

線上問診：業(yè)務(wù)數(shù)據(jù)采集

---

---

前言

暑假躺了兩個月，也沒咋寫博客，準(zhǔn)備在開學(xué)前再做個項目找找感覺，由于之前做過廣告數(shù)倉的案例，這次的博客會相對簡略一些，數(shù)倉包括離線和實時兩個部分，離線用來加深記憶，實時用來學(xué)習(xí)新技術(shù)。

---

一、環(huán)境準(zhǔn)備

由于很多內(nèi)容之前博客都完成過。這里就不過多贅述了。

1.Hadoop

Hadoop學(xué)習(xí)專欄前四章內(nèi)容為Hadoop集群安裝。

2.Zookeeper

Zookeeper安裝

3.Kafka

Kafka安裝

4.Flume

Flume安裝
安裝好之后為了后邊實驗方便，修改兩個參數(shù)



Flume安裝后需要分發(fā)到所有節(jié)點

xsync /opt/module/flume/

5.Mysql

廣告數(shù)倉：采集通道創(chuàng)建

6.Maxwell

這里我們選用1.29.2版本，因為在1.30.0開始，放棄了對java8的支持。

1.上傳并解壓


2.創(chuàng)建Maxwell所需數(shù)據(jù)庫和用戶

CREATE DATABASE maxwell;
CREATE USER 'maxwell'@'%' IDENTIFIED BY 'maxwell';
GRANT ALL ON maxwell.* TO 'maxwell'@'%';
GRANT SELECT, REPLICATION CLIENT, REPLICATION SLAVE ON *.* TO 'maxwell'@'%';

3.修改配置文件

cp config.properties.example config.properties
vim config.properties

producer=kafka
kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092
kafka_topic=topic_db# mysql login info
host=hadoop102
user=maxwell
password=maxwell
jdbc_options=useSSL=false&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true# 指定數(shù)據(jù)按照主鍵分組進(jìn)入Kafka不同分區(qū)，避免數(shù)據(jù)傾斜
producer_partition_by=primary_key

4.啟停腳本
vim ~/bin/mxw.sh

#!/bin/bashMAXWELL_HOME=/opt/module/maxwellstatus_maxwell(){result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`return $result
}start_maxwell(){status_maxwellif [[ $? -lt 1 ]]; thenecho "啟動Maxwell"$MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemonelseecho "Maxwell正在運行"fi
}stop_maxwell(){status_maxwellif [[ $? -gt 0 ]]; thenecho "停止Maxwell"ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9elseecho "Maxwell未在運行"fi
}case $1 instart )start_maxwell;;stop )stop_maxwell;;restart )stop_maxwellstart_maxwell;;
esac

5.添加權(quán)限
chmod +x ~/bin/mxw.sh

二、業(yè)務(wù)數(shù)據(jù)采集

1.數(shù)據(jù)模擬

上傳模擬數(shù)據(jù)

mkdir /opt/module/mock-medical
cd /opt/module/mock-medical

修改配置文件
vim application.yaml

現(xiàn)在我們?yōu)槠鋭?chuàng)建數(shù)據(jù)庫。

mysql -uroot -p000000 -e"drop database if exists medical;create database medical charset utf8mb4 collate utf8mb4_general_ci;"

執(zhí)行jar包
java -jar mock-medical-1.1.jar

在數(shù)據(jù)庫檢查是否獲取到數(shù)據(jù)。

封裝成腳本
vim ~/bin/medical_mock.sh

#!/bin/bashfor ((i=0; i < $1; i++))
doecho "正在執(zhí)行第 $[ $i + 1 ] 次數(shù)據(jù)模擬"ssh hadoop102 "cd /opt/module/mock-medical/; java -jar mock-medical-1.1.jar"
done

2.采集通道

現(xiàn)在我們已經(jīng)將初始數(shù)據(jù)輸入到Mysql，接下來使用Maxwell將數(shù)據(jù)傳輸?shù)終afka。
先啟動Zookeeper,Kafka和Maxwell

Maxwell會實時監(jiān)控MYSQL數(shù)據(jù)，然后將其傳送到Kafka集群，所以我們現(xiàn)打開一個Kafka消費集群。

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic topic_db

現(xiàn)在我們再次生產(chǎn)數(shù)據(jù)輸入到MYSQL，Maxwell就會將產(chǎn)生的數(shù)據(jù)傳送到Kafka集群，我們實現(xiàn)打開的消費者就會收到數(shù)據(jù)。

出現(xiàn)數(shù)據(jù)，說明通道沒問題。到此數(shù)據(jù)采集通道完成。

---

總結(jié)

這里抓一次快照，后邊實時數(shù)倉還要基于這里再次開發(fā)。