大数据-240 离线数仓 - 广告业务 测试 ADS层数据加载 DataX数据导出到 MySQL

点一下关注吧！！！非常感谢！！持续更新！！！

Java篇开始了！

目前开始更新 MyBatis，一起深入浅出！

目前已经更新到了：

• Hadoop（已更完）
• HDFS（已更完）
• MapReduce（已更完）
• Hive（已更完）
• Flume（已更完）
• Sqoop（已更完）
• Zookeeper（已更完）
• HBase（已更完）
• Redis （已更完）
• Kafka（已更完）
• Spark（已更完）
• Flink（已更完）
• ClickHouse（已更完）
• Kudu（已更完）
• Druid（已更完）
• Kylin（已更完）
• Elasticsearch（已更完）
• DataX（已更完）
• Tez（已更完）
• 数据挖掘（已更完）
• Prometheus（已更完）
• Grafana（已更完）
• 离线数仓（正在更新…）

# 章节内容

上节我们完成了如下的内容：

• 广告业务 测试
• FlumeAgent 加载ODS、DWD层

导入数据

加载ADS层

ad_show_page

sh /opt/wzk/hive/ads_load_ad_show_page.sh 2020-07-21

执行结果如下图所示：

查看Hive中的数据：

hive 

use ads;select * from ads_ad_show_place limit 5;

运行结果如下图所示：

继续导入其他数据：

sh /opt/wzk/hive/ads_load_ad_show_page.sh 2020-07-22
sh /opt/wzk/hive/ads_load_ad_show_page.sh 2020-07-23
sh /opt/wzk/hive/ads_load_ad_show_page.sh 2020-07-24
...省略

最终的Hive的数据量如下所示：

select count(*) from ads_ad_show_place;

对应图片为：

ad_show_page_window

sh /opt/wzk/hive/ads_load_ad_show_page_window.sh 2020-07-21

执行结果如下所示：

查看Hive中的数据如下：

hive 

use ads;select * from ads_ad_show_place_window limit 5;

执行结果如下图所示：

继续加载其他的数据：

sh /opt/wzk/hive/ads_load_ad_show_page_window.sh 2020-07-22
sh /opt/wzk/hive/ads_load_ad_show_page_window.sh 2020-07-23
sh /opt/wzk/hive/ads_load_ad_show_page_window.sh 2020-07-24
...省略

Hive中的数据总数如下：

select count(*) from ads_ad_show_place_window;

运行结果如下所示：

导出数据

执行步骤

• 在MySQL创建对应的表
• 创建配置文件（JSON）
• 执行命令，使用JSON配置文件，测试
• 编写执行脚本（Shell）
• Shell脚本的测试

MySQL

droptableifexists dwads.ads_ad_show_place;createtable dwads.ads_ad_show_place(
  ad_action tinyint,hourvarchar(2),
  place varchar(20),
  product_id int,
  cnt int,
  dt varchar(10));

执行结果如下图所示：

DataX

配置文件

vim /opt/wzk/datax/ads_ad_show_place.json

写入的内容如下所示：

{"job":{"setting":{"speed":{"channel":1}},"content":[{"reader":{"name":"hdfsreader","parameter":{"path":"/user/hive/warehouse/ads.db/ads_ad_show_place/dt=$do_date/*","defaultFS":"hdfs://h121.wzk.icu:9000","column":[{"index":0,"type":"string"},{"index":1,"type":"string"},{"index":2,"type":"string"},{"index":3,"type":"string"},{"index":4,"type":"string"},{"type":"string","value":"$do_date"}],"fileType":"text","encoding":"UTF-8","fieldDelimiter":","}},"writer":{"name":"mysqlwriter","parameter":{"writeMode":"insert","username":"hive","password":"[email protected]","column":["ad_action","hour","place","product_id","cnt","dt"],"preSql":[
              "delete from ads_ad_show_place
              where dt='$do_date'"
            ],"connection":[{"jdbcUrl":"jdbc:mysql://h122.wzk.icu:3306/dwads?useUnicode=true&characterEncoding=utf-8","table":["ads_ad_show_place"]}]}}}]}}

写入内容如下所示：

DataX 简介

DataX 是由阿里巴巴开源的分布式离线数据同步工具，用于解决不同存储系统之间的数据传输问题。它被设计为一种高效、稳定、易扩展的工具，能够适应多种复杂的数据同步需求。

核心特点

支持多种数据源：

• 关系型数据库： MySQL, PostgreSQL, Oracle, SQL Server, DB2 等。
• NoSQL 数据库： MongoDB, HBase 等。
• 大数据存储系统： Hive, MaxCompute (ODPS), HDFS。
• 其他： FTP 文件、ElasticSearch 等。

高性能：

• 基于多线程的并行架构，能充分利用机器的多核性能。
• 支持分片传输，提高数据传输的吞吐量。

灵活性和易用性：

• 配置文件化：使用 JSON 格式的配置文件定义任务，简单直观。
• 支持任务调度，可与调度工具集成实现定时任务。
• 兼容多种数据格式和传输协议。

扩展性：

• 插件式架构，开发者可以通过编写 Reader 和 Writer 插件支持新的数据源。

开源与社区支持：

• 基于 Apache 2.0 开源协议，开发者可以自由使用和修改。
• 拥有活跃的社区和丰富的文档支持。

组成部分

Reader：

• 负责从数据源中读取数据。
• 示例：MySQLReader, HdfsReader。

Writer：

• 负责将数据写入目标存储。
• 示例：MySQLWriter, HdfsWriter。

Framework：

• DataX 的核心调度引擎，负责 Reader 和 Writer 的协调工作。
• 提供错误处理、数据校验、性能优化等功能。

Transform：

• 用于对传输的数据进行处理和转换（可选）。
• 例如数据格式的转换、字段的增删改等。

工作流程

初始化：

• 加载用户配置的 JSON 文件，解析 Reader 和 Writer 的配置。
• 准备任务上下文。

读取数据：

• Reader 读取源数据并以批量的方式输出。

转换数据：

• 可选步骤，Transform 模块对数据进行处理。

写入数据：

• Writer 接收 Reader 输出的数据并将其写入目标存储。

任务管理与监控：

• DataX 提供实时的任务运行日志和统计信息，包括速度、成功率、错误信息等。

执行导出

vim /opt/wzk/hive/ads_ad_show_place.sh

写入的内容如下所示：

#!/bin/bash
source /etc/profile
JSON=/opt/wzk/datax
if[-n "$1"];then
do_date=$1else
do_date=`date -d "-1 day" +%F`
fi
python $DATAX_HOME/bin/datax.py -p "-Ddo_date=$do_date" $JSON/ads_ad_show_place.json

写入结果如下图：

执行脚本可以得到结果：

sh /opt/wzk/hive/ads_ad_show_place.sh 2020-07-21

执行过程如下图所示：

查看结果

执行结束

查看数据库的结果如下所示：