婚纱外贸soho建哪种网站好,程序开发用什么笔记本,注册网站要注意什么,django 企业网站开发作者#xff1a;IvanCodes 日期#xff1a;2025年6月7日 专栏#xff1a;Sqoop教程 Apache Sqoop 不仅擅长从关系型数据库 (RDBMS) 向 Hadoop (HDFS, Hive, HBase) 导入数据#xff0c;同样也强大地支持反向操作——将存储在 Hadoop 中的数据导出 (Export) 回关系型数据库。… 作者IvanCodes 日期2025年6月7日 专栏Sqoop教程 Apache Sqoop 不仅擅长从关系型数据库 (RDBMS) 向 Hadoop (HDFS, Hive, HBase) 导入数据同样也强大地支持反向操作——将存储在 Hadoop 中的数据导出 (Export) 回关系型数据库。这在数据分析结果回写、数据仓库ETL、业务系统数据填充等场景中非常有用。 一、Sqoop 导出基础
核心目标将HDFS上的文件数据 (通常是文本文件如CSV、TSV) 解析并插入或更新到目标RDBMS的表中。 关键组件Sqoop Export 会启动MapReduce作业 (或在Sqoop 2中可能是其他执行引擎) 来并行读取HDFS上的数据并将其转换为SQL语句 (INSERT或UPDATE) 在目标数据库上执行。
通用导出参数
sqoop export: Sqoop导出的主命令。--connect jdbc-uri: 目标RDBMS的JDBC连接字符串。--username user: 数据库用户名。--password pass: 数据库密码 (生产环境建议使用密码文件或credential provider)。--table table-name: 目标RDBMS中的表名。--export-dir hdfs-path: 包含源数据的HDFS目录路径。目录下的文件应为文本格式且字段与目标表列对应。--input-fields-terminated-by char: HDFS文件中字段间的分隔符默认为逗号 ,。--input-lines-terminated-by char: HDFS文件中行间的分隔符默认为换行符 \n。--columns col1,col2,...: (可选) 指定HDFS文件中哪些列以及它们的顺序要导出到目标表的对应列。如果不指定Sqoop会假设HDFS文件的列顺序和数量与目标表完全匹配。--m num-mappers 或 --num-mappers num-mappers: 指定并行导出的Map任务数量。
二、默认导出模式 (INSERT 模式)
行为 默认情况下Sqoop Export 会将HDFS中的每一行数据都尝试作为一条新的记录 INSERT 到目标数据库表中。 如果目标表有主键或唯一约束并且HDFS中的数据导致了重复键则该条记录的INSERT操作会失败 (具体行为取决于数据库可能抛出错误并中止该Map任务的部分批次或整个作业失败)。
适用场景
目标表是空的。确定HDFS中的数据在目标表中不存在重复主键。目标表允许重复记录 (如果没有主键或唯一约束)。
一般结构
sqoop export \
--connect jdbc:mysql://db_host:3306/db_name \
--username db_user --password db_password \
--table target_table \
--export-dir /user/hadoop/hdfs_data_dir \
[--input-fields-terminated-by \t] \
[--columns col_a,col_b,col_c] \
--m 4代码案例 假设HDFS目录 /user/data/new_employees 下有如下CSV文件内容 (假设字段分隔符为 ,) part-m-00000:
101,Alice,HR,60000
102,Bob,Engineering,75000目标MySQL表 employees_target (emp_id INT PRIMARY KEY, name VARCHAR(50), department VARCHAR(50), salary INT)。
导出命令
sqoop export \
--connect jdbc:mysql://mysql.example.com:3306/companydb \
--username export_user --password p$$wOrd \
--table employees_target \
--export-dir /user/data/new_employees \
--input-fields-terminated-by , \
--m 1这会将HDFS中的两条记录作为新行插入到 employees_target 表中。
三、更新导出模式 (--update-key 与 --update-mode)
当HDFS中的数据可能包含目标数据库表中已存在的记录并且你希望更新这些记录而不是简单插入 (或因主键冲突而失败) 时需要使用更新导出模式。
关键参数
--update-key column-name(s): 必需。指定一个或多个用逗号分隔的列名这些列构成了目标表的主键或唯一键Sqoop将使用这些列来匹配HDFS记录与数据库表中的现有行。--update-mode mode: 必需。定义当找到匹配记录时的行为。主要有两种模式 updateonly (默认如果指定了 --update-key 但未指定 --update-mode)allowinsert
3.1 更新导出updateonly 模式
行为
Sqoop会读取HDFS中的每一行数据。对于每一行它会使用 --update-key 指定的列的值去目标数据库表中查找匹配的记录。 如果找到匹配记录则更新 (UPDATE) 该数据库记录使用HDFS行中其他列的值。如果未找到匹配记录则忽略该HDFS行不会进行插入操作。
适用场景
你只想更新数据库中已存在的记录不希望HDFS中额外的数据被插入到数据库中。
一般结构
sqoop export \
--connect jdbc-uri \
--username user --password pass \
--table table-name \
--export-dir hdfs-path \
--update-key primary-key-column \
--update-mode updateonly \
[--columns col_to_update1,col_to_update2,primary_key_column] \
--m num-mappers重要如果使用了 --columns确保 --update-key 指定的列也包含在 --columns 列表中并且顺序正确以便Sqoop能正确解析出用于匹配的键值。
代码案例 HDFS目录 /user/data/employee_updates 内容 (字段分隔符 ,顺序: emp_id, new_salary, new_department) part-m-00000:
101,65000,Sales
103,80000,Engineering目标MySQL表 employees_target 中已有记录
(101, Alice, HR, 60000)
(102, Bob, Engineering, 75000)导出命令 (updateonly)
sqoop export \
--connect jdbc:mysql://mysql.example.com:3306/companydb \
--username export_user --password p$$wOrd \
--table employees_target \
--export-dir /user/data/employee_updates \
--update-key emp_id \
--update-mode updateonly \
--columns emp_id,salary,department \
--input-fields-terminated-by , \
--m 1结果
ID为 101 的员工 Alice 的薪水更新为 65000部门更新为 Sales。ID为 103 的记录在数据库中不存在所以会被忽略。ID为 102 的员工 Bob 的信息保持不变。
3.2 更新导出allowinsert 模式
行为
Sqoop会读取HDFS中的每一行数据。对于每一行它会使用 --update-key 指定的列的值去目标数据库表中查找匹配的记录。 如果找到匹配记录则更新 (UPDATE) 该数据库记录。如果未找到匹配记录则插入 (INSERT) 该HDFS行作为一条新记录到数据库表中。 这种行为有时被称为 “Upsert” (Update or Insert)。
适用场景
你需要同步HDFS中的数据到数据库如果记录已存在则更新如果不存在则插入。
一般结构
sqoop export \
--connect jdbc-uri \
--username user --password pass \
--table table-name \
--export-dir hdfs-path \
--update-key primary-key-column \
--update-mode allowinsert \
[--columns col1,col2,primary_key_column,col3] \
--m num-mappers代码案例 使用与 updateonly 案例相同的HDFS数据和初始数据库状态。 导出命令 (allowinsert)
sqoop export \
--connect jdbc:mysql://mysql.example.com:3306/companydb \
--username export_user --password p$$wOrd \
--table employees_target \
--export-dir /user/data/employee_updates \
--update-key emp_id \
--update-mode allowinsert \
--columns emp_id,salary,department,name \
--input-fields-terminated-by , \
--m 1注意为了演示插入假设HDFS数据中包含 name 列并且目标表 employees_target 有 name 列。如果HDFS数据中没有 name而表定义中有则插入时该列可能为 NULL 或默认值 (取决于表定义)。为简化我们假设HDFS数据是 emp_id,salary,department,name 顺序。 HDFS /user/data/employee_updates 内容 (顺序: emp_id, salary, department, name): part-m-00000:
101,65000,Sales,Alice_Updated
103,80000,Engineering,David_New结果
ID为 101 的员工 Alice 的薪水更新为 65000部门更新为 Sales名字更新为 Alice_Updated。ID为 103 的记录在数据库中不存在所以会作为新记录插入 (David_New, Engineering, 80000)。ID为 102 的员工 Bob 的信息保持不变。
四、导出过程中的 Staging Table
为了提高导出的原子性和可靠性 (尤其是在更新模式下)Sqoop 可以使用一个临时的暂存表 (staging table)。
工作流程 Sqoop 首先将HDFS数据批量插入到一个在目标数据库中自动创建的临时暂存表。然后Sqoop 执行一条 (或多条取决于数据库方言) SQL语句将暂存表中的数据与最终目标表进行合并或更新 (基于 --update-key 和 --update-mode)。成功后删除暂存表。 启用暂存表通过 --staging-table staging_table_name 参数指定暂存表名。如果不指定Sqoop 可能不会使用暂存表 (取决于具体情况和数据库)或者自动生成一个临时表名。好处如果在第二步合并/更新时发生错误最终目标表的数据不会被部分修改保持了一致性。
代码案例 (使用暂存表)
sqoop export \
--connect jdbc:mysql://mysql.example.com:3306/companydb \
--username export_user --password p$$wOrd \
--table employees_target \
--export-dir /user/data/employee_updates \
--update-key emp_id \
--update-mode allowinsert \
--columns emp_id,salary,department,name \
--input-fields-terminated-by , \
--staging-table employees_staging_temp \
--clear-staging-table \
--m 1--clear-staging-table: 确保每次运行前如果暂存表已存在则清空它。
五、重要注意事项
目标表结构目标数据库表必须预先存在并且其列类型应与HDFS数据能够兼容转换。事务性Sqoop Export 本身不提供跨多个Map任务或整个作业的端到端事务。每个Map任务会独立地向数据库提交批次。使用暂存表可以增强单个作业运行内的原子性。性能调优 --batch: (已废弃但早期版本有) 控制Sqoop批量执行SQL语句。--num-mappers: 并行度。需要平衡Hadoop集群资源和数据库负载能力。数据库端的索引、表锁定、事务日志等都会影响导出性能。 数据类型转换Sqoop会尝试进行HDFS文本数据到目标数据库列类型的自动转换。确保数据格式正确否则可能导致错误。可使用 --map-column-java 或 --map-column-hive (当从Hive导出时) 指定特定列的Java或Hive类型。空值处理HDFS中的空值表示 (如 \N) 需要与Sqoop的 --input-null-string 和 --input-null-non-string 参数正确配置以便正确地在数据库中插入NULL。
总结 Sqoop Export 提供了多种模式将Hadoop中的数据回写到关系型数据库。理解默认的插入模式以及基于 --update-key 的 updateonly 和 allowinsert 模式并结合暂存表的使用可以灵活高效地完成各种数据同步需求。 练习题 (共5道)
背景 HDFS目录 /user/data/product_feed 包含产品信息CSV格式字段分隔符为逗号 (,)。文件内容示例 part-00000: P101,LaptopX,1200.00,Electronics
P102,DeskLamp,25.50,HomeGoods
P103,CoffeeMug,10.75,Kitchenware字段顺序product_sku, product_name, price, category MySQL数据库 inventory_db 中有一个表 products_live
CREATE TABLE products_live (sku VARCHAR(50) PRIMARY KEY,name VARCHAR(100),current_price DECIMAL(10,2),category_name VARCHAR(50), stock_level INT DEFAULT 0 -- 假设此列不由Sqoop直接更新);题目
首次全量导出假设 products_live 表是空的。写一个Sqoop命令将 /user/data/product_feed 的所有数据导出到 products_live 表。明确指定HDFS数据的列名和顺序 (product_sku,product_name,price,category) 对应到目标表的 sku,name,current_price,category_name。仅更新现有产品价格HDFS目录 /user/data/price_updates 包含需要更新价格的产品信息格式为 sku,new_price。写一个Sqoop命令使用 updateonly 模式根据 sku 更新 products_live 表中的 current_price。更新或插入产品信息HDFS目录 /user/data/daily_product_sync 包含每日的产品信息可能有新产品也可能有现有产品的信息变更格式为 sku,name,price,category。写一个Sqoop命令使用 allowinsert 模式根据 sku 更新或插入记录到 products_live 表。使用暂存表导出在第3题的基础上修改命令使其在导出时使用一个名为 products_sync_stage 的暂存表并在操作开始前清理该暂存表。思考题 如果在执行 updateonly 或 allowinsert 模式的导出时HDFS源数据中 --update-key 指定的列的值在目标数据库表中找不到匹配并且你没有使用 --update-mode allowinsert (即是 updateonly 模式或默认行为)Sqoop会如何处理这条HDFS记录
答案
首次全量导出
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/product_feed \
--input-fields-terminated-by , \
--columns skuproduct_sku,nameproduct_name,current_priceprice,category_namecategory \
--m 1(注意 --columns 的用法这里使用了 target_colsource_col 的格式更准确的应该是直接列出HDFS中的列名Sqoop会按顺序映射。如果HDFS列名与表列名不同且顺序也可能不同则需要预处理HDFS数据或使用更复杂的映射但Sqoop --columns 主要用于选择和排序HDFS列。为了简单起见通常做法是保证HDFS文件中的列顺序和名称能直接对应目标表或仅通过--columns选择子集和重排序。)
更标准的做法是如果HDFS文件列名和顺序与表列名和顺序一致或你能通过--columns使其一致
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/product_feed \
--input-fields-terminated-by , \
--columns product_sku,product_name,price,category \
--map-column-java product_skuString,product_nameString,pricejava.math.BigDecimal,categoryString \
--m 1(这里假设Sqoop能自动按顺序将HDFS的 product_sku 映射到表的 skuproduct_name到name等。如果表列名和HDFS源数据列名不一致Sqoop Export 没有直接的 target_colsource_col 语法通常需要在HDFS端准备好与目标表列序一致的数据或使用 --call 调用存储过程进行更复杂的映射。) 为了简化最常见的做法是确保HDFS文件列顺序与目标表列顺序一致然后Sqoop会自动映射。如果只需导出部分列则用 --columns 指定HDFS中要导出的列名。
让我们假设HDFS文件列序与目标表列序一致且列名在HDFS中就是 sku,name,current_price,category_name:
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/product_feed \
--input-fields-terminated-by , \
--m 1仅更新现有产品价格 (updateonly) (假设 /user/data/price_updates 文件中列顺序是 sku,current_price)
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/price_updates \
--update-key sku \
--update-mode updateonly \
--columns sku,current_price \
--input-fields-terminated-by , \
--m 1更新或插入产品信息 (allowinsert) (假设 /user/data/daily_product_sync 文件中列顺序是 sku,name,current_price,category_name)
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/daily_product_sync \
--update-key sku \
--update-mode allowinsert \
--columns sku,name,current_price,category_name \
--input-fields-terminated-by , \
--m 1使用暂存表导出 (allowinsert)
sqoop export \
--connect jdbc:mysql://your_mysql_host:3306/inventory_db \
--username your_user --password your_password \
--table products_live \
--export-dir /user/data/daily_product_sync \
--update-key sku \
--update-mode allowinsert \
--columns sku,name,current_price,category_name \
--input-fields-terminated-by , \
--staging-table products_sync_stage \
--clear-staging-table \
--m 1思考题答案 如果在执行 updateonly 模式的导出时 (或者指定了 --update-key 但未明确指定 --update-mode此时默认为 updateonly)HDFS源数据中 --update-key 指定的列的值在目标数据库表中找不到匹配的记录Sqoop会简单地忽略这条HDFS记录。它不会尝试插入这条记录也不会报错 (除非有其他问题导致整个批次失败)。这条记录相当于被丢弃了不会影响目标数据库。
