脚本功能：
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
