脚本功能：

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=$1case $1incsv)  FILEFORMAT=",";;tsv)   FILEFORMAT="\t";;*) echo "unknow profile -o $1"; usage;;esac
}
while getopts ':hf:g:o:k:d:t:a:p:m:c:' OPTION
docase $OPTIONinh) 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 ]; thenecho "[`date +"%Y-%m-%d %H:%M:%S,%s"`][$SCRIPTFILE, $$] ERROR OCCURS! - $1" | tee -a $ERRORFILEexit 1;fi
}
function check_ready() {tmp_done_file=`printf "$reportDoneFile" "$TABLE" "$1"`while [ "$isok" = "false" ]; dorsync  --list-only ${tmp_done_file}if [ $? -eq 0 ]; thenisok="true";break;fiif [ "$isok" = "false" ]; thensleep 300fitime_now=`date  +%s`if [ `expr ${time_now} - ${time_start}` -ge $max_interval ]; thenreturn 255;fidonereturn 0;
}
#从数据库列表里选择主库
function selectMongoMaster(){tmp="TARGET_MONGO_HOST_LIST_0$1"TMP_HOST=${!tmp}echo $TMP_HOST#replica setfor DUBHE_MONGO_HOST in $TMP_HOST; doif [ $? -eq 0 ] ; thenbreak;fidone# 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"$SPLITDBNUMecho "split tb num"$SPLITTBNUMecho "Start to split file: "$SPLITFILEawk '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 $SPILTTMPDIRecho "Split file successfully : "$SPLITFILE
}
#导入
function import() {#importDatalocal iter=1;while [ $iter -le $SPLITDBNUM ]; dothread_import $iter &iter=`expr $iter + 1`done#wait for child-threadswait;
}
#导入子线程
function thread_import() {local num=1;targetFileName=$IMPORTTBNAME"_0"$1"0"$numtargetFile=$SPILTTMPDIR/$IMPORTTBNAME"_0"$1"0"$numtargetDB=$IMPORTDBNAME"_0"$1targetCollection=$IMPORTTBNAME"_0"$1"0"$numif [ ! -f $targetFile ]; thenlogInfo "spilt file does not exits : " $targetFilenum=`expr $num + 1`continuefiuser="TARGET_MONGO_USER_0"$1TMP_USER=${!user}password="TARGET_MONGO_PWD_0"$1TMP_PASSWORD=${!password}#选择master   selectMongoMaster $1;#clean dirty dataif [ $CLEAN -gt 0  ]; thenlogInfo "$qdate $targetDB.$targetCollection cleaning up dirty data in mongodb"clean_dirty_datacheckError "whether error occurs during cleaning dirty data from mongodb"fi#import dataimport2mongo $1 $targetFile  $targetDB  $targetCollection#record done filestatusfile="$STATUS_LOG_HOME/$targetFileName.done.`date -d $qdate +"%Y-%m-%d"`"touch $statusfilenum=`expr $num + 1`donelogInfo "thread $1 ends"
}
#把指定的文件导到指定的库指定的表,并建立索引，mongodb自身会判断索引是否存在
#不存在的情况下才创建新索引
function import2mongo(){if [ "$FIELDFILE" != "" ]; thenMONGO_FIELD_FILE=$FIELDFILEelseMONGO_FIELD_FILE=$FULLPATH/../conf/${IMPORTTBNAME}-head-filefiDATAFILE=$2if [ ! -f $DATAFILE ]; thenlogInfo "mongodb [${DB}.${COLL}] imported 0 objects"return 0fiTMPLOGFILE=$INFO_LOG_HOME/$DB.$COLL.tmp.logtmp=$?if [ "$tmp" != "0" ]; thenreturn $tmpfi#data check_mongo_count=`tail $TMPLOGFILE | grep imported`_mongo_count=`expr 0$_mongo_count + 0`#start to ensure indexensureIndexlogInfo "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\"}}"filogInfo "waiting for the clean task.."echo  $_mongo_conditiontmp=$?if [ "$tmp" != "0" ]; thenreturn $tmpfisleep 5slogInfo "dirty data cleaned: "$targetDB  $targetCollection  $dirtyCountecho "dirty data cleaned: "$targetDB  $targetCollection  $dirtyCountreturn $tmp
}
#parameter check
function checkParams() {if [ 1 -ne $CLEAN -a 2 -ne $CLEAN ]; thenlogInfo "-g the parameter clean is not in [1, 2] : "$CLEANreturn 1;fiif [  $FILEFORMAT != "," -a  $FILEFORMAT != "\t"  ]; thenlogInfo "-o the parameter file format  is not in [csv, tsv] : "$FILEFORMATreturn 1;fiif [ $SPLITKEY -lt 1 ]; thenlogInfo "-k split key must not be less  than 1 : "$SPLITKEYreturn 1;fiif [ $SPLITDBNUM -lt 1 ]; thenlogInfo "-d database number must not  be less  than 1 : "$SPLITDBNUMreturn 1;fiif [ $SPLITTBNUM -lt 1 ]; thenlogInfo "-t collection number must not  be less  than 1 : "$SPLITTBNUMreturn 1;fiif [ ! -f  $FIELDFILE ];  thenlogInfo "-a field file is not a common file or not exits : "$FIELDFILEreturn 1;fiif [ "" = $IMPORTDBNAME ] ; thenlogInfo "-m import  database name is empty  : "$IMPORTDBNAMEreturn 1;fiif [ "" = $IMPORTTBNAME ] ; thenlogInfo "-m import  table name is empty  : "$IMPORTTBNAMEreturn 1;fi
}
#主函数
function main() {set +xecho "check split file and profile: " $SPLITFILE   $PROFILEif [ ! -f  $SPLITFILE ];  thenecho  "-f split file is not a common file or not exits : "$SPLITFILEreturn 1;fiif [ ! -f  $PROFILE ];  thenecho  "-p profile file is not a common file or not exits : "$PROFILEreturn 1;fisource $PROFILEqdate=`date +"%Y-%m-%d"`last_day=`date -d "-1day" +"%Y-%m-%d"`BASEFILENAME=$(basename $SPLITFILE)echo "base split file name is : "$BASEFILENAMEif [ ! -d $LOG_HOME ] ; thenlogInfo  " log home  is not a common directory or not exits : "$LOG_HOMEreturn 1;fiLOGFILE=$INFO_LOG_HOME/$BASEFILENAME.$qdate.logif [ -f $LOGFILE ]; thenmv $LOGFILE $LOGFILE.$last_dayfitouch $LOGFILEERRORFILE=$ERROR_LOG_HOME/$BASEFILENAME.error.logif [ -f $ERRORFILE ]; thenmv $ERRORFILE $ERRORFILE.$last_dayfitouch $ERRORFILE#空行echoechologInfo "start to check parameters!"checkParamscheckError "whether error occurs during check parameters : $SPLITFILE"#空行echoechologInfo "start to split file: "$SPLITFILEif [ ! -d $DATA_SPLIT_HOME ] ; thenlogInfo  " data split home  is not a common directory or not exits : "$DATA_SPLIT_HOMEreturn 1;fiSPILTTMPDIR=$DATA_SPLIT_HOME/$BASEFILENAMEecho "split temple directory : "$SPILTTMPDIRif [ -d ${SPILTTMPDIR} ]; thenrm -rf ${SPILTTMPDIR}fimkdir -p ${SPILTTMPDIR}splitcheckError "whether error occurs during split data : $SPLITFILE"logInfo "split data completely : $SPLITFILE"statusfile=$STATUS_LOG_HOME/$BASEFILENAME".split.done."$qdatetouch  ${statusfile}#空行echoechologInfo "start to import split  file to mongodb"importlogInfo "import data completely : $SPLITFILE"statusfile=$STATUS_LOG_HOME/$BASEFILENAME".import.done."$qdatetouch  ${statusfile}#空行echoecho#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 ]; thenmkdir -p $INFO_LOG_HOME
fi
if [ ! -d $STATUS_LOG_HOME ]; thenmkdir -p $STATUS_LOG_HOME
fi
if [ ! -d $DATA_HOME ]; thenmkdir -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