系列文章目录

线上问诊：业务数据采集

---

---

前言

暑假躺了两个月，也没咋写博客，准备在开学前再做个项目找找感觉，由于之前做过广告数仓的案例，这次的博客会相对简略一些，数仓包括离线和实时两个部分，离线用来加深记忆，实时用来学习新技术。

---

一、环境准备

由于很多内容之前博客都完成过。这里就不过多赘述了。

1.Hadoop

Hadoop学习专栏前四章内容为Hadoop集群安装。

2.Zookeeper

Zookeeper安装

3.Kafka

Kafka安装

4.Flume

Flume安装
安装好之后为了后边实验方便，修改两个参数



Flume安装后需要分发到所有节点

xsync /opt/module/flume/

5.Mysql

广告数仓：采集通道创建

6.Maxwell

这里我们选用1.29.2版本，因为在1.30.0开始，放弃了对java8的支持。

1.上传并解压


2.创建Maxwell所需数据库和用户

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

# 指定数据按照主键分组进入Kafka不同分区，避免数据倾斜
producer_partition_by=primary_key

4.启停脚本
vim ~/bin/mxw.sh

#!/bin/bash

MAXWELL_HOME=/opt/module/maxwell

status_maxwell(){
    result=`ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | wc -l`
    return $result
}


start_maxwell(){
    status_maxwell
    if [[ $? -lt 1 ]]; then
        echo "启动Maxwell"
        $MAXWELL_HOME/bin/maxwell --config $MAXWELL_HOME/config.properties --daemon
    else
        echo "Maxwell正在运行"
    fi
}


stop_maxwell(){
    status_maxwell
    if [[ $? -gt 0 ]]; then
        echo "停止Maxwell"
        ps -ef | grep com.zendesk.maxwell.Maxwell | grep -v grep | awk '{print $2}' | xargs kill -9
    else
        echo "Maxwell未在运行"
    fi
}


case $1 in
    start )
        start_maxwell
    ;;
    stop )
        stop_maxwell
    ;;
    restart )
       stop_maxwell
       start_maxwell
    ;;
esac

5.添加权限
chmod +x ~/bin/mxw.sh

二、业务数据采集

1.数据模拟

上传模拟数据

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

修改配置文件
vim application.yaml

现在我们为其创建数据库。

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

执行jar包
java -jar mock-medical-1.1.jar

在数据库检查是否获取到数据。

封装成脚本
vim ~/bin/medical_mock.sh

#!/bin/bash

for ((i=0; i < $1; i++))
do
    echo "正在执行第 $[ $i + 1 ] 次数据模拟"
    ssh hadoop102 "cd /opt/module/mock-medical/; java -jar mock-medical-1.1.jar"
done

2.采集通道

现在我们已经将初始数据输入到Mysql，接下来使用Maxwell将数据传输到Kafka。
先启动Zookeeper,Kafka和Maxwell

Maxwell会实时监控MYSQL数据，然后将其传送到Kafka集群，所以我们现打开一个Kafka消费集群。

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

现在我们再次生产数据输入到MYSQL，Maxwell就会将产生的数据传送到Kafka集群，我们实现打开的消费者就会收到数据。

出现数据，说明通道没问题。到此数据采集通道完成。

---

总结

这里抓一次快照，后边实时数仓还要基于这里再次开发。