distcp 并行同步脚本

#!/bin/bash


if [ ! -n "$1" ] ;then
    echo "you have not input a file!"
else
    echo "the file you input is $1"
fi

dataline=$(cat $1 | awk -F "," '{print $1}')

# 源集群的schema
sourceCluster=hdfs://172.16.77.187:4007

# 目标集群的schema
targetCluster=hdfs://172.16.77.128:4007

#最大discp任务并行个数
runningnum=10

export HADOOP_USER_NAME=hadoop
num=0

for line in $dataline
do

   num=$[$num+1]
    echo "source Cluster files:"  ${sourceCluster}${line} 
    echo "target Cluster files:"  ${targetCluster}${line}
    echo "The number of discp tasks:  $num "
    task_name=`echo $line | awk -F"/" '{print $NF}'`
    echo "discp task name:  $task_name "   
    runningapp=$(ps -ef | grep distcp |wc -l)
    echo "The number of discp tasks running in the cluster:" $runningapp

    while [ $runningapp -ge $runningnum ]
    do
      runningapp=$(ps -ef | grep distcp |wc -l)
      echo "yarn正在执行的 distcp 任务数："$runningapp",sleep 5秒...."
      sleep 5
    done

    dir=`dirname $1`
    cd $dir
    fullPath=`pwd -P`
    mkdir ${fullPath}/logs   &>>/dev/null
    
    #可按照具体的情况调整
    #使用-update -delete 进行同步
    nohup hadoop distcp -Dmapreduce.job.name=$task_name  -Dmapreduce.map.memory.mb=4096 -pbugpt  -i -strategy dynamic -skipcrccheck -update -delete -numListstatusThreads 30  -bandwidth 100 -m 10 ${sourceCluster}${line}  ${targetCluster}${line} > ${fullPath}/logs/${task_name}.log 2>&1 &
    #使用 -overwrite  进行同步
    #nohup hadoop distcp -Dmapreduce.job.name=$task_name  -Dmapreduce.map.memory.mb=4096 -pbugpt -i -strategy dynamic -skipcrccheck -overwrite -numListstatusThreads 30  -bandwidth 100 -m 10 ${sourceCluster}${line}  ${targetCluster}${line} > ${fullPath}/logs/${task_name}.log 2>&1 &

sleep 0.1
done
wait