sqoop导入

sqoop导入全指南：从关系型数据库到 Hadoop 生态

Sqoop 作为 Hadoop 与关系型数据库之间的数据同步工具，其核心功能之一是将关系型数据库（如 MySQL）的数据导入到 HDFS、Hive 等分布式存储系统中。本文将详细介绍 Sqoop 导入的多种场景（全量导入、条件导入、Hive 导入等），并解析核心参数的使用方法，帮助读者灵活应对不同的数据同步需求。

Sqoop 导入核心原理

Sqoop 导入的本质是将 SQL 查询转换为 MapReduce 作业，通过以下步骤完成数据同步：

Sqoop 解析用户输入的 SQL 命令或表信息，生成 MapReduce 作业；
Map 任务通过 JDBC 连接关系型数据库，并行读取数据；
数据被写入目标存储（HDFS 或 Hive），并按指定格式分割（如 \t 分隔）；
若导入 Hive，额外执行 HiveQL 将 HDFS 数据加载到 Hive 表中。

整个过程仅使用 Map 任务（无 Reduce 任务），避免数据 shuffle，提高效率。

Sqoop 导入实战场景

全量导入到 HDFS

将关系型数据库中的整张表数据完整导入到 HDFS 指定目录，适用于初始化数据或全量同步场景。

命令示例

sqoop import \  
  --connect jdbc:mysql://localhost:3306/company \  # 数据库连接 URL  
  --username root \                                # 数据库用户名  
  --password 123456 \                              # 数据库密码  
  --table staff \                                  # 要导入的表名  
  --target-dir /user/sqoop/company \               # HDFS 目标目录  
  --delete-target-dir \                            # 若目标目录存在则删除（避免数据重复）  
  --num-mappers 1 \                                # Map 任务数量（并行度）  
  --fields-terminated-by "\t"                      # 字段分隔符（默认逗号）  

关键参数解析

--table：指定源表名，Sqoop 会自动读取表结构并生成对应的数据文件；
--delete-target-dir：防止目标目录已存在导致的导入失败，适合定时全量覆盖场景；
--num-mappers：控制并行度，过多会增加数据库压力，建议根据表大小设置（1-10 为宜）。

基于 SQL 查询的条件导入

通过自定义 SQL 语句筛选数据后导入 HDFS，适用于只需要部分字段或过滤条件的数据同步。

命令示例

sqoop import \  
  --connect jdbc:mysql://localhost:3306/company \  
  --username root \  
  --password 123456 \  
  --query 'select name, sex, age from staff where sex = 0 and $CONDITIONS;' \  # 自定义 SQL  
  --target-dir /user/sqoop/company_filter \  
  --delete-target-dir \  
  --num-mappers 1 \  
  --fields-terminated-by "\t"  

注意事项

$CONDITIONS 必须包含：Sqoop 会用该占位符替换并行查询条件（如分块查询），缺失会报错；
引号处理：若 SQL 用双引号包裹，$CONDITIONS 需加转义符（\$CONDITIONS），避免被 Shell 解析为变量；
替代参数：--query 可替代 --table、--columns、--where，更灵活地控制查询逻辑。

指定列导入到 HDFS

仅导入表中的部分列，减少无效数据传输，适用于表字段较多但业务只需部分字段的场景。

命令示例

sqoop import \  
  --connect jdbc:mysql://localhost:3306/company \  
  --username root \  
  --password 123456 \  
  --table staff \  
  --columns "name, sex" \  # 指定要导入的列（逗号分隔）  
  --target-dir /user/sqoop/company_columns \  
  --delete-target-dir \  
  --num-mappers 1 \  
  --fields-terminated-by "\t"  

说明

--columns 需与表中实际列名一致，区分大小写；
若需同时过滤行，可结合 --where 参数（如 --where "age > 18"）。

导入到 Hive

直接将数据导入 Hive 表，适用于数据仓库 ETL 流程，避免手动加载 HDFS 数据到 Hive。

命令示例

sqoop import \  
  --connect jdbc:mysql://localhost:3306/company \  
  --username root \  
  --password 123456 \  
  --table staff \  
  --num-mappers 1 \  
  --hive-import \  # 启用 Hive 导入模式  
  --fields-terminated-by "\t" \  # 需与 Hive 表分隔符一致  
  --hive-overwrite \  # 覆盖 Hive 表中现有数据  
  --hive-table staff_hive  # 目标 Hive 表名（需提前创建或允许自动创建）  

导入流程

第一步：数据先导入到 HDFS 临时目录（默认 /user/<username>/<table>）；
第二步：通过 HiveQL（LOAD DATA INPATH）将 HDFS 数据加载到 Hive 表。

依赖与配置

需将 Hive 依赖包（如 hive-common.jar、hive-exec.jar）复制到 $SQOOP_HOME/lib/；
确保 Hive metastore 服务正常运行，且 Sqoop 能访问 Hive 配置（hive-site.xml 建议放在 $SQOOP_HOME/conf/）。

核心参数详解

参数	作用	适用场景
`--append`	数据追加到现有文件，而非覆盖	增量数据同步（避免删除历史数据）
`--where`	过滤行的条件（如 `--where "age > 18"`）	简单行过滤，无需复杂 SQL
`--fields-terminated-by <char>`	设定字段分隔符（如 `\t`、`,`）	与目标存储格式匹配（如 Hive 表分隔符）
`--lines-terminated-by <char>`	设定行分隔符（默认 `\n`）	特殊格式数据（如 Windows 换行符 `\r\n`）
`--hive-overwrite`	覆盖 Hive 表中现有数据	全量同步，确保数据最新
`--hive-table`	指定目标 Hive 表名	导入 Hive 时必选，需与表结构匹配
`--num-mappers`	并行 Map 任务数量	控制导入速度（数量越多越快，但消耗数据库资源越多）
`--delete-target-dir`	导入前删除目标目录	全量同步，避免数据重复

常见问题与优化

1. 导入速度慢

优化并行度：适当增加 --num-mappers（如 4-8），但需确保数据库能承受并发连接；
启用批量读取：添加 --batch 参数，减少 JDBC 交互次数；
使用直接模式：MySQL 可添加 --direct 参数，启用 mysqldump 加速导出。

2. 数据格式不匹配

分隔符冲突：若字段包含分隔符（如逗号），需使用 --escaped-by 转义（如 --escaped-by "\\"）；
Hive 表结构不匹配：确保 Hive 表列数、类型与源表一致，或通过 --map-column-hive 手动映射（如 --map-column-hive age=INT）。

3. 依赖缺失错误

Hive 导入报错：缺少 Hive 依赖时，复制 $HIVE_HOME/lib/ 下的 hive-common-*.jar、hive-exec-*.jar 到 $SQOOP_HOME/lib/；
ClassNotFoundException：检查 JDBC 驱动是否正确放置在 $SQOOP_HOME/lib/，版本是否与数据库兼容。

4. 增量导入策略

对于频繁更新的表，全量导入效率低，可采用增量导入：

基于递增 ID：

sqoop import \  
  --connect jdbc:mysql://localhost:3306/company \  
  --table staff \  
  --incremental append \  
  --check-column id \  # 递增列（如自增 ID）  
  --last-value 1000  # 上次导入的最大 ID  

基于时间戳：

sqoop import \  
  --incremental lastmodified \  
  --check-column update_time \  # 时间戳列  
  --last-value "2023-01-01 00:00:00"  