四、Sqoop 导入表数据子集

作者：IvanCodes
日期：2025年6月4日
专栏：Sqoop教程

当不需要将关系型数据库中的整个表一次性导入，而是只需要表中的一部分数据时，Sqoop 提供了多种方式来实现数据子集的导入。这通常通过过滤条件或选择特定列来完成。

一、导入子集的核心方法

主要通过以下两种参数组合实现：

1. 使用 --table 配合 --where 子句：

   • 这是最直接和常用的方式，用于从单个表中根据行级别的条件筛选数据。
   • Sqoop 会将 --where 条件附加到从源表 SELECT 数据的SQL语句中。

2. 使用 --table 配合 --columns 参数：

   • 用于从单个表中选择特定的列进行导入。

3. 使用 --query 参数：

   • 提供最大的灵活性，允许你编写任意复杂的SQL查询语句，可以包含 WHERE 子句、列选择、多表连接、函数等，从而精确定义要导入的数据子集。

二、关键参数详解 (针对子集导入)

• --table <db-table-name>: (方法1和2使用) 指定要从中导入数据的源表名。

• --where "<sql-where-condition>": (配合 --table 使用)

  • 定义一个SQL的WHERE子句 (不包含 WHERE 关键字本身) 来过滤行。
  • 示例: --where "status = 'active' AND age > 30"
  • 注意: 如果条件中包含字符串，确保正确引用 (通常是单引号)。如果条件本身包含引号，可能需要转义。

• --columns "<col1>,<col2>,<col_n>": (配合 --table 使用)

  • 指定要导入的列名列表，列名之间用逗号分隔。
  • 示例: --columns "id,name,email"
  • 如果不指定此参数，Sqoop 默认导入表中的所有列。

• --query "<custom-sql-select-query>": (方法3使用)

  • 直接提供一个完整的SELECT查询语句。
  • 示例: --query "SELECT o.order_id, c.customer_name, o.order_total FROM orders o JOIN customers c ON o.customer_id = c.customer_id WHERE o.order_date >= '2023-01-01' AND \$CONDITIONS"
  • 重要提示:
    • 使用 --query 时，必须指定 --target-dir (即使是导入到Hive，也建议指定一个临时HDFS目录)。
    • 如果使用多个Mapper (-m > 1) 进行并行导入，查询语句中必须包含 \$CONDITIONS 占位符，并且需要配合 --split-by (以及可能的 --boundary-query) 来有效分割查询结果集。如果 -m 1，则 \$CONDITIONS 不是必需的。

• 其他参数如 --connect, --username, --password, --target-dir, --hive-import, -m, --split-by, 文件格式和压缩参数等，与全量导入时的用法基本相同。

三、导入数据子集示例 (MySQL)

场景： 假设MySQL数据库 mydb 中有表 employees (id INT PK, name STRING, department VARCHAR(50), salary DECIMAL(10,2), hire_date DATE)。

1. 使用 --table 和 --where 导入满足条件的行

(A) 导入 ‘Sales’ 部门且薪水大于60000的员工到HDFS

sqoop import \
--connect jdbc:mysql://localhost:3306/mydb \
--username dbuser \
-P \
--table employees \
--where "department = 'Sales' AND salary > 60000" \
--target-dir /data/sales_high_salary_employees \
-m 1

• 说明：这里 -m 1 是因为没有指定 --split-by，Sqoop可能难以安全地并行化任意 WHERE 条件。如果 id 是主键且适合分割，可以尝试 -m > 1 并添加 --split-by id。

2. 使用 --table 和 --columns 导入特定列

(A) 仅导入所有员工的 id, name, 和 department 列到HDFS

sqoop import \
--connect jdbc:mysql://localhost:3306/mydb \
--username dbuser \
-P \
--table employees \
--columns "id,name,department" \
--target-dir /data/employee_subset_columns \
-m 2 \
--split-by id

• 说明：只选择了三列数据进行导入。

3. 结合 --table, --where, 和 --columns

(A) 导入 ‘HR’ 部门员工的 name 和 hire_date 列到Hive表 hr_employees_info

sqoop import \
--connect jdbc:mysql://localhost:3306/mydb \
--username dbuser \
-P \
--table employees \
--columns "name,hire_date" \
--where "department = 'HR'" \
--hive-import \
--hive-table default.hr_employees_info \
--create-hive-table \
--target-dir /temp/hr_staging \
-m 1

4. 使用 --query 导入复杂子集

(A) 导入2022年后入职且薪水排名前10的员工的ID、姓名和薪水到HDFS，按薪水降序

sqoop import \
--connect jdbc:mysql://localhost:3306/mydb \
--username dbuser \
-P \
--query "SELECT id, name, salary FROM employees WHERE hire_date >= '2022-01-01' ORDER BY salary DESC LIMIT 10 AND \$CONDITIONS" \
--target-dir /data/top_recent_employees \
-m 1
# 注意: 这里的 ORDER BY 和 LIMIT 是在源数据库执行的。
# 如果要并行化(-m > 1)，需要 \$CONDITIONS 和 --split-by (split-by的列必须在SELECT中)
# 且这种带 LIMIT 的查询并行化会比较复杂，通常对于这种取TOP N的场景，-m 1更直接。
# 或者，先全量/较大范围导入，再用Hive/Spark处理排序和LIMIT。

• 说明：\$CONDITIONS 在 -m 1 时可以省略。如果需要并行，并且 id 是主键，可以添加 --split-by id。但对于已经包含 ORDER BY LIMIT 的查询，并行分割的意义和实现需要仔细考虑，可能不如单mapper直接。

四、核心注意事项

1. --where 和 --query 中的SQL语法：确保遵循源数据库的SQL语法。
2. 并行化与 \$CONDITIONS：当使用 --query 并行导入 (-m > 1) 时，\$CONDITIONS 占位符至关重要，它允许Sqoop为每个Map任务生成不同的数据范围。同时，必须配合 --split-by (指定用于分割的列，该列必须在 --query 的 SELECT列表中) 和可选的 --boundary-query (如果Sqoop无法自动获取分割列的边界)。
3. --split-by 列的选择：无论是 --table 还是 --query 模式，选择一个合适的 --split-by 列对于并行导入的效率和数据均衡非常重要。理想的列是有索引、数据分布均匀的数值型或日期型列。
4. Hive表创建：当使用 --columns 或 --query 导入到Hive并使用 --create-hive-table 时，Sqoop会根据选择的列来创建Hive表结构。
5. 性能考虑：复杂的 --where 条件或 --query 可能会增加源数据库的查询负载。如果可能，尽量利用源数据库的索引。

---

练习题与解析

假设环境：

• MySQL数据库 ecommerce，包含表 orders (order_id INT PRIMARY KEY, customer_id INT, order_date DATE, total_amount DECIMAL(10,2), status VARCHAR(20), shipping_city VARCHAR(100))。
• Hadoop集群已配置。
• MySQL连接信息：jdbc:mysql://db.example.com:3306/ecommerce，用户 importer，密码存储在HDFS文件 /user/sqoop_pass/ecommerce.pass。

题目：

1. 练习题1：按条件过滤行导入HDFS
   请编写Sqoop命令，将 ecommerce.orders 表中所有状态 (status) 为 ‘COMPLETED’ 并且订单金额 (total_amount) 大于 1000 的订单数据，导入到HDFS的 /retail_data/completed_high_value_orders 目录下。使用默认的并行度，并尝试让Sqoop自动选择分割列 (假设 order_id 是主键)。

2. 练习题2：选择特定列并按条件导入Hive
   请编写Sqoop命令，仅导入 ecommerce.orders 表中 shipping_city 为 ‘New York’ 的订单的 order_id, customer_id, 和 order_date 这三列数据。将这些数据导入到Hive表 mart.ny_orders_subset。如果Hive表不存在则创建，如果存在则追加 (注意：Sqoop默认是覆盖，追加需要特定技巧或多步骤，这里假设我们接受覆盖或手动处理追加)。使用单个Map任务。

3. 练习题3：使用自定义查询导入特定列和行到HDFS
   请编写Sqoop命令，使用自定义查询从 ecommerce.orders 表中选择2023年1月份 (即 order_date 在 ‘2023-01-01’ 和 ‘2023-01-31’ 之间) 的所有订单的 order_id 和 total_amount。将结果导入到HDFS的 /finance_reports/jan_2023_orders_summary 目录，并确保使用2个Map任务并行处理，以 order_id 进行分割。

解析：

1. 练习题1答案与解析：

sqoop import \
--connect jdbc:mysql://db.example.com:3306/ecommerce \
--username importer \
--password-file /user/sqoop_pass/ecommerce.pass \
--table orders \
--where "status = 'COMPLETED' AND total_amount > 1000" \
--target-dir /retail_data/completed_high_value_orders
# 默认 -m 4，Sqoop会尝试使用主键 order_id 进行分割

• --table orders --where "...": 从 orders 表中根据 status 和 total_amount 进行过滤。
• --password-file: 使用密码文件。
• 没有显式指定 -m，Sqoop将使用默认的4个mapper。
• 没有显式指定 --split-by，Sqoop会尝试使用主键 (order_id)进行数据分割。

2. 练习题2答案与解析：

sqoop import \
--connect jdbc:mysql://db.example.com:3306/ecommerce \
--username importer \
--password-file /user/sqoop_pass/ecommerce.pass \
--table orders \
--columns "order_id,customer_id,order_date" \
--where "shipping_city = 'New York'" \
--hive-import \
--hive-table mart.ny_orders_subset \
--create-hive-table \
# --hive-overwrite (如果需要覆盖)
# 要实现追加，通常需要先导入到HDFS临时目录，再用Hive的LOAD DATA INPATH ... INTO TABLE ... APPEND
# 或者使用Sqoop的--append参数（但这通常用于增量导入场景，且对目标HDFS目录有要求）
# 为简单起见，这里只演示创建/覆盖，如果题目明确要求追加，则需要更复杂操作或说明其限制。
# 此题中说“如果存在则追加”，但Sqoop import到hive的行为默认是覆盖（如果用了--hive-overwrite）或失败（如果表已存在且没用--hive-overwrite）。
# 若要严格追加，一般做法是：
# 1. sqoop import --table ... --where ... --columns ... --target-dir /temp/ny_orders_subset_new -m 1
# 2. hive -e "LOAD DATA INPATH '/temp/ny_orders_subset_new' INTO TABLE mart.ny_orders_subset;" (如果表已存在)
# 或者，如果表不存在，先创建再加载。
# 为了简化sqoop命令本身，我们这里假设--create-hive-table会处理，或者如果表已存在就覆盖。
--target-dir /temp/hive_staging_ny_orders \
-m 1

• --columns "order_id,customer_id,order_date": 只选择这三列。
• --where "shipping_city = 'New York'": 过滤出纽约市的订单。
• --hive-import --hive-table mart.ny_orders_subset --create-hive-table: 导入到Hive，如果表不存在则创建。
• --target-dir /temp/hive_staging_ny_orders: 为Hive导入指定一个HDFS上的临时/暂存目录。
• -m 1: 使用单个Map任务。
• 关于追加：Sqoop 的 --append 参数主要用于增量导入到HDFS目录，并且要求目标数据是基于上次导入的最大值进行追加的。直接通过 sqoop import --hive-import 实现对Hive表的“追加”比较tricky，通常如果表已存在且没有 --hive-overwrite，命令会失败。如果用了 --hive-overwrite 则是覆盖。要实现纯粹的追加，标准做法是先用Sqoop将子集导入HDFS的一个新目录，然后使用Hive的 LOAD DATA INPATH ... INTO TABLE ... (不带 OVERWRITE)命令将HDFS的数据加载到Hive表中。

3. 练习题3答案与解析：

sqoop import \
--connect jdbc:mysql://db.example.com:3306/ecommerce \
--username importer \
--password-file /user/sqoop_pass/ecommerce.pass \
--query "SELECT order_id, total_amount FROM orders WHERE order_date >= '2023-01-01' AND order_date <= '2023-01-31' AND \$CONDITIONS" \
--split-by order_id \
--target-dir /finance_reports/jan_2023_orders_summary \
-m 2

• --query "...": 使用自定义查询选择1月份的订单ID和总金额，并包含 \$CONDITIONS。
• --split-by order_id: 告知Sqoop根据 order_id 列来分割自定义查询的结果集给2个Map任务。
• --target-dir ...: 数据导入到HDFS的指定目录。
• -m 2: 使用2个Map任务。