用户日志分析系统实战（三）

朱雀 2022-06-05 04:15 216阅读 0赞

### 用户日志分析系统实战（三） ###

#### 1 资源管理（YARN） ####

##### 1.1 YARN产生背景 #####

*  运维成本  
    
    
     *  “一个框架一个集群”的模式会增加运维成本，而共享模式通常需要少数管理员即可完成多个框架的统一管理。
     *  数据共享。共享集群模式可让多种框架共享数据和硬件资源，将大大减少数据移动带来的成本。

##### 1.2 YARN #####

*  YARN由来
    
     *  Yet Another Resource Negotiator的简称
     *  出现在Hadoop 2.x版本中
 *  YARN是什么
    
     *  集群资源管理系统
     *  负责集群的统一管理和调度
 *  YARN运行过程

![这里写图片描述][SouthEast]

*  YARN资源调度器
    
     *  多类型资源调度  
        
        
         *  采用DRF算法
         *  目前支持CPU和内存两种资源
     *  提供多种资源调度器  
        
        
         *  FIFO  
            
            
             *  将所有应用程序放到一个队列中
             *  队列前面的应用程序先获得资源
             *  局限性明显：  
                
                
                 *  资源利用率低，无法交叉运行作业
                 *  不够灵活，比如紧急的作业无法插队
         *  多队列调度器  
            
            
             *  Fair Scheduler和Capacity Scheduler
             *  将所有应用程序放到多个队列中
             *  每个队列可单独实现调度策略
             *  每个队列对应一定比例的资源
             *  优势明显：  
                
                
                 *  按队列组织资源和用户，符合生产需要
                 *  不同队列的资源分配策略不同，更加灵活
     *  多租户资源调度器  
        
        
         *  支持资源按比例分配
         *  支持层级队列划分方式
         *  支持资源抢占
 *  Capacity Scheduler（如下图所示）
    
     *  由Yahoo开源，共享集群调度器
     *  以队列方式组织作业
     *  每个队列内部采用FIFO调度策略
     *  每个队列分配一定比例资源
     *  可限制每个用户使用资源量  
        ![这里写图片描述][SouthEast 1]
 *  基于标签的调度机制：动机
    
     *  YARN基于标签的调度
     *  思路：  
        
        
         *  将一些高配置机器打上highmemory/highdisk标签，并结合队列配置（比如reporting队列）使之生效
         *  将所有mapreduce程序提交到mr队列中  
            
            
             *  hadoop jar XXX.jar -Dmapreduce.job.queuename=mr….
 *  基于标签的调度机制：原理  
    ![这里写图片描述][SouthEast 2]
 *  基于标签的调度机制：启用步骤
    
     *  设置系统级别的标签
        
            yarn rmadmin -addToClusterNodeLabels "normal,highmem"
     *  给指定节点打标签
        
            yarn rmadmin -replaceLabelsOnNode "chinahadoop-1:35895,normal"
     *  修改Capacity Scheduler，设置队列对应的标签
        
            <property>
                <name>yarn.scheduler.capacity.root.hadoop.accessible-node-labels.highmem.capacity</name>
                <value>10</value>
            </property>
            <property>
                <name>yarn.scheduler.capacity.root.hadoop.default-node-label-expression</name>
                <value>normal</value>
            </property>
 *  资源划分  
    在capacity-scheduler.xml中配置:  
    ![这里写图片描述][SouthEast 3]

#### 2 ETL（MR） ####

##### 2.1 MR #####

*  MR数据流

![这里写图片描述][SouthEast 4]

![这里写图片描述][SouthEast 5]

针对MR详细执行流程，请参照我的博客[MapReduce作业执行流程][MapReduce]

*  MR编程模型  
    
    
     *  MR将整个作业划分为两个阶段：Map阶段和Reduce阶段
     *  Map阶段由一定数量的Map Task组成  
        
        
         *  输入数据格式解析：InputFormat
         *  输入数据处理：Mapper
         *  数据分组：Partition
     *  Reduce阶段由一定数量的Reduce Task组成  
        
        
         *  数据远程拷贝
         *  数据按key排序
         *  数据处理：Reducer
         *  数据输出格式：OutputFormat

##### 2.2 ETL #####

*  Extract & Transform代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    
    import sys,time
    
    
    def timestamp_datetime(value):
        format = '%Y-%m-%d %H:%M:%S'
        value = time.localtime(value)
        dt = time.strftime(format,value)
        return dt
    
    # input comes from STDIN (standard input)
    for line in sys.stdin:
        # remove leading and trailing whitespace
        columns = line.split(",");
    
        if len(columns) != 12:
            continue
    
        for i in range(len(columns)):
            columns[i] = columns[i].strip()
    
        columns[3]=timestamp_datetime(int(columns[3]))
        print columns[0]+","+columns[1]+","+columns[2]+","+columns[3]+","+columns[4]+","+columns[5]+","+columns[6]+","+columns[7]+","+columns[8]+","+columns[9]

*  Hadoop Streaming代码:

#!/usr/bin/env python
    # -*- coding: utf-8 -*-
    
    import os,time
    
    INTPUT_PATH="hdfs://bigdata:9000/flume/record/"
    OUTPUT_PATH="hdfs://bigdata:9000/etl/record/"
    LOAD_CMD="java -cp /home/bigdata/etl/etl-1.0-SNAPSHOT-jar-with-dependencies.jar bigdata.etl.loadDataToHive %s %s %s"
    HADOOP_CMD="hadoop jar /home/bigdata/hadoop-2.7.3/share/hadoop/tools/lib/hadoop-streaming-2.7.3.jar -D mapred.reduce.tasks=0 -D mapred.map.tasks=1 -input %s -output %s -mapper /home/bigdata/etl/etl.py -file /home/bigdata/etl/etl.py"
    
    def getCurrentYmdHM():
        time_struct=time.localtime(time.time()-60*10)
    
        H= time.strftime('%H',time_struct)
        M= int(time.strftime('%M',time_struct))
        M= "%02d" % ((M/10)*10)
        Ymd=time.strftime('%Y-%m-%d',time_struct)
    
        return Ymd+"/"+H+M
    
    def startETL():
        subPath=getCurrentYmdHM()
        input=INTPUT_PATH+subPath
        output=OUTPUT_PATH+subPath
    
        hadoop_cmd=HADOOP_CMD %(input, output)
        print hadoop_cmd
        os.system(hadoop_cmd)
        print 'loading data into Hive'
        load_cmd= LOAD_CMD %(output,subPath.split("/")[0],subPath.split("/")[1])
        os.system(load_cmd)
    
    if __name__ == '__main__':
        startETL()

*  Hadoop Command:

hadoop jar /home/bigdata/hadoop-2.7.3/share/hadoop/tools/lib/hadoop-streaming-2.7.3.jar -D mapred.reduce.tasks=0 -D mapred.map.tasks=1 -input %s-output %s-mapper /home/bigdata/etl/etl.py -file /home/bigdata/etl/etl.py

*  Load

package bigdata.etl;
    
    import java.sql.*;
    
    /** * Created by qianxi.zhang on 11/25/16. */
    public class loadDataToHive { 
      private static String LOAD_CMD =
          "load data inpath '%s' overwrite into table record partition(partition_date='%s',hour_minute='%s')";
    
      private static String driverName = "org.apache.hive.jdbc.HiveDriver";
    
      public static void loadData(String dataDir, String date, String hour_minute) throws SQLException {
        try {
          Class.forName(driverName);
        } catch (ClassNotFoundException e) {
          e.printStackTrace();
          System.exit(1);
        }
        Connection con =
            DriverManager.getConnection("jdbc:hive2://bigdata:10000/default", "bigdata", "bigdata");
        Statement stmt = con.createStatement();
        String sql = String.format(LOAD_CMD, dataDir, date, hour_minute);
        stmt.execute(sql);
      }
    
      public static void main(String[] args) throws SQLException {
        if (args.length != 3) throw new IllegalArgumentException("need 3 args");
        loadDataToHive.loadData(args[0], args[1], args[2]);
      }
    }

#### 3 MR优化小技巧 ####

##### 3.1 内存参数调优 #####

（1）mapreduce.map.memory.mb  
（2）mapreduce.reduce.memory.mb  
（3）mapreduce.map.java.opts  
（4）mapreduce.reduce.java.opts

![这里写图片描述][SouthEast 6]

e g:

-Dmapreduce.map.memory.mb=5120
    -Dmapreduce.reduce.memory.mb=5120
    -Dmapreduce.map.java.opts="-Xms4096m -Xmx4096m"
    -Dmapreduce.reduce.java.opts="-Xms4096m -Xmx4096m"

Note: -Dmapreduce.map.java.opts配置的值一定要比-Dmapreduce.map.memory.mb小，reduce也一样。

##### 3.2 容错参数 #####

![这里写图片描述][SouthEast 7]

*  mapreduce.map.maxattempts
    
     *  map task重试次数，默认是4
 *  mapreduce.reduce.maxattempts
    
     *  reduce task重试次数，默认是4
 *  mapreduce.map.failures.maxpercent
    
     *  最大允许的map失败率，默认是0
 *  mapreduce.reduce.failures.maxpercent
    
     *  最大允许的reduce失败率，默认是0

##### 3.3 推测执行 #####

![这里写图片描述][SouthEast 8]

*  mapreduce.map.speculative
    
     *  是否打开map阶段的推测执行机制
 *  mapreduce.reduce.speculative
    
     *  是否打开reduce阶段的推测执行机制

##### 3.4 调整作业的任务数目 #####

![这里写图片描述][SouthEast 9]

*  mapreduce.input.fileinputformat.split.minsize  
    
    
     *  每个map task处理的数据量（大于一个block的大小，以减少map task数目）

[SouthEast]: /images/20220605/1367c03d5cca44fdbe672ad53bdc5129.png
[SouthEast 1]: /images/20220605/b4439bb71ec94a2293db7e9ea0c4669e.png
[SouthEast 2]: /images/20220605/9024dd46088e4a64b26b31ed95bd03cb.png
[SouthEast 3]: /images/20220605/686c564d8a724f128f51f1478845dd82.png
[SouthEast 4]: /images/20220605/3e67192637fb48e9a776a8d6cb8eda81.png
[SouthEast 5]: /images/20220605/6b3bb90334f1473795a137a711afcef7.png
[MapReduce]: http://blog.csdn.net/qq_17776287/article/details/78176515
[SouthEast 6]: /images/20220605/1a7cc6bb46124b13a7ebe09322d39924.png
[SouthEast 7]: /images/20220605/0bd4537ae1b44d7a805f0752a206b552.png
[SouthEast 8]: /images/20220605/4802844c663547db83cb0e6f0ca5e662.png
[SouthEast 9]: /images/20220605/5b1c33e467394444ab0dad590687eaba.png