13518219792
脚本功能：
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/profiledemeter_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/mongodbxuri-cpc-head-file
a b c d e f g h i j k l m n 0 p q r shost：
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另外有需要云服务器可以了解下创新互联cdcxhl.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
