棣廷头像

初识数据仓库——数仓分层到DataX数据采集

一、引言

在之前的博客中，我们学习了Hadoop的底层组件（HDFS、MapReduce、YARN）以及Hive数据仓库工具。我们知道了如何用Hive建表、加载数据、写SQL进行统计分析。

但现实世界的数据并不是一开始就躺在Hive里的。数据的源头，通常是各个业务系统（如MySQL、Oracle、业务后台等）。业务数据是分散的、局部的、面向事务的，而数据分析需要的是全局的、集成的、面向主题的数据。

如何将业务系统中的数据源源不断地抽取到数据仓库中？这就引出了数仓建设的核心流程，以及数据采集工具（如DataX）。

本篇博客将结合企业级数仓建设方法论，从数仓分层架构讲起，并实战演示如何使用DataX完成从MySQL到HDFS/Hive的全量/增量数据采集。

二、数据仓库分层架构

在构建数仓之前，首先需要明确分层架构。数据仓库通常分为以下几层，目的是为了解耦、复用和提升性能：

分层英文全称核心作用
ODSOperational Data Store贴源层。直接映射业务数据库，不作过多清洗。
DWDData Warehouse Detail明细层。进行数据清洗、脱敏、规范化，构建事实表。
DIMDimension维度层。存放各类公共维度数据（如时间、地点、用户）。
DWSData Warehouse Summary汇总层。轻度汇总，按主题聚合，形成宽表。
ADSApplication Data Service应用层。面向具体报表和业务的直接输出结果。

建库SQL：

CREATE DATABASE ods LOCATION '/daas/ods';
CREATE DATABASE dwd LOCATION '/daas/dwd';
CREATE DATABASE dim LOCATION '/daas/dim';
CREATE DATABASE dws LOCATION '/daas/dws';
CREATE DATABASE ads LOCATION '/daas/ads';

三、模拟业务系统数据源

为了完整模拟实战流程，我们在MySQL中创建业务表（模拟刷卡系统）。

1. 创建刷卡源表 szt_data：

CREATE TABLE szt.szt_data(
    car_no VARCHAR(255) COMMENT '车牌号',
    card_no VARCHAR(255) COMMENT '卡号',
    close_date VARCHAR(255) COMMENT '结算日期',
    company_name VARCHAR(255) COMMENT '公司名称',
    conn_mark INT COMMENT '联程标记',
    deal_date VARCHAR(255) COMMENT '交易日期时间',
    deal_money DECIMAL(10,2) COMMENT '交易金额',
    deal_type VARCHAR(255) COMMENT '交易类型',
    deal_value DECIMAL(10,2) COMMENT '交易值',
    equ_no VARCHAR(255) COMMENT '设备编码',
    station VARCHAR(255) COMMENT '线路站点'
);

2. 导入测试数据：

-- 开启本地文件加载
SET GLOBAL local_infile=1;
LOAD DATA LOCAL INFILE '/root/szt_data.txt' INTO TABLE szt.szt_data FIELDS TERMINATED BY ',';

同样，可以创建用户信息表 szt_user 并导入数据。

四、数据采集神器——DataX

4.1 DataX简介

DataX 是阿里巴巴开源的一款异构数据源离线同步工具。它实现了包括 MySQL、Oracle、HDFS、Hive、HBase 等在内的各种异构数据源之间稳定高效的数据同步功能。

核心架构：DataX采用框架+插件模式，Reader（读插件）和Writer（写插件）由框架调度，通过Channel通道进行高速数据传输。

4.2 下载与安装

# 解压并配置环境变量
tar -xvf datax.tar.gz
vim /etc/profile
export DATAX_HOME=/usr/local/soft/datax
export PATH=$PATH:$DATAX_HOME/bin
source /etc/profile

五、实战：DataX从MySQL采集到HDFS

5.1 从MySQL读取数据（测试）

首先，我们可以配置DataX从MySQL读取数据并打印到控制台，用于测试连通性。

配置 szt_data.json：

{
    "job": {
        "setting": {
            "speed": {
                 "channel": 3   
            }
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "column": [
                            "car_no", "card_no", "close_date", "company_name", 
                            "conn_mark", "deal_date", "deal_money", "deal_type", 
                            "deal_value", "equ_no", "station"
                        ],
                        "splitPk": "equ_no",
                        "connection": [
                            {
                                "table": ["szt_data"],
                                "jdbcUrl": ["jdbc:mysql://master:3306/szt"]
                            }
                        ]
                    }
                },
                "writer": {
                    "name": "streamwriter",
                    "parameter": {
                        "print": true
                    }
                }
            }
        ]
    }
}

执行：datax.py szt_data.json

5.2 增量同步到HDFS（写入ODS层）

在数仓日常调度中，通常是按天进行增量同步。我们需要在 reader 中增加 where 条件，在 writer 中指定HDFS路径。

DataX 核心配置：

{
    "job": {
        "setting": {
            "speed": {"channel": 3},
            "errorLimit": {"record": 0, "percentage": 0.02}
        },
        "content": [
            {
                "reader": {
                    "name": "mysqlreader",
                    "parameter": {
                        "username": "root",
                        "password": "123456",
                        "column": ["car_no", "card_no", "deal_date", "deal_money", "equ_no"],
                        "where": "date_format(deal_date,'%Y%m%d')='${dt}'",
                        "splitPk": "equ_no",
                        "connection": [
                            {
                                "table": ["szt_data"],
                                "jdbcUrl": ["jdbc:mysql://master:3306/szt"]
                            }
                        ]
                    }
                },
                "writer": {
                    "name": "hdfswriter",
                    "parameter": {
                        "defaultFS": "hdfs://master:9000",
                        "fileType": "orc",
                        "path": "/daas/ods/ods_szt_data/dt=${dt}",
                        "fileName": "part",
                        "writeMode": "append",
                        "fieldDelimiter": "\u0001"
                    }
                }
            }
        ]
    }
}

关键参数解析：

  • ${dt}：DataX 动态传参占位符。

  • fileType: orc：目标文件格式为 ORC，压缩率高。

  • fieldDelimiter: \u0001：使用特殊不可见字符作为分隔符，防止与业务数据中的逗号冲突。

  • channel: 3：并发数为3，加快传输速度。

5.3 在Hive ODS层创建表

在执行DataX之前，需要在Hive中建好对应的外部表，并指定相同的HDFS路径。

CREATE EXTERNAL TABLE ods.ods_szt_data(
    car_no STRING COMMENT '车牌号',
    card_no STRING COMMENT '卡号',
    close_date STRING COMMENT '结算日期',
    company_name STRING COMMENT '公司名称',
    conn_mark INT COMMENT '联程标记',
    deal_date STRING COMMENT '交易日期时间',
    deal_money DECIMAL(10,2) COMMENT '交易金额',
    deal_type STRING COMMENT '交易类型',
    deal_value DECIMAL(10,2) COMMENT '交易值',
    equ_no STRING COMMENT '设备编码',
    station STRING COMMENT '线路站点'
)
PARTITIONED BY (dt STRING COMMENT '按天分区')
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\u0001'
STORED AS ORC
LOCATION '/daas/ods/ods_szt_data';

六、脚本优化：自动化采集流程

在真实的企业环境中，每天的数据采集不可能靠手工执行一条条命令。我们需要将过程Shell脚本化，方便后续接入 Azkaban 或 DolphinScheduler 调度。

6.1 编写Shell调度脚本 ods_szt_data.sh

#!/bin/bash

# 接收外部传入的参数（日期，如 20180831）
dt=$1

if [ -z "$dt" ]; then
    echo "请传入日期参数：如 bash ods_szt_data.sh 20180831"
    exit 1
fi

echo "开始采集 ${dt} 的数据..."

# 1. 删除HDFS上该分区已有数据（保证幂等性，防止重复执行导致数据翻倍）
hdfs dfs -rmr -skipTrash /daas/ods/ods_szt_data/dt=${dt}

# 2. 创建HDFS分区目录
hdfs dfs -mkdir -p /daas/ods/ods_szt_data/dt=${dt}

# 3. 执行DataX采集脚本，传入日期参数
python $DATAX_HOME/bin/datax.py /root/szt_data.json -p"-Ddt=${dt}"

# 4. 检查DataX是否执行成功
if [ $? -eq 0 ]; then
    echo "DataX采集完成，开始同步Hive元数据..."
    # 5. 在Hive中增加对应分区元数据
    hive -e "ALTER TABLE ods.ods_szt_data ADD IF NOT EXISTS PARTITION(dt='${dt}');"
    echo "采集任务全流程成功！"
else
    echo "DataX执行失败，请检查日志！"
    exit 1
fi

6.2 执行优化脚本

# 赋予执行权限
chmod +x ods_szt_data.sh

# 执行采集任务（采集8月31日的数据）
bash ods_szt_data.sh 20180831

对于用户信息表 szt_user，只需将 ods_szt_data.sh 中的表名、路径和 JSON 脚本替换，即可用同样的流程完成采集。

七、总结

本篇博客完整走了一遍从业务数据库到数据仓库ODS层的数据接入流程。

步骤核心操作关键技术
1. 数仓规划建立ODS、DWD、DIM、DWS、ADS库Hive DDL
2. 业务模拟在MySQL建表并导入数据MySQL
3. 抽取工具下载并配置DataXJSON配置
4. 数据同步MySQL → HDFS (ORC)DataX Reader/Writer
5. 映射建表ODS外部表映射HDFS路径Hive External Table
6. 自动化Shell脚本封装建目录、DataX、加分区Bash Scripting

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/2301_77064095/article/details/169366123

文章来源转载

评论

评论列表

微信小程序
QQ小程序

关于作者

点赞数：0
关注数：0
粉丝：0
文章：0
关注标签：0
加入于：--