DataX script task development record

# job Template ：/home/work/datax/job/fn/example/fn_t_bank_return_days
{
	"job": {
		"content": [{
			"reader": {
				"name": "mysqlreader",
				"parameter": {
					"columnMeta": [{
						"value": "dt",
						"label": "DATE_FORMAT(f_create_time,'%Y-%m-%d')",
						"type": "custom"
					}, {
						"value": "F_seq",
						"label": "F_seq",
						"type": null
					}, {
						"value": "F_enabled",
						"label": "F_enabled",
						"type": null
					}, {
						"value": "F_bank_id",
						"label": "F_bank_id",
						"type": null
					}, {
						"value": "F_bill_no",
						"label": "F_bill_no",
						"type": null
					}, {
						"value": "F_type",
						"label": "F_type",
						"type": null
					}, {
						"value": "F_currency",
						"label": "F_currency",
						"type": null
					}, {
						"value": "F_amount",
						"label": "F_amount",
						"type": null
					}, {
						"value": "F_return_id",
						"label": "F_return_id",
						"type": null
					}, {
						"value": "F_account_type",
						"label": "F_account_type",
						"type": null
					}, {
						"value": "F_payer_account",
						"label": "F_payer_account",
						"type": null
					}, {
						"value": "F_payer_name",
						"label": "F_payer_name",
						"type": null
					}, {
						"value": "F_date",
						"label": "REPLACE(REPLACE(F_date, 'null', '1970-01-01'), '0000-00-00', '1970-01-01')",
						"type": null
					}, {
						"value": "F_time",
						"label": "F_time",
						"type": null
					}, {
						"value": "F_rdate",
						"label": "REPLACE(REPLACE(F_rdate, 'null', '1970-01-01'), '0000-00-00', '1970-01-01')",
						"type": null
					}, {
						"value": "F_error",
						"label": "F_error",
						"type": null
					}, {
						"value": "F_proc",
						"label": "F_proc",
						"type": null
					}, {
						"value": "F_proc_time",
						"label": "REPLACE(REPLACE(F_proc_time, 'null', '1970-01-01 00:00:00'), '0000-00-00 00:00:00', '1970-01-01 00:00:00')",
						"type": null
					}, {
						"value": "F_proc_uid",
						"label": "F_proc_uid",
						"type": null
					}, {
						"value": "F_proc_ip",
						"label": "F_proc_ip",
						"type": null
					}, {
						"value": "F_create_time",
						"label": "F_create_time",
						"type": null
					}, {
						"value": "F_modify_time",
						"label": "F_modify_time",
						"type": null
					}, {
						"value": "F_comment",
						"label": "F_comment",
						"type": null
					}, {
						"value": "F_file_id",
						"label": "F_file_id",
						"type": null
					}, {
						"value": "F_notify_time",
						"label": "REPLACE(REPLACE(F_notify_time, 'null', '1970-01-01 00:00:00'), '0000-00-00 00:00:00', '1970-01-01 00:00:00')",
						"type": null
					}, {
						"value": "F_notify_comment",
						"label": "F_notify_comment",
						"type": null
					}, {
						"value": "F_notify_state",
						"label": "ifnull(F_notify_state,0)",
						"type": null
					}, {
						"value": "F_bank_element",
						"label": "F_bank_element",
						"type": null
					}, {
						"value": "F_batch_id",
						"label": "F_batch_id",
						"type": null
					}, {
						"value": "F_match_refund",
						"label": "F_match_refund",
						"type": null
					}, {
						"value": "F_match_refund_review",
						"label": "F_match_refund_review",
						"type": null
					}, {
						"value": "F_match_refund_bounce",
						"label": "F_match_refund_bounce",
						"type": null
					}, {
						"value": "F_match_refund_bounce_review",
						"label": "F_match_refund_bounce_review",
						"type": null
					}, {
						"value": "F_drcr_status",
						"label": "F_drcr_status",
						"type": null
					}, {
						"value": "F_fee_amount",
						"label": "F_fee_amount",
						"type": null
					}, {
						"value": "F_batch_part",
						"label": "F_batch_part",
						"type": null
					}, {
						"value": "F_fee_amount_theory",
						"label": "F_fee_amount_theory",
						"type": null
					}, {
						"value": "F_fee_ext",
						"label": "F_fee_ext",
						"type": null
					}],
					"where": "",
					"connectionPattern": {
						"urls": ["jdbc:mysql://*.*.*.*:6012"],
						"database": "dyp_db_finance",
						"tablename": "t_bank_return",
						"arguments": null
					},
					"connection": [JDBCSTR],
					"mandatoryEncoding": "GBK",
					"username": "datax_r",
					"password": "******",
					"column": ["DATE_FORMAT(f_create_time,'%Y-%m-%d')", "F_seq", "F_enabled", "F_bank_id", "F_bill_no", "F_type", "F_currency", "F_amount", "F_return_id", "F_account_type", "F_payer_account", "F_payer_name", "REPLACE(REPLACE(F_date, 'null', '1970-01-01'), '0000-00-00', '1970-01-01')", "F_time", "REPLACE(REPLACE(F_rdate, 'null', '1970-01-01'), '0000-00-00', '1970-01-01')", "F_error", "F_proc", "REPLACE(REPLACE(F_proc_time, 'null', '1970-01-01 00:00:00'), '0000-00-00 00:00:00', '1970-01-01 00:00:00')", "F_proc_uid", "F_proc_ip", "F_create_time", "F_modify_time", "F_comment", "F_file_id", "REPLACE(REPLACE(F_notify_time, 'null', '1970-01-01 00:00:00'), '0000-00-00 00:00:00', '1970-01-01 00:00:00')", "F_notify_comment", "ifnull(F_notify_state,0)", "F_bank_element", "F_batch_id", "F_match_refund", "F_match_refund_review", "F_match_refund_bounce", "F_match_refund_bounce_review", "F_drcr_status", "F_fee_amount", "F_batch_part", "F_fee_amount_theory", "F_fee_ext"],
					"jdbc_encoding": "CP1252"
				}
			},
			"writer": {
				"name": "streamwriter",
				"parameter": {
					"cryptMode": "no",
					"roleType": "",
					"fieldDelimiter": "\t",
					"fileName": "r_ext_fn_t_bank_return_old.dat",
					"groupname": "dyp_gp_pay",
					"lineDelimiter": "\n",
					"path": "/home/work/pay-datax-admin/ext/"
				}
			}
		}],
		"setting": {
			"speed": {
				"channel": 5,
				"byte": 20971520
			}
		}
	}
}

#!/bin/bash
# 0 */1 * * * (cd /home/work/daiyuanpei; bash t_bank_return_days.sh 9 > /home/work/daiyuanpei/t_bank_return_days.out 2>&1 &)
. ~/.bashrc
dateEnd=`date +"%Y%m%d"`
dateBegin=`date +%Y%m%d -d"-30 day $dateEnd"`
if [ $# == 1 ]; then
    if [ ! -n "$(echo $1| sed -n "/^[0-9]\+$/p")" ]; then
        echo "FATAL: The parameter is not a number."
        exit -1
    fi
    if [ $1 -gt 100 ]; then
        echo "FATAL: Parameter n cannot be greater than 100."
        exit -1
    fi
    dateBegin=`date +%Y%m%d -d"-$1 day $dateEnd"`
fi
delDateBegin=`date +%Y-%m-%d -d"+1 day $dateBegin"`
echo $dateBegin $dateEnd $delDateBegin
schema="fn"
tbname="t_bank_return"
example="/home/work/datax/job/fn/example/${schema}_${tbname}_days"
tempdir="/home/work/datax/job/fn"
basepath=$(cd `dirname $0`; pwd)
cd $basepath
jsonfile="${tempdir}/${schema}_${tbname}_days"
if [[ "$dateBegin" > "$dateEnd" ]]; then
    echo "FATAL: dateEnd must be greater than or equal to dateBegin. dateBegin: $dateBegin, dateEnd: $dateEnd"
    exit -1
fi
jdbcTpl="{\"jdbcUrl\":[\"jdbc:mysql://*.*.*.*:6012/dyp_db_finance_DBDATE\"],\"table\":[\"t_bank_return\"]}"
jdbcStr=""
while [[ "$dateBegin" < "$dateEnd" ]] || [[ "$dateBegin" == "$dateEnd" ]]; do
    dayStr="${jdbcTpl/DBDATE/$dateBegin}"
    jdbcStr="$jdbcStr,$dayStr"
    dateBegin=`date +%Y%m%d -d"+1 day $dateBegin"`
done
echo "while done"
#  Remove the first character 
jdbcStr=`echo $jdbcStr | awk '{print substr($1,2)}'`
#  below sed  So you need to escape slashes and double quotes 
#  Single slash replace once   Double slashes replace all matches 
jdbcStr="${jdbcStr//\"/\\\"}"
jdbcStr="${jdbcStr//\//\/}"
echo $jdbcStr
sed "s/JDBCSTR/${jdbcStr}/g;s/CURRENTDATE/${dateEnd}/g" ${example} > ${jsonfile}
echo "${jsonfile} done"
~/datax/bin/start_datax.sh ${jsonfile} && bash /home/work/datax/sh-load/t_bank_return_days.sh ${schema} ${tbname} ${delDateBegin}
echo "*****************************"
echo "*****************************"
echo "*****************************"

#!/bin/bash
# /home/work/datax/sh-load/t_bank_return_days.sh
. ~/.bashrc
if [ $# -lt 3 ]
then
    echo "./loaddata schema tablename delDateBegin"
    exit
fi
export LANG="zh_CN.UTF-8"
export LC_ALL="zh_CN.UTF-8"
schema=$1
tablename=$2
delDateBegin=$3
data_path=" /home/work/pay-datax-admin/ext/r_ext_${schema}_${tablename}_old.dat"
basepath=$(cd `dirname $0`; pwd)
cd ${basepath}
if [ ! -e ${data_path} ];then
 echo "no exist filepath:${data_path}"
 exit
fi

#DEL_SQL="delete from ${schema}.${tablename} where dt>='${delDateBegin}';"
LOAD_SQL="delete from ${schema}.${tablename} where dt>='${delDateBegin}';insert into ${schema}.${tablename} select * from pay_data_ext.r_ext_${schema}_${tablename}_old;"
#echo ${DEL_SQL}
echo ${LOAD_SQL}
tempfile="${basepath}/temp.log"
export PGPASSWORD=******
/usr/bin/psql -h *.*.*.* -p 5432 -d dyp_gp_pay -U pay_cloud_data -c "$LOAD_SQL" > ${tempfile} 2>&1
if [ $? -eq 0 ]
   then
        logmsg=`cat ${tempfile}`
        echo "load to success ${tablename}:${logmsg}"
        emailTitle="[ On the cloud ][${schema}.${tablename}] load success"
        emailContent=`echo "datax load to gppay success: ${schema}.${tablename}: ${logmsg}"`
     else
        logmsg=`cat ${tempfile}`
        msg=`echo "datax load to gppay failed:${tablename}:${logmsg}"`
        emailTitle="[ On the cloud ][${schema}.${tablename}] load failed"
        emailContent=`echo "datax load to gppay failed: ${schema}.${tablename}: ${logmsg}"`
        time=$(date "+%Y-%m-%d %H:%M:%S")
        echo "${time} ${schema}.${tablename} FATAL" >> /home/work/datax/sh-load/log/datax.log.wf
fi
echo ${emailContent} | mail -s "${emailTitle}" [email protected]