离线数仓（三）:业务数据采集模块搭建

超、凢脫俗 2023-01-02 01:30 208阅读 0赞

### 文章目录 ###

*  一、MySQL 安装
 *  二、Sqoop 安装
 *  三、同步策略
 *  四、业务数据导入 HDFS
 *  五、Hive 环境准备
 *   *  5.1 基本使用
     *  5.2 Hive On Spark 配置
     *  5.3 Yarn 容量调度器队列配置

**服务架构设计：**  
![在这里插入图片描述][20210105160038827.png]

# 一、MySQL 安装 #

**① 卸载自带的 Mysql-libs**

[root@hadoop100 software]# rpm -qa | grep -i -E mysql\|mariadb | xargs -n1 rpm -e --nodeps

**② 安装 mysql 依赖**

[root@hadoop100 software]# rpm -ivh mysql-community-common-5.7.29-1.el7.x86_64.rpm
    [root@hadoop100 software]# rpm -ivh mysql-community-libs-5.7.29-1.el7.x86_64.rpm
    [root@hadoop100 software]# rpm -ivh mysql-community-libs-compat-5.7.29-1.el7.x86_64.rpm

**③ 安装 mysql-client 和 mysql-server**

[root@hadoop100 software]# rpm -ivh mysql-community-client-5.7.29-1.el7.x86_64.rpm
    [root@hadoop100 software]# rpm -ivh mysql-community-server-5.7.29-1.el7.x86_64.rpm

**④ 启动 mysql**

[root@hadoop100 software]# systemctl start mysqld

**⑤ 配置 mysql 密码**

1.  查看`mysql`密码：`# cat /var/log/mysqld.log | grep password`
2.  进入到`mysql`：`# mysql -uroot -p’password’`
3.  更改`mysql`密码策略：`mysql> set global validate_password_length=4;`和`mysql> global validate_password_policy=0;`
4.  设置简单密码：`mysql> set password=password("1234");`

**⑥ 修改登录权限**

1.  选择`mysql`库：`mysql> use mysql`
2.  查询`user`表：`mysql> select user, host from user;`
3.  修改`user`表，把`Host`表内容修改为`%`：`mysql> update user set host="%" where user="root";`
4.  刷新规则：`mysql> flush privileges;`

# 二、Sqoop 安装 #

**① 解压 sqoop 安装包**

[root@hadoop100 software]# tar -zxf sqoop-1.4.6.bin__hadoop-2.0.4-alpha.tar.gz -C /opt/module/
    
    [root@hadoop100 module]# mv sqoop-1.4.6.bin__hadoop-2.0.4-alpha/ sqoop-1.4.6

**② 修改配置文件**

[root@hadoop100 conf]# mv sqoop-env-template.sh sqoop-env.sh && vim sqoop-env.sh
    export HADOOP_COMMON_HOME=/opt/module/hadoop-3.1.3
    export HADOOP_MAPRED_HOME=/opt/module/hadoop-3.1.3
    export HIVE_HOME=/opt/module/hive-3.1.2
    export ZOOKEEPER_HOME=/opt/module/zookeeper-3.5.7
    export ZOOCFGDIR=/opt/module/zookeeper-3.5.7/conf

**③ 拷贝 JDBC 驱动**

[root@hadoop100 software]# cp mysql-connector-java-5.1.48.jar /opt/module/sqoop-1.4.6/lib/

**④ 验证 Sqoop**

[root@hadoop100 sqoop-1.4.6]# bin/sqoop help

⑤ 测试 Sqoop 是否能够成功连接数据库

[root@hadoop100 sqoop-1.4.6]# bin/sqoop list-databases --connect jdbc:mysql://hadoop100:3306/ --username root --password 1234

# 三、同步策略 #

数据同步策略的类型包括：全量表、增量表、新增及变化表、特殊表

*  全量表：存储完整的数据；
 *  增量表：存储新增加的数据；
 *  新增及变化表：存储新增加的数据和变化的数据；
 *  特殊表：只需要存储一次；

**① 全量同步策略**

每日全量，就是每天存储一份完整数据，作为一个分区。

适用于表数据量不大，且每天既会有新数据插入，也会有旧数据的修改的场景。

例如：编码字典表、品牌表、商品三级分类、商品二级分类、商品一级分类、优惠规则表、活动表、活动参与商品表、加购表、商品收藏表、优惠卷表、SKU商品表、SPU商品表

**② 增量同步策略**

每日增量，就是每天存储一份增量数据，作为一个分区。

适用于表数据量大，且每天只会有新数据插入的场景。

例如：退单表、订单状态表、支付流水表、订单详情表、活动与订单关联表、商品评论表。

**③ 新增及变化策略**

每日新增及变化，就是存储创建时间和操作时间都是今天的数据。

适用场景为，表的数据量大，既会有新增，又会有变化。

例如：用户表、订单表、优惠卷领用表。

**④ 特殊策略**

某些特殊的维度表，可不必遵循上述同步策略。

*  客观世界维度：没变化的客观世界的维度（比如性别，地区，民族，政治成分，鞋子尺码）可以只存一份固定值
 *  日期维度：日期维度可以一次性导入一年或若干年的数据
 *  地区维度：省份表、地区表

# 四、业务数据导入 HDFS #

表同步策略：  
![在这里插入图片描述][20210104201142829.png]  
**Sqoop 导入脚本：**

#! /bin/bash
    
    sqoop=/opt/module/sqoop-1.4.6/bin/sqoop
    do_date=`date -d '-1 day' +%F`
    
    if [[ -n "$2" ]]; then
        do_date=$2
    fi
    
    import_data(){ 
    $sqoop import \
    --connect jdbc:mysql://hadoop100:3306/mall \
    --username root \
    --password 1234 \
    --target-dir /origin_data/mall/db/$1/$do_date \
    --delete-target-dir \
    --query "$2 and \$CONDITIONS" \
    --num-mappers 1 \
    --fields-terminated-by '\t' \
    --compress \
    --compression-codec lzop \
    --null-string '\\N' \
    --null-non-string '\\N'
    
    hadoop jar /opt/module/hadoop-3.1.3/share/hadoop/common/hadoop-lzo-0.4.20.jar \
    com.hadoop.compression.lzo.DistributedLzoIndexer /origin_data/mall/db/$1/$do_date
    }
    
    import_order_info(){ 
      import_data order_info " select id, final_total_amount, order_status, user_id, out_trade_no, create_time, operate_time, province_id, benefit_reduce_amount, original_total_amount, feight_fee from order_info where (date_format(create_time,'%Y-%m-%d')='$do_date' or date_format(operate_time,'%Y-%m-%d')='$do_date')"
    }
    
    import_coupon_use(){ 
      import_data coupon_use "select id, coupon_id, user_id, order_id, coupon_status, get_time, using_time, used_time from coupon_use where (date_format(get_time,'%Y-%m-%d')='$do_date' or date_format(using_time,'%Y-%m-%d')='$do_date' or date_format(used_time,'%Y-%m-%d')='$do_date')"
    }
    
    import_order_status_log(){ 
      import_data order_status_log "select id, order_id, order_status, operate_time from order_status_log where date_format(operate_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_activity_order(){ 
      import_data activity_order "select id, activity_id, order_id, create_time from activity_order where date_format(create_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_user_info(){ 
      import_data "user_info" "select id, name, birthday, gender, email, user_level, create_time, operate_time from user_info where (DATE_FORMAT(create_time,'%Y-%m-%d')='$do_date' or DATE_FORMAT(operate_time,'%Y-%m-%d')='$do_date')"
    }
    
    import_order_detail(){ 
      import_data order_detail "select od.id, order_id, user_id, sku_id, sku_name, order_price, sku_num, od.create_time from order_detail od join order_info oi on od.order_id=oi.id where DATE_FORMAT(od.create_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_payment_info(){ 
      import_data "payment_info"  "select id, out_trade_no, order_id, user_id, alipay_trade_no, total_amount, subject, payment_type, payment_time from payment_info where DATE_FORMAT(payment_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_comment_info(){ 
      import_data comment_info "select id, user_id, sku_id, spu_id, order_id, appraise, comment_txt, create_time from comment_info where date_format(create_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_order_refund_info(){ 
      import_data order_refund_info "select id, user_id, order_id, sku_id, refund_type, refund_num, refund_amount, refund_reason_type, create_time from order_refund_info where date_format(create_time,'%Y-%m-%d')='$do_date'"
    }
    
    import_sku_info(){ 
      import_data sku_info "select id, spu_id, price, sku_name, sku_desc, weight, tm_id, category3_id, create_time from sku_info where 1=1"
    }
    
    import_base_category1(){ 
      import_data "base_category1" "select id, name from base_category1 where 1=1"
    }
    
    import_base_category2(){ 
      import_data "base_category2" "select id, name, category1_id from base_category2 where 1=1"
    }
    
    import_base_category3(){ 
      import_data "base_category3" "select id, name, category2_id from base_category3 where 1=1"
    }
    
    import_base_province(){ 
      import_data base_province "select id, name, region_id, area_code, iso_code from base_province where 1=1"
    }
    
    import_base_region(){ 
      import_data base_region "select id, region_name from base_region where 1=1"
    }
    
    import_base_trademark(){ 
      import_data base_trademark "select tm_id, tm_name from base_trademark where 1=1"
    }
    
    import_spu_info(){ 
      import_data spu_info "select id, spu_name, category3_id, tm_id from spu_info where 1=1"
    }
    
    import_favor_info(){ 
      import_data favor_info "select id, user_id, sku_id, spu_id, is_cancel, create_time, cancel_time from favor_info where 1=1"
    }
    
    import_cart_info(){ 
      import_data cart_info "select id, user_id, sku_id, cart_price, sku_num, sku_name, create_time, operate_time, is_ordered, order_time from cart_info where 1=1"
    }
    
    import_coupon_info(){ 
      import_data coupon_info "select id, coupon_name, coupon_type, condition_amount, condition_num, activity_id, benefit_amount, benefit_discount, create_time, range_type, spu_id, tm_id, category3_id, limit_num, operate_time, expire_time from coupon_info where 1=1"
    }
    
    import_activity_info(){ 
      import_data activity_info "select id, activity_name, activity_type, start_time, end_time, create_time from activity_info where 1=1"
    }
    
    import_activity_rule(){ 
        import_data activity_rule "select id, activity_id, condition_amount, condition_num, benefit_amount, benefit_discount, benefit_level from activity_rule where 1=1"
    }
    
    import_base_dic(){ 
        import_data base_dic "select dic_code, dic_name, parent_code, create_time, operate_time from base_dic where 1=1" 
    }
    
    case $1 in
      "order_info")
         import_order_info
    ;;
      "base_category1")
         import_base_category1
    ;;
      "base_category2")
         import_base_category2
    ;;
      "base_category3")
         import_base_category3
    ;;
      "order_detail")
         import_order_detail
    ;;
      "sku_info")
         import_sku_info
    ;;
      "user_info")
         import_user_info
    ;;
      "payment_info")
         import_payment_info
    ;;
      "base_province")
         import_base_province
    ;;
      "base_region")
         import_base_region
    ;;
      "base_trademark")
         import_base_trademark
    ;;
      "activity_info")
          import_activity_info
    ;;
      "activity_order")
          import_activity_order
    ;;
      "cart_info")
          import_cart_info
    ;;
      "comment_info")
          import_comment_info
    ;;
      "coupon_info")
          import_coupon_info
    ;;
      "coupon_use")
          import_coupon_use
    ;;
      "favor_info")
          import_favor_info
    ;;
      "order_refund_info")
          import_order_refund_info
    ;;
      "order_status_log")
          import_order_status_log
    ;;
      "spu_info")
          import_spu_info
    ;;
      "activity_rule")
          import_activity_rule
    ;;
      "base_dic")
          import_base_dic
    ;;
    
    "first")
       import_base_category1
       import_base_category2
       import_base_category3
       import_order_info
       import_order_detail
       import_sku_info
       import_user_info
       import_payment_info
       import_base_province
       import_base_region
       import_base_trademark
       import_activity_info
       import_activity_order
       import_cart_info
       import_comment_info
       import_coupon_use
       import_coupon_info
       import_favor_info
       import_order_refund_info
       import_order_status_log
       import_spu_info
       import_activity_rule
       import_base_dic
    ;;
    "all")
       import_base_category1
       import_base_category2
       import_base_category3
       import_order_info
       import_order_detail
       import_sku_info
       import_user_info
       import_payment_info
       import_base_trademark
       import_activity_info
       import_activity_order
       import_cart_info
       import_comment_info
       import_coupon_use
       import_coupon_info
       import_favor_info
       import_order_refund_info
       import_order_status_log
       import_spu_info
       import_activity_rule
       import_base_dic
    ;;
    esac

说明：

*  `Hive`中的`Null`在底层是以`\N`来存储，而`MySQL`中的`Null`在底层就是`Null`，为了保证数据两端的一致性。在导出数据时采用`--input-null-string`和`--input-null-non-string`两个参数。导入数据时采用`--null-string`和`--null-non-string`
 *  如果`query`后使用的是双引号，则`$CONDITIONS`前必须加转移符，防止`Shell`识别为自己的变量

# 五、Hive 环境准备 #

## 5.1 基本使用 ##

**① 解压安装**

[root@hadoop100 software]# tar -zxvf apache-hive-3.1.2-bin.tar.gz -C /opt/module/

**② 修改环境变量**

[root@hadoop100 software]# vim /etc/profile
    #HIVE_HOME
    export HIVE_HOME=/opt/module/hive-3.1.2
    export PATH=$PATH:$HIVE_HOME/bin
    
    [root@hadoop100 software]# source /etc/profile

**③ 解决日志 Jar 包冲突，进入hive lib 目录**

[root@hadoop100 lib]# mv log4j-slf4j-impl-2.10.0.jar log4j-slf4j-impl-2.10.0.jar.bak

**④ Hive 元数据配置到 MySql**

1.  将`JDBC`驱动拷贝到`Hive`的`lib`目录下：`cp /opt/software/mysql-connector-java-5.1.48.jar /opt/module/hive/lib/`
2.  配置`Metastore`到`MySql`，在`$HIVE_HOME/conf`目录下新建`hive-site.xml`文件

<?xml version="1.0"?>
    <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
    <configuration>
        <property>
            <name>javax.jdo.option.ConnectionURL</name>
            <value>jdbc:mysql://hadoop100:3306/metastore?useSSL=false</value>
        </property>
    
        <property>
            <name>javax.jdo.option.ConnectionDriverName</name>
            <value>com.mysql.jdbc.Driver</value>
        </property>
    
        <property>
            <name>javax.jdo.option.ConnectionUserName</name>
            <value>root</value>
        </property>
    
        <property>
            <name>javax.jdo.option.ConnectionPassword</name>
            <value>1234</value>
        </property>
    
        <property>
            <name>hive.metastore.warehouse.dir</name>
            <value>/user/hive/warehouse</value>
        </property>
    
        <property>
            <name>hive.metastore.schema.verification</name>
            <value>false</value>
        </property>
    
        <property>
            <name>hive.metastore.uris</name>
            <value>thrift://hadoop100:9083</value>
        </property>
    
        <property>
        <name>hive.server2.thrift.port</name>
        <value>10000</value>
        </property>
    
        <property>
            <name>hive.server2.thrift.bind.host</name>
            <value>hadoop100</value>
        </property>
    
        <property>
            <name>hive.metastore.event.db.notification.api.auth</name>
            <value>false</value>
        </property>
        
        <property>
            <name>hive.cli.print.header</name>
            <value>true</value>
        </property>
    
        <property>
            <name>hive.cli.print.current.db</name>
            <value>true</value>
        </property>
    </configuration>

**⑤ 初始化元数据库**

1.  登陆`MySQL`：`# mysql -uroot -p1234`
2.  新建`Hive`元数据库：`mysql> create database metastore;`
3.  初始化`Hive`元数据库：`# schematool -initSchema -dbType mysql -verbose`

**⑥ 启动 metastore 和 hiveserver2**

1.  创建`hive`服务启动脚本

#!/bin/bash
    HIVE_LOG_DIR=$HIVE_HOME/logs
    
    mkdir -p $HIVE_LOG_DIR
    
    #检查进程是否运行正常，参数1为进程名，参数2为进程端口
    function check_process()
    { 
        pid=$(ps -ef 2>/dev/null | grep -v grep | grep -i $1 | awk '{print $2}')
        ppid=$(netstat -nltp 2>/dev/null | grep $2 | awk '{print $7}' | cut -d '/' -f 1)
        echo $pid
        [[ "$pid" =~ "$ppid" ]] && [ "$ppid" ] && return 0 || return 1
    }
    
    function hive_start()
    { 
        metapid=$(check_process HiveMetastore 9083)
        cmd="nohup hive --service metastore >$HIVE_LOG_DIR/metastore.log 2>&1 &"
        cmd=$cmd" sleep 4; hdfs dfsadmin -safemode wait >/dev/null 2>&1"
        [ -z "$metapid" ] && eval $cmd || echo "Metastroe服务已启动"
        server2pid=$(check_process HiveServer2 10000)
        cmd="nohup hive --service hiveserver2 >$HIVE_LOG_DIR/hiveServer2.log 2>&1 &"
        [ -z "$server2pid" ] && eval $cmd || echo "HiveServer2服务已启动"
    }
    
    function hive_stop()
    { 
        metapid=$(check_process HiveMetastore 9083)
        [ "$metapid" ] && kill $metapid || echo "Metastore服务未启动"
        server2pid=$(check_process HiveServer2 10000)
        [ "$server2pid" ] && kill $server2pid || echo "HiveServer2服务未启动"
    }
    
    case $1 in
    "start")
        hive_start
        ;;
    "stop")
        hive_stop
        ;;
    "restart")
        hive_stop
        sleep 2
        hive_start
        ;;
    "status")
        check_process HiveMetastore 9083 >/dev/null && echo "Metastore服务运行正常" || echo "Metastore服务运行异常"
        check_process HiveServer2 10000 >/dev/null && echo "HiveServer2服务运行正常" || echo "HiveServer2服务运行异常"
        ;;
    *)
        echo Invalid Args!
        echo 'Usage: '$(basename $0)' start|stop|restart|status'
        ;;
    esac

1.  添加执行权限：`# chmod +x hiveservices.sh`
2.  启动`Hive`后台服务：`# hiveservices.sh start`
3.  查看`Hive`后台服务运行情况：`# hiveservices.sh status`
4.  启动`Hive`客户端

## 5.2 Hive On Spark 配置 ##

**① 解压 spark-2.4.5-bin-without-hive.tgz**

[root@hadoop100 software]# tar -zxvf spark-2.4.5-bin-without-hive.tgz -C /opt/module
    [root@hadoop100 module]# mv spark-2.4.5-bin-without-hive spark-2.4.5

**② 配置 Spark 环境变量**

[root@hadoop100 software]# vim /etc/profile
    export SPARK_HOME=/opt/module/spark-2.4.5
    export PATH=$PATH:$SPARK_HOME/bin
    [root@hadoop100 software]# source /etc/profile

**③ 配置 spark 运行环境**

[root@hadoop100 spark-2.4.5]# mv conf/spark-env.sh.template conf/spark-env.sh
    [root@hadoop100 spark-2.4.5]# vim conf/spark-env.sh
    export SPARK_DIST_CLASSPATH=$(hadoop classpath)

**④连接 spark jar 包到 hive ，如 hive 中已存在则跳过**

[root@hadoop100 spark-2.4.5]# ln -s /opt/module/spark-2.4.5/jars/scala-library-2.11.12.jar /opt/module/hive-3.1.2/lib/scala-library-2.11.12.jar
    [root@hadoop100 spark-2.4.5]# ln -s /opt/module/spark-2.4.5/jars/spark-core_2.11-2.4.5.jar /opt/module/hive-3.1.2/lib/spark-core_2.11-2.4.5.jar
    [root@hadoop100 spark-2.4.5]# ln -s /opt/module/spark-2.4.5/jars/spark-network-common_2.11-2.4.5.jar /opt/module/hive-3.1.2/lib/spark-network-common_2.11-2.4.5.jar

**⑤ 新建 spark 配置文件**

[root@hadoop100 hive-3.1.2]# vim conf/spark-defaults.conf
    spark.master                                    yarn
    spark.master                                    yarn
    spark.eventLog.enabled                          true
    spark.eventLog.dir                              hdfs://hadoop100:8020/spark-history
    spark.driver.memory                             2g
    spark.executor.memory                           2g
    
    [root@hadoop100 hive-3.1.2]# hadoop fs -mkdir /spark-history

**⑥ 上传 Spark 依赖到 HDFS**

[root@hadoop100 hive-3.1.2]# hadoop fs -mkdir /spark-jars
    [root@hadoop100 hive-3.1.2]# hadoop fs -put /opt/module/spark-2.4.5/jars/* /spark-jars

**⑦ 修改 hive-site.xml**

[root@hadoop100 hive-3.1.2]# vim hive-site.xml
      
      <property>
        <name>spark.yarn.jars</name>
        <value>hdfs://hadoop100:8020/spark-jars/*</value>
      </property>
      
      
      <property>
        <name>hive.execution.engine</name>
        <value>spark</value>
      </property>

**⑧ Hive on Spark 测试**

hive (default)> create external table student(id int, name string) location '/student';
    hive (default)> insert into table student values(1,'a');
    hive (default)> select * from student;
    OK
    student.id	student.name
    1	a
    Time taken: 0.164 seconds, Fetched: 1 row(s)

## 5.3 Yarn 容量调度器队列配置 ##

默认`Yarn`的配置下，容量调度器只有一条`Default`队列。在`capacity-scheduler.xml`中可以配置多条队列，修改以下属性，增加`hive`队列：

<property>
        <name>yarn.scheduler.capacity.root.queues</name>
        <value>default,hive</value>
        <description> The queues at the this level (root is the root queue)</description>
    </property>
    <property>
        <name>yarn.scheduler.capacity.root.default.capacity</name>
    	<value>50</value>
        <description>default队列的容量为50%</description>
    </property>
    <property>
        <name>yarn.scheduler.capacity.root.hive.capacity</name>
    	<value>50</value>
        <description>hive队列的容量为50%</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.user-limit-factor</name>
    	<value>1</value>
        <description>一个用户最多能够获取该队列资源容量的比例</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.maximum-capacity</name>
    	<value>80</value>
        <description>hive队列的最大容量</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.state</name>
        <value>RUNNING</value>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.acl_submit_applications</name>
    	<value>*</value>
        <description>访问控制，控制谁可以将任务提交到该队列</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.acl_administer_queue</name>
    	<value>*</value>
        <description>访问控制，控制谁可以管理(包括提交和取消)该队列的任务</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.acl_application_max_priority</name>
    	<value>*</value>
    	<description>访问控制，控制用户可以提交到该队列的任务的最大优先级</description>
    </property>
    
    <property>
        <name>yarn.scheduler.capacity.root.hive.maximum-application-lifetime</name>
    	<value>-1</value>
        <description>hive队列中任务的最大生命时长</description>
    </property>
    <property>
        <name>yarn.scheduler.capacity.root.hive.default-application-lifetime</name>
    	<value>-1</value>
        <description>default队列中任务的最大生命时长</description>
    </property>

**配置hive客户端任务提交到hive队列**

为方便后续`hive`客户端的测试和`shell`脚本中的任务能同时执行，我们将`hive`客户端的测试任务提交到`hive`队列，让`shell`脚本中的任务使用默认值，提交到`default`队列。

每次进入`hive`客户端时，执行以下命令：`hive (default)> set mapreduce.job.queuename=hive;`

[20210105160038827.png]: /images/20221120/0c61904f140545f2919511c1f85691a6.png
[20210104201142829.png]: /images/20221120/d558e613ade64b72a04fa28921322773.png