mongodb拆库分表脚本

客官°小女子只卖身不卖艺 2022-02-26 02:35 271阅读 0赞

脚本功能：

1. 将指定的报告文件按照指定的字段、切库切表策略切分

2. 将切分后的文件并发导入到对应的Mongodb中

3. 生成日志文件和done标识文件

使用手册：

-h    打印帮助信息,并退出";

-f     需要切分的数据文件";

-g    清理昨日或历史全部数据： 1 昨日数据  2 历史全部数据";

-k     拆分字段在文件中列数，从1开始";

-o    需要切分的数据文件格式 tsv或csv ";

-d    切分的库数目";

-t     切分的表数目";

-m   切分后，需要入库的mongodb未拆分库名，比如拆分前cpc, 拆分后cpc\_01";

-c    切分后，需要入库的mongodb未拆分库名，比如拆分前cpc, 拆分后cpc\_0102";

-a    入库fieldFile";

-p    配置文件",

使用步骤：

1. 在配置文件中设置日志、切割后数据临时路径$LOG\_HOME 和 $DATA\_SPLIT\_HOME目录，如果不存在，则手动创建；

在配置文件中设置目标Mongodb参数信息，用来作为导入数据的目标库；

在配置文件中设置Mongodb程序的主目录$MONGO;

2. 按照具体的参数意义，仿照下面的格式执行脚本：

举例：./mongo-split-importer.sh -f /data/shell/test.ata -g 1 -o tsv -k 3 -d 3 -t 3 -m idea -c idea -p ../conf/demeter\_conf\_qa.sh -a ../conf/idea-head-file

-f 切分目标文件   -o 文件格式 tsv    -k 切割字段，第三个  -d 切割成3个库 -t 每个库3个表

-m 导入的mongodb未拆分名称idea -c 导入的mongodb未拆分表名idea -p 环境配置文件 -a 导入目标表的fieldFile文件 -g 清理昨日数据

mongo-split-importer.sh执行脚本：

#!/bin/bash
    SPLITFILE="" #目标切割文件
    FILEFORMAT="" # 目标切割文件格式 , \t
    FILEFORMATNAME="" #切割目标文件格式名称 csv tsv
    SPLITKEY=1
    SPLITDBNUM="" #目标切割库数目
    SPLITTBNUM="" #目标切割表数目
    IMPORTDBNAME="" # 目标入库未分割库名
    IMPORTTBNAME="" #目标入库未切割表名
    PROFILE="" #配置文件
    FIELDFILE="" #入库fieldFile
    CLEAN=0  #清理数据， 0：默认不清理， 1 ： 清理昨日的数据    2： 清理所有以前的数据
    SPILTTMPDIR="" #目标切割文件存放临时目录
    FULLPATH=$(cd `dirname $0`;pwd -P)
    SCRIPTFILE=`basename $0`
    TOTLE_RECORD_NUM=0 #文件切割前的记录条目
    SUBFILE_RECORD_NUM=0 #切割后所有文件汇总的记录条目
    _mongo_count="-1"
    #------------------------------------------------函数---------------------------------------------------------------
    function usage(){
            echo "$SCRIPTFILE - 分库分表后将数据导数据到mongodb"
            echo "SYNOPSIS"
            echo "OPTIONS" 
            echo "  -h    打印帮助信息,并退出";
            echo "  -f     需要切分的数据文件";
            echo "  -g    是否清理历史数据，默认不清理   1：清理昨日数据  2：清理以前所有数据";
            echo "  -k     拆分字段在文件中列数，从1开始";
            echo "  -o    需要切分的数据文件格式 tsv或csv ";
            echo "  -d    切分的库数目";
            echo "  -t     切分的表数目";
            echo "  -m   切分后，需要入库的mongodb未拆分库名，比如拆分前cpc, 拆分后cpc_01";
            echo "  -c    切分后，需要入库的mongodb未拆分库名，比如拆分前cpc, 拆分后cpc_0102";
            echo "  -a    入库fieldFile";
            echo "  -p    配置文件,绝对或相对路径文件",
            exit
    }
    function setFileFormat(){
            FILEFORMATNAME=$1
            case $1
            in
                    csv)  FILEFORMAT=",";;
                    tsv)   FILEFORMAT="\t";;
                    *) echo "unknow profile -o $1"; usage;;
            esac
    }
    while getopts ':hf:g:o:k:d:t:a:p:m:c:' OPTION
    do
            case $OPTION
            in
                    h) usage;;
                    f) SPLITFILE=$OPTARG;;
                    g)CLEAN=$OPTARG;;
                    o) setFileFormat $OPTARG;;
                    k) SPLITKEY=$OPTARG;;
                    d) SPLITDBNUM=$OPTARG;;
                    t) SPLITTBNUM=$OPTARG;;
                    a) FIELDFILE=$OPTARG;;
                    p) PROFILE=$OPTARG;;
                    m) IMPORTDBNAME=$OPTARG;;
                    c) IMPORTTBNAME=$OPTARG;;
                    :) echo "选项 \"-$OPTARG\" 后面缺少对应值, 将使用默认值";;
                    \?)echo " 错误的选项 -$OPTARG, 将退出"; usage;;
            esac
    done
    #记录日志信息
    function logInfo(){
      echo "[`date +"%Y-%m-%d %H:%M:%S"`] $@ " | tee -a $LOGFILE
    }
    function checkError(){
            if [ $? -ne 0 ]; then
                    echo "[`date +"%Y-%m-%d %H:%M:%S,%s"`][$SCRIPTFILE, $$] ERROR OCCURS! - $1" | tee -a $ERRORFILE
                    exit 1;
            fi
    }
    function check_ready() {
            tmp_done_file=`printf "$reportDoneFile" "$TABLE" "$1"`
            while [ "$isok" = "false" ]; do
                    rsync  --list-only ${tmp_done_file}
                    if [ $? -eq 0 ]; then
                            isok="true";
                            break;
                    fi
                    if [ "$isok" = "false" ]; then
                            sleep 300
                    fi
                    time_now=`date  +%s`
                    if [ `expr ${time_now} - ${time_start}` -ge $max_interval ]; then
                            return 255;
                    fi
            done
            return 0;
    }
    #从数据库列表里选择主库
    function selectMongoMaster(){
                    tmp="TARGET_MONGO_HOST_LIST_0$1"
                    TMP_HOST=${!tmp}
                    echo $TMP_HOST
                    #replica set
                    for DUBHE_MONGO_HOST in $TMP_HOST; do
                            if [ $? -eq 0 ] ; then
                                    break;
                            fi
                    done
                    # single server
                    #for DUBHE_MONGO_HOST in $TMP_HOST; do
                            #TARGET_MONGO_HOST=$DUBHE_MONGO_HOST
                            #echo $TARGET_MONGO_HOST
                    #done
    }
    #切割
    function split() {
            logInfo "spilt data file"
            echo "split db num"$SPLITDBNUM
            echo "split tb num"$SPLITTBNUM
            echo "Start to split file: "$SPLITFILE
            awk '
            BEGIN {
                    FS="'${FILEFORMAT}'";
            }
            ARGIND==1{
                            #分库分表
                    DBN=$'${SPLITKEY}' % '${SPLITDBNUM}' + 1;
                    TBN=int($'${SPLITKEY}' / '${SPLITDBNUM}')
                            TBN=TBN % '${SPLITTBNUM}' + 1;
                    DBN="0"DBN;
                    TBN="0"TBN;
                            print $0 > "'${SPILTTMPDIR}'""/""'${IMPORTTBNAME}'""_"DBN""TBN
            }
            END {
            }
            ' ${SPLITFILE};
            ls $SPILTTMPDIR
            echo "Split file successfully : "$SPLITFILE
    }
    #导入
    function import() {
            #importData
            local iter=1;
        while [ $iter -le $SPLITDBNUM ]; do
                    thread_import $iter &
                    iter=`expr $iter + 1`
            done
            #wait for child-threads
            wait;
    }
    #导入子线程
    function thread_import() {
            local num=1;
                                    targetFileName=$IMPORTTBNAME"_0"$1"0"$num
                                    targetFile=$SPILTTMPDIR/$IMPORTTBNAME"_0"$1"0"$num
                                    targetDB=$IMPORTDBNAME"_0"$1
                                    targetCollection=$IMPORTTBNAME"_0"$1"0"$num
                                    if [ ! -f $targetFile ]; then
                                            logInfo "spilt file does not exits : " $targetFile
                                            num=`expr $num + 1`
                                            continue
                                    fi
                                    user="TARGET_MONGO_USER_0"$1
                                    TMP_USER=${!user}
                                    password="TARGET_MONGO_PWD_0"$1
                                    TMP_PASSWORD=${!password}
                                     #选择master   
                                    selectMongoMaster $1;
                                    #clean dirty data
                                if [ $CLEAN -gt 0  ]; then
                                            logInfo "$qdate $targetDB.$targetCollection cleaning up dirty data in mongodb"
                                            clean_dirty_data
                                            checkError "whether error occurs during cleaning dirty data from mongodb"
                                    fi
                                    #import data
                                    import2mongo $1 $targetFile  $targetDB  $targetCollection
                    #record done file
                    statusfile="$STATUS_LOG_HOME/$targetFileName.done.`date -d $qdate +"%Y-%m-%d"`"
                    touch $statusfile
                    num=`expr $num + 1`
            done
            logInfo "thread $1 ends"
    }
    #把指定的文件导到指定的库指定的表,并建立索引，mongodb自身会判断索引是否存在
    #不存在的情况下才创建新索引
    function import2mongo(){
        if [ "$FIELDFILE" != "" ]; then
                    MONGO_FIELD_FILE=$FIELDFILE
            else
                    MONGO_FIELD_FILE=$FULLPATH/../conf/${IMPORTTBNAME}-head-file
            fi
            DATAFILE=$2
        if [ ! -f $DATAFILE ]; then
            logInfo "mongodb [${DB}.${COLL}] imported 0 objects"
            return 0
        fi
        TMPLOGFILE=$INFO_LOG_HOME/$DB.$COLL.tmp.log
            tmp=$?
            if [ "$tmp" != "0" ]; then
                    return $tmp
            fi
            #data check
            _mongo_count=`tail $TMPLOGFILE | grep imported`
            _mongo_count=`expr 0$_mongo_count + 0`
            #start to ensure index
            ensureIndex
            logInfo "mongodb [${DB}.${COLL}] imported $_mongo_count objects"
            return $tmp
    }
    function ensureIndex(){
    }
    #垃圾数据清理
    function clean_dirty_data(){
            day=`date -d ${1:-' -1day'} +"%y%m%d"`
            if [ $CLEAN -eq 1  ]; then
                    _mongo_condition="{\"_id\":{\"\$gte\":\"${day}_0\",\"\$lte\":\"${day}_9\"}}"
            else
                    _mongo_condition="{\"_id\":{\"\$lte\":\"${day}_9\"}}"
            fi
            logInfo "waiting for the clean task.."
            echo  $_mongo_condition
            tmp=$?
            if [ "$tmp" != "0" ]; then
                    return $tmp
            fi
            sleep 5s
            logInfo "dirty data cleaned: "$targetDB  $targetCollection  $dirtyCount
            echo "dirty data cleaned: "$targetDB  $targetCollection  $dirtyCount
            return $tmp
    }
    #parameter check
    function checkParams() {
            if [ 1 -ne $CLEAN -a 2 -ne $CLEAN ]; then
                    logInfo "-g the parameter clean is not in [1, 2] : "$CLEAN
                    return 1;
            fi
            if [  $FILEFORMAT != "," -a  $FILEFORMAT != "\t"  ]; then
                    logInfo "-o the parameter file format  is not in [csv, tsv] : "$FILEFORMAT
                    return 1;
            fi
            if [ $SPLITKEY -lt 1 ]; then
                            logInfo "-k split key must not be less  than 1 : "$SPLITKEY
                            return 1;
            fi
            if [ $SPLITDBNUM -lt 1 ]; then
                            logInfo "-d database number must not  be less  than 1 : "$SPLITDBNUM
                            return 1;
            fi
            if [ $SPLITTBNUM -lt 1 ]; then
                            logInfo "-t collection number must not  be less  than 1 : "$SPLITTBNUM
                            return 1;
            fi
            if [ ! -f  $FIELDFILE ];  then
                    logInfo "-a field file is not a common file or not exits : "$FIELDFILE
                    return 1;
            fi
            if [ "" = $IMPORTDBNAME ] ; then
                            logInfo "-m import  database name is empty  : "$IMPORTDBNAME
                            return 1;
            fi
            if [ "" = $IMPORTTBNAME ] ; then
                            logInfo "-m import  table name is empty  : "$IMPORTTBNAME
                            return 1;
            fi
    }
    #主函数
    function main() {
                    set +x
                    echo "check split file and profile: " $SPLITFILE   $PROFILE
                    if [ ! -f  $SPLITFILE ];  then
                            echo  "-f split file is not a common file or not exits : "$SPLITFILE
                            return 1;
                    fi
                    if [ ! -f  $PROFILE ];  then
                            echo  "-p profile file is not a common file or not exits : "$PROFILE
                            return 1;
                    fi
                    source $PROFILE
                    qdate=`date +"%Y-%m-%d"`
                    last_day=`date -d "-1day" +"%Y-%m-%d"`
                    BASEFILENAME=$(basename $SPLITFILE)
                    echo "base split file name is : "$BASEFILENAME
                    if [ ! -d $LOG_HOME ] ; then
                            logInfo  " log home  is not a common directory or not exits : "$LOG_HOME
                            return 1;
                    fi
                    LOGFILE=$INFO_LOG_HOME/$BASEFILENAME.$qdate.log
                    if [ -f $LOGFILE ]; then
                            mv $LOGFILE $LOGFILE.$last_day
                    fi
                    touch $LOGFILE
                    ERRORFILE=$ERROR_LOG_HOME/$BASEFILENAME.error.log
                    if [ -f $ERRORFILE ]; then
                            mv $ERRORFILE $ERRORFILE.$last_day
                    fi
                    touch $ERRORFILE
                    #空行
                    echo
                    echo
                    logInfo "start to check parameters!"
                    checkParams
                    checkError "whether error occurs during check parameters : $SPLITFILE"
                    #空行
                    echo
                    echo
                    logInfo "start to split file: "$SPLITFILE
                    if [ ! -d $DATA_SPLIT_HOME ] ; then
                            logInfo  " data split home  is not a common directory or not exits : "$DATA_SPLIT_HOME
                            return 1;
                    fi
                    SPILTTMPDIR=$DATA_SPLIT_HOME/$BASEFILENAME
                    echo "split temple directory : "$SPILTTMPDIR
                    if [ -d ${SPILTTMPDIR} ]; then
                            rm -rf ${SPILTTMPDIR}
                    fi
                    mkdir -p ${SPILTTMPDIR}
                    split
                    checkError "whether error occurs during split data : $SPLITFILE"
                    logInfo "split data completely : $SPLITFILE"
                    statusfile=$STATUS_LOG_HOME/$BASEFILENAME".split.done."$qdate
                    touch  ${statusfile}
                    #空行
                    echo
                    echo
                    logInfo "start to import split  file to mongodb"
                    import
                    logInfo "import data completely : $SPLITFILE"
                    statusfile=$STATUS_LOG_HOME/$BASEFILENAME".import.done."$qdate
                    touch  ${statusfile}
                    #空行
                    echo
                    echo
                    #remove temple directory
            #       if [ -d ${SPILTTMPDIR} ]; then
            #               rm -rf ${SPILTTMPDIR}
            #       fi
    }
    #-------------------------------------------------入口----------------------------------------------------------------
    source /etc/profile

demeter\_conf\_cpc\_qa.sh 脚本：

#!/bin/bash
    source /etc/profile
    #logger path
    INFO_LOG_HOME="${LOG_HOME}/info"
    STATUS_LOG_HOME="${LOG_HOME}/status"
    if [ ! -d $ERROR_LOG_HOME ]; then
    if [ ! -d $INFO_LOG_HOME ]; then
            mkdir -p $INFO_LOG_HOME
    fi
    if [ ! -d $STATUS_LOG_HOME ]; then
            mkdir -p $STATUS_LOG_HOME
    fi
    if [ ! -d $DATA_HOME ]; then
            mkdir -p $DATA_HOME
    fi
    #data path for source  and target data path
    DATA_SPLIT_HOME=/data/demeter/sdata
    #import target mongodbs
    TARGET_MONGO_PORT_01=XXX
    TARGET_MONGO_USER_01=XXX
    TARGET_MONGO_PWD_01=XXX
    TARGET_MONGO_HOST_LIST_01="test01.mongodb01:$TARGET_MONGO_PORT_01 test01.mongodb02:$TARGET_MONGO_PORT_01 test01.mongodb03:$
    TARGET_MONGO_PORT_01"
    TARGET_MONGO_PORT_02=XXX
    TARGET_MONGO_USER_02=XXX
    TARGET_MONGO_PWD_02=XXX
    TARGET_MONGO_HOST_LIST_02="testt02.mongodb01:$TARGET_MONGO_PORT_02 test02.mongodb02:$TARGET_MONGO_PORT_02 test02.mongodb03:$
    TARGET_MONGO_PORT_02"
    TARGET_MONGO_PORT_03=XXX
    TARGET_MONGO_USER_03=XXX
    TARGET_MONGO_PWD_03=XXX
    TARGET_MONGO_HOST_LIST_03="test03.mongodb01:$TARGET_MONGO_PORT_03 test03.mongodb02:$TARGET_MONGO_PORT_03 test03.mongodb03:$
    TARGET_MONGO_PORT_03"
    #mongodb utils
    MONGO=/opt/mongodb

xuri-cpc-head-file

a
    b
    c
    d
    e
    f
    g
    h
    i
    j
    k
    l
    m
    n
    0
    p
    q
    r
    s

host：

XX.XX.XX.XX  test01.mongodb01
    XX.XX.XX.XX  test01.mongodb02
    XX.XX.XX.XX  testt01.mongodb03
    XX.XX.XX.XX  test02.mongodb01
    XX.XX.XX.XX  test02.mongodb02
    XX.XX.XX.XX  test02.mongodb03
    XX.XX.XX.XX  test03.mongodb01
    XX.XX.XX.XX  test03.mongodb02
    XX.XX.XX.XX  test03.mongodb03

转载于：[http://www.voidcn.com/article/p-mvxhfhwr-zo.html][http_www.voidcn.com_article_p-mvxhfhwr-zo.html]

[http_www.voidcn.com_article_p-mvxhfhwr-zo.html]: http://www.voidcn.com/article/p-mvxhfhwr-zo.html