Sqoop与Shell脚本数据迁移实战

本文主要是介绍Sqoop与Shell脚本数据迁移实战，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

---

前言

在数据驱动的时代，高效准确地迁移数据是每个数据工程师不可或缺的技能。本教程将深入探讨如何使用Sqoop工具和Shell脚本，实现从关系型数据库到Hadoop生态系统的数据迁移。通过实战示例，我们将一步步展示如何配置和执行数据导入，确保您能够快速掌握这些关键技术，提升数据处理效率。

---

一、sqoop实战示例

1. 获取所有数据库

sqoop list-databases \--connect "jdbc:mysql://127.0.0.1:3306/" \--username 'root' \--password 'root'

2. 获取指定数据库的所有表

sqoop list-tables \--connect "jdbc:mysql://127.0.0.1:3306/db_name?useSSL=false" \--username 'root' \--password 'root'

3. 查询数据

sqoop eval \--connect "jdbc:mysql://127.0.0.1:3306/db_name?useSSL=false" \--username 'root' \--password 'root' \--query "select * from tb_name limit 10"

4. 把指定数据库的所有表导入指定hive数据库

sqoop import-all-tables \--connect "jdbc:mysql://127.0.0.1:3306/db_name?useSSL=false" \--username 'root' \--password 'root' \--hive-import \--hive-overwrite \--hive-database 'hive_db_name' \--as-parquetfile \--fields-terminated-by '\0001' \--lines-terminated-by '\n' \--hive-drop-import-delims \--compress

5. 把指定表导入hive数据库的指定表

sqoop import \--connect "jdbc:mysql://127.0.0.1:3306/db_name?useSSL=false" \--username 'root' \--password 'root' \--table 'tb_name' \--as-parquetfile \--hive-import \--hive-overwrite \--compress \--hive-database 'hive_db_name' \--hive-table 'hive_tb_name' \--fields-terminated-by '\0001' \--lines-terminated-by '\n' \--hive-drop-import-delims \--null-string '\\N' \--null-non-string '\\N' \-m 1

6. 查询数据导入到指定表

sqoop import \--connect "jdbc:mysql://127.0.0.1:3306/db_name?useSSL=false" \--username 'root' \--password 'root' \--query "select * from tb_name where \$CONDITIONS" \--hcatalog-database 'hive_db_name' \--drop-and-create-hcatalog-table \--hcatalog-table 'hive_tb_name' \--compress \--fields-terminated-by '\0001' \--lines-terminated-by '\n' \--hive-drop-import-delims \--null-string '\\N' \--null-non-string '\\N' \-m 1

---

二、shell脚本实战示例

1. shell脚本

#!/bin/bashDB_DATABASE=''
DB_TABLE=''
DB_HOST=''
DB_PORT=''
DB_CONNECTOR="jdbc:sqlserver://$BS_DB_HOST:$BS_DB_PORT;database=$BS_DB_DATABASE"
DB_USER=''
DB_PASSWORD=''
HIVE_DATABASE='ods_'
HIVE_TABLE="ods_${BS_DB_TABLE}_df"
IMPALA_CMD='impala-shell'
##########################################################################################
FIELDS=''
##########################################################################################
# 创建Impala表
create_impala_table() {echo "Executing DDL for table $1.$2:"# 构建Impala建表语句create_table_sql="CREATE TABLE IF NOT EXISTS $1.$2 (${FIELDS//,/ STRING,} STRING)WITH SERDEPROPERTIES ('serialization.format'='1')STORED AS PARQUETTBLPROPERTIES ('DO_NOT_UPDATE_STATS'='true', 'parquet.compression'='snappy');"echo "$create_table_sql"# 创建表$IMPALA_CMD -q "$create_table_sql"
}
##########################################################################################
# 函数：清空表数据
truncate_table() {$IMPALA_CMD -q "TRUNCATE TABLE $1.$2"
}
##########################################################################################
# 刷新hive元数据
hive_table_refresh() {$IMPALA_CMD -q "REFRESH $1.$2"
}
##########################################################################################
# 获取hive表字段
get_hive_table_fields() {$IMPALA_CMD --quiet -q "DESCRIBE $1.$2;" | awk 'NR>3{print $2}' | paste -sd, | sed '$ s/,$//'
}
##########################################################################################
# 把sqlserver数据导入hive
db_import_to_hive() {local db_connector=$1local db_user=$2local db_pw=$3local table_name=$4local hive_dbname=$5local hive_tbname=$6sqoop import "-Dorg.apache.sqoop.splitter.allow_text_splitter=true" \--connect "$db_connector" \--username "$db_user" \--password "$db_pw" \--query "select $FIELDS from ${table_name} where \$CONDITIONS" \--hcatalog-database "$hive_dbname" \--hcatalog-table "$hive_tbname" \--fields-terminated-by '\0001' \--lines-terminated-by '\n' \--hive-drop-import-delims \--null-string '\\N' \--null-non-string '\\N' \-m 1
}
##########################################################################################
# 如果表不存在则创建表
create_impala_table $HIVE_DATABASE $HIVE_TABLE
# 比较Impala表的字段与脚本字段是否一样
# 开启不区分大小写
#hive_fields=$(get_hive_table_fields $HIVE_DATABASE $HIVE_TABLE)
#shopt -s nocasematch
#if [ "$FIELDS" == "$hive_fields" ]; then
#    echo "字符串内容一致（忽略大小写）"
#else
#    echo "字符串内容不一致"
#fi
# 关闭不区分大小写
#shopt -u nocasematch
# 清空表数据
truncate_table $HIVE_DATABASE $HIVE_TABLE
# 导入数据到hive
db_import_to_hive $DB_CONNECTOR $DB_USER $DB_PASSWORD $DB_TABLE $HIVE_DATABASE $HIVE_TABLE
# 刷新hive元数据
hive_table_refresh $HIVE_DATABASE $HIVE_TABLEexit 0

2. 解释

这个脚本是一个用于数据迁移的Bash脚本，其主要目的是将SQL Server数据库中的数据导入到Hive表中，适用于全量覆盖更新的数据导入。以下是该脚本的步骤和功能的详细解释：

1. 变量初始化：脚本开始部分定义了一些变量，包括数据库和表的相关参数（如DB_DATABASE, DB_TABLE等），以及 Impala 的命令 IMPALA_CMD。

2. 创建Impala表：函数 create_impala_table 用于在 Impala 中创建一个表。这个表是根据 Hive 表的结构来定义的，使用了 Parquet 格式。

3. 清空表数据：函数 truncate_table 用于删除 Impala 表中的所有数据。

4. 刷新hive元数据：函数 hive_table_refresh 用于在 Impala 中刷新 Hive 表的元数据，可能是为了确保 Impala 表反映最新的 Hive 表信息。

5. 获取hive表字段：函数 get_hive_table_fields 用于查询 Hive 表的字段，并返回字段列表。

6. 数据库数据导入Hive：函数 db_import_to_hive 使用 Sqoop 工具从 SQL Server 数据库导入数据到 Hive 表中。这里使用了 --query 参数来指定要导入的数据，以及一些其他参数来定义字段分隔符和行分隔符等。

7. 执行脚本：

   • 首先检查 Impala 表是否存在，如果不存在则创建。
   • 清空 Impala 表的数据。
   • 导入 SQL Server 数据库的数据到 Hive 表。
   • 刷新 Impala 表的元数据，以确保数据一致性。

---

总结

通过本教程的学习，您已经了解了如何使用Sqoop和Shell脚本进行数据迁移的详细步骤和技巧。从基本的Sqoop命令到复杂的Shell脚本编写，每一步都旨在帮助您更高效地管理和迁移数据。希望这些知识能助您在数据处理的道路上更进一步，实现数据价值的最大化。

希望本教程对您有所帮助！如有任何疑问或问题，请随时在评论区留言。感谢阅读！

参考连接：

• Sqoop用户指南

这篇关于Sqoop与Shell脚本数据迁移实战的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---