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/bash

DB_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 PARQUET
    TBLPROPERTIES ('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=$1
    local db_user=$2
    local db_pw=$3
    local table_name=$4
    local hive_dbname=$5
    local hive_tbname=$6

    sqoop 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_TABLE

exit 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用户指南