shell_同步增量json数据到数据库

shell_同步增量json数据到数据库
数据流转：json–>>kafka–>>slipstream–>>数据库
代码：https://github.com/reader-sword/reader-sword-zhejiangdiankeyuan_append_jsonToDB

#!/bin/bash
#1.加载配置文件
parentPath="/mnt/software/work"
logPath="${parentPath}/log"
scriptPath="${parentPath}/script/every_day"
dataPath="${parentPath}/data/every_day"
source "${scriptPath}/global.conf"

#2.声明变量
currentDay=`date  +'%Y-%m-%d'`

#打印日志
function print_log(){
    log_date=`date  +'%Y-%m-%d %H:%M:%S'`
    echo "[${log_date} $1]" >>${logPath}/${currentDay}_log.log
}
#查看服务器是否正常
function connect_server(){
    ping -c 3 -W 5 47.99.48.155 &>/dev/null
    if [ "$?" == "0" ];then
        return 0
    else
        return 1
    fi
}
#错误文件处理

#初始化
function init(){
    #1.创建当天存放数据的目录
    mkdir -p ${dataPath}/${currentDay} >>/dev/null
    #2.测试数据源服务器是否正常
    connect_server
    if [ $? -ne 0 ]
    then
        print_log "server not health"
        exit 1
    fi
}
#主程序函数
function main(){
    init
    #获取开始抽数时间（默认从当天00:00:00 开始抽取）
    #if [ -z $start_time ];then
    #     min_time="${currentDay} 00:00:00"
    #else 
    #     min_time=$1
    #fi
    cmd_begin=`date +'%s'`
    min_time=$1
    next_5=`date -d "${min_time} 5 minute" +"%Y-%m-%d %H:%M:%S"`
    curl="curl -G \"http://$ip/query?pretty=true\" --data-urlencode \"db=scada\" --data-urlencode \"q=SELECT \\\"value\\\" FROM $tab where time>='${min_time}' and time < '${next_5}' tz('Asia/Shanghai') \""
    fileName01=`echo ${min_time} | awk -F " " '{print$1}'`
    fileName02=`echo ${min_time} | awk -F " " '{print$2}' | sed 's#:#-#g'`
    fileName="${fileName01}_${fileName02}.json"
    #print_log "开始执行 $curl"
    print_log "json 写入 $fileName"
    echo $curl | bash > ${dataPath}/${currentDay}/${fileName}
    print_log "数据文件 写入 ${fileName01}_${fileName02}.txt"
    grep -i "error" ${dataPath}/${currentDay}/${fileName} >> /dev/null
    if [ $? -eq 0 ];
       then
       echo "${dataPath}/${currentDay}/${fileName}" >> ${scriptPath}/jars/black_list.txt
    fi
    java -jar ${scriptPath}/jars/JsonParse.jar ${dataPath}/${currentDay}/${fileName} >>${logPath}/${currentDay}_log.log
    cmd_end=`date +'%s'`
    run_time=$((cmd_end-cmd_begin))
        sleep_time=$((300-run_time))
    print_log "本次运行时间：${run_time}s"
    /home/dsadm/TDH-Client/kafka/bin/kafka-console-producer.sh --broker-list zjdky01:9092,zjdky02:9092,zjdky03:9092 zjdky04:9092 --topic scada_battery_r3p3 < ${dataPath}/${currentDay}/${fileName01}_${fileName02}.txt
    sleep ${sleep_time}
    while [ 1=1 ]
    do  
    cmd_begin=`date +'%s'`
        currentDay=`date +'%Y-%m-%d'`
        mkdir -p ${dataPath}/${currentDay} >>/dev/null
        beginTime=${next_5} 
        next_5=`date -d "${next_5} 5 minute" +"%Y-%m-%d %H:%M:%S"`
        fileName01=`echo ${beginTime} | awk -F " " '{print$1}'`
        fileName02=`echo ${beginTime} | awk -F " " '{print$2}' | sed 's#:#-#g'`
        fileName="${fileName01}_${fileName02}.json"
        curl="curl -G \"http://$ip/query?pretty=true\" --data-urlencode \"db=scada\" --data-urlencode \"q=SELECT \\\"value\\\" FROM $tab where time>='${beginTime}' and time < '${next_5}' tz('Asia/Shanghai') \""
        #print_log "开始执行 $curl"
        print_log "json 写入 $fileName"
        echo $curl | bash > ${dataPath}/${currentDay}/${fileName}
    #如果请求错误
    grep -i "error" ${dataPath}/${currentDay}/${fileName} >> /dev/null
        if [ $? -eq 0 ];
           then
           echo "${dataPath}/${currentDay}/${fileName}" >> ${scriptPath}/jars/black_list.txt
        fi
        print_log "数据文件 写入 ${fileName01}_${fileName02}.txt"
        java -jar ${scriptPath}/jars/JsonParse.jar ${dataPath}/${currentDay}/${fileName} >>${logPath}/${currentDay}_log.log
    cmd_end=`date +'%s'`
    run_time=$((cmd_end-cmd_begin))
        sleep_time=$((300-run_time))
        print_log "本次运行时间：${run_time}s"
     /home/dsadm/TDH-Client/kafka/bin/kafka-console-producer.sh --broker-list zjdky01:9092,zjdky02:9092,zjdky03:9092 zjdky04:9092 --topic scada_battery_r3p3 < ${dataPath}/${currentDay}/${fileName01}_${fileName02}.txt
     sleep ${sleep_time}
    done
}
#程序开始
#main "${currentDay} $start_time"
main "${start_date}"