azkaban流程调度

2023-02-25 15:26:09 浏览数 (2)

1.搜集数据

upload.job

代码语言：javascript复制
#upload.job
type=command
command=bash upload.sh

upload.sh

代码语言：javascript复制
#!/bin/bash

#set java env
export JAVA_HOME=/soft/jdk/
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/soft/hadoop/
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

#日志文件存放的目录
log_src_dir=/home/centos/logs/log/ 

#待上传文件存放的目录
log_toupload_dir=/home/centos/logs/toupload/

#得到昨天的日期
day_01=`date -d'-1 day'  %Y-%m-%d`
#得到昨天的年份
syear=`date --date=$day_01  %Y`
#得到昨天的月份
smonth=`date --date=$day_01  %m`
#得到昨天的日份
sday=`date --date=$day_01  %d`

#日志文件上传到hdfs的根路径
hdfs_root_dir=/data/clickLog/$syear/$smonth/$sday

#创建hdfs上的路径文件夹
hadoop fs -mkdir -p $hdfs_root_dir

#读取日志文件的目录，判断是否有需要上传的文件
ls $log_src_dir | while read fileName
do
    if [[ "$fileName" == access.log ]]; then
    # if [ "access.log" = "$fileName" ];then
        date=`date  %Y_%m_%d_%H_%M_%S`
        #将文件移动到待上传目录并重命名
        mv $log_src_dir$fileName $log_toupload_dir"xxxxx_click_log_$fileName"$date
        #将待上传的文件path写入一个列表文件willDoing
        echo $log_toupload_dir"xxxxx_click_log_$fileName"$date >> $log_toupload_dir"willDoing."$date
    fi

done

#找到列表文件willDoing
ls $log_toupload_dir | grep will |grep -v "_COPY_" | grep -v "_DONE_" | while read line
do
    #将待上传文件列表willDoing改名为willDoing_COPY_
    mv $log_toupload_dir$line $log_toupload_dir$line"_COPY_"
    #读列表文件willDoing_COPY_的内容（一个一个的待上传文件名）  ,此处的line 就是列表中的一个待上传文件的path
    cat $log_toupload_dir$line"_COPY_" |while read line
    do
        hadoop fs -put $line $hdfs_root_dir
    done    
    mv $log_toupload_dir$line"_COPY_"  $log_toupload_dir$line"_DONE_"
done

2.清洗数据

clean.job

代码语言：javascript复制
# clean.job
type=command
dependencies=upload
command=bash clean.sh

clean.sh

代码语言：javascript复制
#!/bin/bash

#set java env
export JAVA_HOME=/soft/jdk
export JRE_HOME=${JAVA_HOME}/jre 
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib 
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/soft/hadoop
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

#获取昨天的日期
day_01=`date -d'-1 day'  %Y-%m-%d`
#获取昨天的年份
syear=`date --date=$day_01  %Y`
#获取昨天的月份
smonth=`date --date=$day_01  %m`
#获取昨天的日份
sday=`date --date=$day_01  %d`

#日志在hdfs上的路径
log_hdfs_dir=/data/clickLog/$syear/$smonth/$sday
#mapreduce程序的入口路径
click_log_clean=clickLog.AccessLogDriver
#清洗后的数据路径
clean_dir=/cleaup/$syear/$smonth/$sday
#清洗后的数据路径不可存在（删除操作）
hadoop fs -rm -r -f $clean_dir
#运行mapreduce程序的jar文件（jar文件的位置；程序的入口路径；数据输入路径；数据输出路径）
hadoop jar /home/centos/hivedemo/mrclick.jar $click_log_clean $log_hdfs_dir $clean_dir

3.数据绑定到hive

hivesql.job

代码语言：javascript复制
# hivesql.job
type=command
dependencies=clean
command=bash hivesql.sh

hivesql.sh

代码语言：javascript复制
#!/bin/bash

#set java env
export JAVA_HOME=/soft/jdk
export JRE_HOME=${JAVA_HOME}/jre 
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib 
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/soft/hadoop
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

#set hive env
export HIVE_HOME=/soft/hive
export PATH=${HIVE_HOME}/bin:$PATH

#获取昨天的日期
day_01=`date -d'-1 day'  %Y-%m-%d`
#获取昨天的年份
syear=`date --date=$day_01  %Y`
#获取昨天的月份
smonth=`date --date=$day_01  %m`
#获取昨天的日份
sday=`date --date=$day_01  %d`

#清洗后的数据在hdfs上的路径
clean_dir=/cleaup/$syear/$smonth/$sday

#hive sql 导入数据到hive
HQL_origin="load data inpath '$clean_dir' into table mydb.accesslog"

#执行sql语句
hive -e  "$HQL_origin"

4.查询数据

ip.job

代码语言：javascript复制
# ip.job
type=command
dependencies=hivesqljob
command=bash ip.sh

ip.sh

代码语言：javascript复制
#!/bin/bash

#set java env
export JAVA_HOME=/soft/jdk
export JRE_HOME=${JAVA_HOME}/jre 
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib 
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/soft/hadoop
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

 #set hive env
export HIVE_HOME=/soft/hive
export PATH=${HIVE_HOME}/bin:$PATH

#hive sql 从一张表查询出数据放到结果集表中
HQL_origin="insert into  mydb.upflow  select ip,sum(upflow) as sum from mydb.accesslog group by ip order by sum desc "

#执行sql语句
hive -e  "$HQL_origin"

5.导出到mysql

mysql.job

代码语言：javascript复制
# mysql.job
type=command
dependencies=ipjob
command=bash mysql.sh

mysql.sh

代码语言：javascript复制
#!/bin/bash

#set java env
export JAVA_HOME=/soft/jdk
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

#set hadoop env
export HADOOP_HOME=/soft/hadoop
export PATH=${HADOOP_HOME}/bin:${HADOOP_HOME}/sbin:$PATH

#set hive env
export HIVE_HOME=/soft/hive
export PATH=${HIVE_HOME}/bin:$PATH

#set sqoop env
export SQOOP_HOME=/soft/sqoop
export PATH=${SQOOP_HOME}/bin:$PATH

#sqoop语句 从hive导出到mysql（导出到mysql的表中从hive的路径文件下）
sqoop export --connect jdbc:mysql://s201:3306/userdb --username sqoop --password sqoop --table upflow --export-dir /user/hive/warehouse/mydb.db/upflow --input-fields-terminated-by ','

五个job有依赖关系

bash bash指令 数据库 云数据库SQLServer sql

0 人点赞