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

🚀 AI 篇持续更新中！（长期更新）

目前 2025 年 06 月 16 日更新到：AI 炼丹日志-29 - 字节跳动 DeerFlow 深度研究框斜体样式架 私有部署 测试上手 架构研究，持续打造实用 AI 工具指南！📐🤖

💻 Java 篇正式开启！（300 篇）

目前 2025 年 06 月 26 日更新到：Java-55 深入浅出 分布式服务 分布式一致性 强一致、弱一致、单调读一致、最终一致 MyBatis 已完结，Spring 已完结，Nginx 已完结，Tomcat 已完结，分布式服务正在更新！深入浅出助你打牢基础！

📊 大数据板块已完成多项干货更新（300 篇）：

包括 Hadoop、Hive、Kafka、Flink、ClickHouse、Elasticsearch 等二十余项核心组件，覆盖离线+实时数仓全栈！目前 2025 年 06 月 13 日更新到：大数据-278 Spark MLib - 基础介绍 机器学习算法 梯度提升树 GBDT 案例 详解

章节内容

上节我们完成了：

• Sqoop 介绍

• Sqoop 下载配置环境等

• Sqoop 环境依赖：Hadoop、Hive、JDBC 等环境补全

背景介绍

这里是三台公网云服务器，每台 2C4G，搭建一个 Hadoop 的学习环境，供我学习。

• 2C4G 编号 h121

• 2C4G 编号 h122

• 2C2G 编号 h123

注意事项

Apache Sqoop 于 2021 年 6 月迁入 Attic。Apache Sqoop 的使命是创建和维护与 Apache Hadoop 和结构化数据存储相关的大规模数据传输软件。虽然项目已经可能过时，但是如果你的时间充裕的话，可以适当地学习一下，可能会对你有一定的帮助的！！！

Sqoop 是离线数据导入的利器，尽管项目在 Apache 官方已归档（进入 attic），但在老旧 Hadoop 架构中仍被大量使用。新项目中若需同步数据，建议评估替代方案如：

• Apache NiFi（支持拖拽式数据流）

• DataX（阿里开源，支持丰富数据源）

• Flink CDC（流式同步）

基本介绍

Apache Sqoop 是一款专为在 关系型数据库（如 MySQL）与 Hadoop（如 HDFS、Hive、HBase） 之间高效传输数据而设计的工具。

• 主要用于**批量导入（Import）和批量导出（Export）**数据。

• 架构上依赖于 MapReduce，每个导入任务默认使用多个 Map 任务并发执行，提高传输效率。

核心用途

• 数据导入：将关系数据库的数据导入 HDFS、Hive、HBase 等

• 数据导出：将 Hadoop 系统中的数据导出回 RDBMS

• 批量传输：支持大批量并行传输，提高效率

• 数据仓库同步：与 Hive 集成，方便大数据分析

核心流程图

MySQL表 --> Sqoop调度器（生成MapReduce Job）--> 多个Mapper任务并发抓取数据 --> 写入HDFS

ETL

ETL 全称：Extract（抽取）、Transform（转换）、Load（加载），是指从多个数据源提取数据，进行清洗、格式转换、聚合等操作，最后加载到目标系统（如数据仓库、数据湖或分析平台）中的过程。

为什么需要 ETL？

• 整合异构系统数据（如 MySQL、Excel、日志、API 数据等）

• 保障数据质量（清洗异常值、统一格式）

• 支撑数据分析、报表、机器学习模型训练等下游需求

ETL 三个阶段

Extract 抽取

从各种来源系统中提取原始数据：

• 数据源类型：关系型数据库（MySQL、Oracle）、NoSQL（MongoDB）、CSV 文件、REST API、日志文件等

• 技术示例：JDBC、API 拉取、日志监听、消息队列（Kafka）

⚠️ 重点在于高效、准确、不丢数据

Transform 转换

对提取的数据进行处理，确保数据可用、准确、一致：

• 清洗：空值填充、重复数据处理、异常值识别

• 格式转换：日期格式统一、编码统一（如 UTF-8）

• 衍生字段：如从订单数据中计算用户生命周期价值（LTV）

• 业务规则应用：如收入必须为正数、时间不能晚于当前时间

• 数据汇总：如按照天/周/月聚合销售数据

✅ 转换阶段是提升数据“质量”和“可用性”的关键

Load 加载

将转换后的数据写入目标系统（如数据仓库或数据湖）：

• 批量导入：如将每日销售数据加载到 Hive、ClickHouse、Snowflake

• 增量加载：仅加载新增或变更部分（节省资源）

• 覆盖式加载：全量覆盖旧数据

⚠️ 需关注目标系统的写入能力、负载压力和一致性保障

基本导入命令

sqoop import \  --connect jdbc:mysql://localhost:3306/dbname \  --username root \  --password password \  --table user_table \  --target-dir /user/hdfs/user_table \  --num-mappers 4 \  --as-textfile \  --fields-terminated-by '\t'

参数说明：

• --connect：JDBC 连接字符串

• --username / --password：数据库账户

• --table：要导入的 MySQL 表名

• --target-dir：在 HDFS 上的输出目录

• --num-mappers：Mapper 数量，默认 4，可控制并发度

• --as-textfile：导入为纯文本文件（也可以选 --as-parquetfile、--as-avrodatafile）

• --fields-terminated-by：字段分隔符，默认,，可以用 \t 表示 TSV

导入模式

基于表的导入

--table 表名

自定义查询导入

sqoop import \  --connect jdbc:mysql://localhost:3306/dbname \  --username root \  --password password \  --query "SELECT id, name FROM user_table WHERE \$CONDITIONS" \  --split-by id \  --target-dir /user/hdfs/filtered_user \  --num-mappers 4 \  --as-textfile

注意：必须加 $CONDITIONS 作为占位符，Sqoop 会自动注入切分条件。

优化建议

• 并发导入，增加 --num-mappers（前提是数据分布均衡）

• 分片字段，使用 --split-by 明确指定用于 Map 任务切分的字段（必须是整型且唯一性较好）

• 网络 IO 瓶颈，Sqoop 进程建议部署在靠近数据库的节点上（或采用压缩）

• 输出格式，使用 --as-parquetfile 可以配合 Hive 更高效读取

• 数据压缩，加 --compression-codec 使用 Gzip 或 Snappy 压缩结果

增量导入

sqoop import \  --connect jdbc:mysql://localhost:3306/dbname \  --username root \  --password password \  --table user_table \  --target-dir /user/hdfs/user_table_increment \  --incremental append \  --check-column id \  --last-value 100 \  --num-mappers 4

• --incremental：支持 append 或 lastmodified

• --check-column：用于比较增量的字段

• --last-value：上次已导入的最大值（需自行记录）

生成数据

创建新库

连接我们的 MySQL 数据库，执行如下的 SQL，生成一些数据出来：

-- 用于在 Mysql 中生成测试数据CREATE DATABASE sqoop;use sqoop;CREATE TABLE sqoop.goodtbl(  gname varchar(50),  serialNumber int,  price int,  stock_number int,  create_time date);

移除函数

防止遇到干扰，先移除之前可能存在的函数：

DROP FUNCTION IF EXISTS `rand_string`;DROP PROCEDURE IF EXISTS `batchInsertTestData`;

编写脚本

-- 替换语句默认的执行符号，将；替换成 //DELIMITER //CREATE FUNCTION `rand_string` (n INT) RETURNS VARCHAR(255)CHARSET 'utf8'BEGIN  DECLARE char_str varchar(200) DEFAULT  '0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ';  DECLARE return_str varchar(255) DEFAULT '';  DECLARE i INT DEFAULT 0;  WHILE i < n DO    SET return_str = concat(return_str,    substring(char_str, FLOOR(1 + RAND()*36), 1));    SET i = i+1;  END WHILE;  RETURN return_str;END//CREATE PROCEDURE `batchInsertTestData` (m INT, n INT)BEGINDECLARE i INT DEFAULT 0;  WHILE i < n DO    insert into goodtbl (gname, serialNumber, price,    stock_number, create_time)    values (rand_string(6), i+m, ROUND(RAND()*100),    FLOOR(RAND()*100), now());    SET i = i+1;  END WHILE;END//delimiter ;call batchInsertTestData(1, 100);

执行脚本

运行脚本后：

查看数据

使用 SQL 查看，或者可视化查看，如下图：

依赖服务

请确保以下的服务都是正常运行的：

• HDFS

• YARN

• MySQL

MySQL->HDFS 全量导入

运行脚本

sqoop import \--connect jdbc:mysql://h122.wzk.icu:3306/sqoop \--username hive \--password hive@wzk.icu \--table goodtbl \--target-dir /root \--delete-target-dir \--num-mappers 1 \--fields-terminated-by "\t"

参数解释

上述参数的解释：

• target-dir 将数据导入 HDFS 的路径

• delete-target-dir 如果目标文件夹在 HDFS 上已经存在，那么再次运行就会报错。

• num-mappers 启动多少个 MapTask，默认启动 4 个

• fields-terminated-by HDFS 中数据的分隔符

分配任务

观察任务的分配过程

等待执行

可以观察到 Map -> Reduce 的过程

需要等待好一段时间，执行完成后效果如下图：

查看结果

到 HDFS 集群中查看数据，发现数据已经导入了：