huanghai/aiData
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

'commit'

Browse Source
This commit is contained in:
HuangHai
2026-02-05 13:24:59 +08:00
parent 00375a80b2
commit 8e0309bb7b
69 changed files with 2084 additions and 3124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

171
DataX/run_jobs_full.sh
View File

@@ -1,21 +1,36 @@
#!/bin/bash
# 环境变量补丁 (针对 Crontab 环境下可能找不到 java 的问题)
# 环境变量补丁
export PATH=$PATH:/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin
[ -f /etc/profile ] && . /etc/profile
[ -f ~/.bash_profile ] && . ~/.bash_profile
[ -f ~/.profile ] && . ~/.profile
# 配置路径
DATAX_HOME="/usr/local/datax"
# 获取当前脚本所在目录
# 检查参数
TARGET=$1
if [ "$TARGET" != "doris" ] && [ "$TARGET" != "mysql" ]; then
echo "使用方法: $0 [doris|mysql]"
exit 1
fi
# 配置文件
SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )"
# JSON配置文件目录指向 json_full
CONF_DIR="$SCRIPT_DIR/json_full"
CONF_DIR="$SCRIPT_DIR/json"
CONFIG_FILE="$SCRIPT_DIR/db_config.sh"
# 加载数据库配置
if [ -f "$CONFIG_FILE" ]; then
source "$CONFIG_FILE"
else
echo "错误: 找不到配置文件 $CONFIG_FILE"
exit 1
fi
DATAX_HOME="/usr/local/datax"
BIN_DIR="$DATAX_HOME/bin"
DATAX_PY="$BIN_DIR/datax.py"
# 明确指定要执行的15个文件（全量同步）
# 明确指定要执行的15个文件
JOBS=(
"t_equipment_charge_order.json"
"t_equipment_charge_order_detail.json"
@@ -34,20 +49,29 @@ JOBS=(
"t_user_upload_fault.json"
)
# 定义全局参数
PARAMS="-Dsrc_user=ylt -Dsrc_pwd=Ycharge666 -Dsrc_jdbc=jdbc:mysql://rm-bp1ux6tuk49er80t9xo.mysql.rds.aliyuncs.com:3306/yltcharge -Ddest_user=root -Ddest_pwd=DsideaL147258369 -Ddest_load_url=10.10.14.204:8030 -Ddest_jdbc=jdbc:mysql://10.10.14.204:9030/yltcharge"
# 日志配置
LOG_DIR="$SCRIPT_DIR/logs"
if [ ! -d "$LOG_DIR" ]; then
mkdir -p "$LOG_DIR"
fi
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
LOG_FILE="$LOG_DIR/run_full_${TARGET}_${TIMESTAMP}.log"
echo "日志将同时记录到: $LOG_FILE"
# 根据目标设置参数
if [ "$TARGET" == "doris" ]; then
DEST_PARAMS="$DORIS_DEST_PARAMS"
TITLE="MySQL -> Doris Full Sync"
else
DEST_PARAMS="$MYSQL_DEST_PARAMS"
TITLE="MySQL -> MySQL Full Sync"
fi
# 在脚本最开始记录开始时间
SCRIPT_START_TIME=$(date '+%Y-%m-%d %H:%M:%S')
SCRIPT_START_TIMESTAMP=$(date +%s)
echo "DataX全量同步脚本 (Full Sync)"
echo "=========================="
echo "DataX路径: $DATAX_PY"
echo "配置文件目录: $CONF_DIR"
echo "任务数量: ${#JOBS[@]}"
echo "脚本开始时间: $SCRIPT_START_TIME"
echo "DataX全量同步脚本 ($TITLE)"
echo "=========================="
# 切换到配置文件目录
@@ -56,138 +80,59 @@ cd "$CONF_DIR" || {
exit 1
}
echo "当前目录: $(pwd)"
echo
# 执行计数器
SUCCESS_COUNT=0
FAIL_COUNT=0
TOTAL=${#JOBS[@]}
CURRENT=0
# 数组存储每个任务的执行详情
declare -a TASK_RESULTS=()
declare -a TASK_DURATIONS=()
declare -a TASK_STATUSES=()
for JOB in "${JOBS[@]}"; do
CURRENT=$((CURRENT + 1))
TABLE_NAME="${JOB%.json}"
echo "任务 [$((SUCCESS_COUNT + FAIL_COUNT + 1))/$TOTAL] $JOB"
echo "任务 [$CURRENT/$TOTAL] $JOB"
echo "任务开始时间: $(date '+%H:%M:%S')"
echo "----------------------------------------"
# 动态构造全量 SQL：SELECT * FROM table
QUERY_SQL="SELECT * FROM $TABLE_NAME"
# 动态构造清空 SQL：TRUNCATE TABLE table
PRE_SQL="TRUNCATE TABLE $TABLE_NAME"
# 检查文件是否存在
if [ ! -f "$JOB" ]; then
echo "✗ 错误: 文件不存在 - $JOB"
FAIL_COUNT=$((FAIL_COUNT + 1))
TASK_RESULTS+=("$JOB")
TASK_STATUSES+=("文件不存在")
TASK_DURATIONS+=("0")
echo "----------------------------------------"
continue
fi
# 注意：这里需要仔细处理参数引用，将变量作为整体传递
PARAMS="$SRC_BASE_PARAMS $DEST_PARAMS -Dsrc_query_sql=\"$QUERY_SQL\" -Ddest_pre_sql=\"$PRE_SQL\""
# 打印最终的执行命令（去除敏感信息后），用于调试
# echo "执行命令: python $DATAX_PY -p \"$PARAMS\" $JOB"
# 记录开始时间（用于计算耗时）
START_TIME=$(date +%s)
# 使用 eval 来正确处理带引号的参数，或者直接修改调用方式
# 由于PARAMS中包含空格和引号，直接引用可能会导致参数解析错误
# 最稳妥的方式是将 -p 后面的参数用单引号包起来，但内部又有双引号，比较麻烦
# 我们可以尝试直接将所有参数拼接好，让python去解析
# 执行任务 (添加 -p 参数传递变量)
/usr/bin/python "$DATAX_PY" -p "$PARAMS" "$JOB"
EXIT_CODE=$?
# 计算耗时
/usr/bin/python "$DATAX_PY" -p "$PARAMS" "$JOB" 2>&1 | tee -a "$LOG_FILE"
EXIT_CODE=${PIPESTATUS[0]}
END_TIME=$(date +%s)
DURATION=$((END_TIME - START_TIME))
# 保存任务详情
TASK_RESULTS+=("$JOB")
TASK_DURATIONS+=("$DURATION")
# 处理执行结果
if [ $EXIT_CODE -eq 0 ]; then
echo "✓ 成功: $JOB (耗时: ${DURATION}秒)"
SUCCESS_COUNT=$((SUCCESS_COUNT + 1))
TASK_STATUSES+=("成功")
else
echo "✗ 失败: $JOB (耗时: ${DURATION}秒)"
echo "错误代码: $EXIT_CODE"
FAIL_COUNT=$((FAIL_COUNT + 1))
TASK_STATUSES+=("失败")
# 询问是否继续执行后续任务
# 全量同步失败可能需要人工干预，但这里保持自动询问或直接继续
# 为方便自动化，这里改为仅提示，不阻塞 (或者保持原样)
# 保持原样:
# echo "----------------------------------------"
# read -p "任务失败，是否继续执行下一个任务？(y/n): " -n 1 -r
# echo
# if [[ ! $REPLY =~ ^[Yy]$ ]]; then
# echo "用户选择停止执行"
# break
# fi
fi
echo "任务结束时间: $(date '+%H:%M:%S')"
echo "----------------------------------------"
echo
# 可选：任务间等待时间（秒）
# sleep 1
done
# 记录脚本结束时间
SCRIPT_END_TIME=$(date '+%Y-%m-%d %H:%M:%S')
SCRIPT_END_TIMESTAMP=$(date +%s)
TOTAL_DURATION=$((SCRIPT_END_TIMESTAMP - SCRIPT_START_TIMESTAMP))
# ... (汇总逻辑与之前相同)
echo "=========================="
echo "执行结果汇总 (Full Sync)"
echo "执行结果汇总 (Full Sync to $TARGET)"
echo "=========================="
echo "脚本开始时间: $SCRIPT_START_TIME"
echo "脚本结束时间: $SCRIPT_END_TIME"
echo "脚本总耗时: ${TOTAL_DURATION}秒"
echo "=========================="
echo "任务执行详情:"
echo "----------------------------------------"
# 输出每个任务的执行结果
for i in "${!TASK_RESULTS[@]}"; do
STATUS_ICON="✓"
if [ "${TASK_STATUSES[$i]}" = "失败" ] || [ "${TASK_STATUSES[$i]}" = "文件不存在" ]; then
STATUS_ICON="✗"
fi
printf "%-40s %-2s %-10s %-8s秒\n" \
"${TASK_RESULTS[$i]}" \
"$STATUS_ICON" \
"${TASK_STATUSES[$i]}" \
"${TASK_DURATIONS[$i]}"
done
echo "----------------------------------------"
echo "汇总统计:"
echo "总任务数: $TOTAL"
echo "成功: $SUCCESS_COUNT"
echo "失败: $FAIL_COUNT"
if [ $TOTAL -gt 0 ]; then
SUCCESS_RATE=$((SUCCESS_COUNT * 100 / TOTAL))
echo "成功率: ${SUCCESS_RATE}%"
fi
# 计算平均耗时
if [ ${#TASK_DURATIONS[@]} -gt 0 ]; then
TOTAL_TASK_TIME=0
VALID_TASKS=0
for duration in "${TASK_DURATIONS[@]}"; do
if [ "$duration" -gt 0 ]; then
TOTAL_TASK_TIME=$((TOTAL_TASK_TIME + duration))
VALID_TASKS=$((VALID_TASKS + 1))
fi
done
if [ $VALID_TASKS -gt 0 ]; then
AVG_DURATION=$((TOTAL_TASK_TIME / VALID_TASKS))
echo "平均任务耗时: ${AVG_DURATION}秒"
fi
fi
echo "=========================="