检查依赖数据到达后定时触发任务

你的名字 2022-10-30 11:25 12阅读 0赞

#!/bin/bash
    HOST_NAME='xxxxxxxx'
    PORT=3306
    USER_NAME=root
    PASSWORD='xxxxx'
    
    DBNAME='edg'
    
    # 任务队列变量数组
    export declare -a task_queus=()
    #===========================校验YARN任务队列中是否存在正在运行或就绪的任务========================================
    function IsExistsYarnTask(){
        
      # 声明app状态数组 
      appStates=(NEW NEW_SAVING SUBMITTED ACCEPTED RUNNING)
      # 遍历校验状态任务数
      for(( i=0;i<${appStates[@]};i++ ))
        do
          runingAppIds=`yarn application -list -appStates ${
         appStates[i]} |grep application_ |grep -v yarn-ats |awk '{print $1}'`
          if [[ !${#runingAppIds[@]} -eq 0 ]];then
            return 1;
          fi
        done
      return 0; 
    }
    
    #==========================将当前调度任务task_name加入任务队列task_queues数组中（调度记录，便于后期排查）=======
    function push2TaskQueues(task_name){
        
          # 验证当前队列是否已存在该task_name
          if [[ "${task_queues[@]}" =~ "${task_name}"  ]]; then
             # 当前任务已存在队列中,睡一微秒
             usleep
          else
             # push到任务队列中
             task_queues[${#task_queues[@]}]=${task_name}
          fi
    }
    #==========================校验task_name是否存在任务队列task_queues中===========================================
    function IsInTaskQueues(task_name){
        
          # 验证当前队列是否已存在该task_name
          if [[ "${task_queues[@]}" =~ "${task_name}"  ]]; then
             # 当前任务已存在队列中,睡一微秒
             return 1
          else
             return 0
          fi
    
    }
    #==========================提交任务到YARN=======================================================================
    function submitJob2Yarn(task_name){
        
       # 验证YARN中是否存在任务
       isNotExists=IsExistsYarnTask()
       # 验证task_queues队列中已存在该task_name
       isNotIn=IsInTaskQueues(task_name)
       if [[ $isNotExists -eq 0 -a $isNotIn -eq 0]]; then
         echo "执行${task_name}"
         # 获取当前任务appId
         let appId=0
         while [!$appId -eq 0]
         do
           appId=`yarn application -list -appStates FINISHED |grep application_ |grep -v yarn-ats |awk '{print $1}'`
         done
         # 将当前任务push到任务队列中
         push2TaskQueues(task_name)
       fi
       
    }
    #==========================条件(每天23点整)清空task_queues数组==================================================
    function cleanTaskQueues(task_queues){
        
      cleanHour=$(date "+%H")
      echo $cleanHour
      if [[ ${cleanHour} -eq 23 ]];then
         task_queues=()
      fi
    }
    #===============================================================================================================
    # 任务名称
    task_name=${1}
    # 依赖基础表数组
    str_tables=${2}
    declare -a base_tables=(${str_tables//,/ })
    
    # 变量
    index=0
    
    # 数据日期
    dt=$(date +"%Y%m%d" -d "-2day")
    
    # 链接数据库
    mysql_conn="mysql -h${HOST_NAME} -P${PORT} -u${USER_NAME} -p${PASSWORD} -D${DBNAME} -s -e"
    
    #遍历依赖基础表数组
    if [[ !${#base_tables[@]} -eq 0 ]]; then
      for (( i=0;i<${#base_tables[@]};i++ ))
       do
         # 查询语句
         query_sql="select count(id) from  hdfs_file_desc where table_name='${base_tables[i]}' and arrive_date=${dt}"
         echo "=========================${query_sql}======================"
         rs_count=$($mysql_conn "${query_sql}")
         echo ${rs_count}
         if [[ ${rs_count} -gt 0 ]]; then
             let index+=1
             continue
         else
             break
         fi
      done
    fi
    
    # 验证依赖表是否全部达到
    if [[ ${#base_tables[@]} -eq ${index} ]]; then
      echo "${1}:任务依赖基础表${2}全部准备就绪，任务调度器执行中..."
    else
      echo "${1}:任务依赖基础表${2}未全部准备就绪，请稍后执行..."
    fi