未记录事实核验 · 状态：维护中

记录仅说明核验时间与维护状态，不保证适用于所有环境；使用前请留意正文中的版本、适用条件与风险提示。

warning 适用边界与维护说明

本文适合已有 Bash、Python、MySQL 和 Spark/YARN 基础的读者学习日志排查、批量文件处理与自动化流程组织，不是可直接部署的生产方案；本次仅作全文结构与代码静态审阅，未执行脚本或进行外部事实核验。按行拆分和逗号分列仅适用于无字段内逗号、无嵌入换行且列位置符合约定的简单 CSV，日志统计依赖时间和模块字段布局。当前合并脚本仍缺少临时文件初始化、无匹配输入及输出目录检查；拆分脚本重用输出目录时可能再次处理旧分片；ETL 未固定字段顺序，空订单可能导致缺文件或沿用旧数据，同日装载也未提供幂等机制。实际接口、目标表、本地装载配置、cron 环境及 Spark 业务 JAR 和 YARN 集群须自行准备，Spark 示例仅作提交和初步定位参考。服务器初始化须确认包管理器、架构、现有账户及权限，系统更新、递归属主变更和按修改时间删除数据须经过变更评估，并满足保留、备份及恢复要求。已取消无法确定目标的可点击引用「Hadoop」，保留原提及；Missing link target: Hadoop 已取消无法确定目标的可点击引用「MySQL」，保留原提及；Missing link target: MySQL 已取消无法确定目标的可点击引用「Hive」，保留原提及；Missing link target: Hive

Linux 实战场景 - 日志分析 / 数据处理 / 自动化脚本

abstract 本文概要

综合运用前面所有知识，解决真实工作中的数据处理场景：

日志分析：快速从 GB 级日志中提取关键信息
批量数据处理：大文件拆分、清洗、合并
服务器环境搭建：新服务器快速初始化
自动化 ETL 脚本：需适配接口、数据契约和运行环境的管道示例

场景一：日志快速分析

背景

凌晨 3 点，数据管道报警：订单数据同步失败。你需要登上服务器，5 分钟内定位问题。

解决步骤

# 1. 找到最新的日志文件
ls -lt /var/log/etl/ | head -5

# 2. 看最后 100 行（快速了解当前状态）
tail -100 /var/log/etl/etl_20260318.log

# 3. 搜索错误关键词
grep -i "error\|exception\|failed\|warn" /var/log/etl/etl_20260318.log | tail -50

# 4. 找到第一个错误发生的时间
grep -i "error" /var/log/etl/etl_20260318.log | head -1

# 5. 看错误附近的上下文（前后各 10 行）
grep -n "ERROR" /var/log/etl/etl_20260318.log | head -5
# 假设第一个错误在第 234 行
sed -n '224,244p' /var/log/etl/etl_20260318.log

# 6. 统计各种错误类型的数量
grep "ERROR" /var/log/etl/etl_20260318.log \
    | awk '{print $4}' \
    | sort | uniq -c | sort -rn

# 7. 如果是 Java 应用，找 Exception 堆栈
grep -A 20 "Exception" /var/log/app/app.log | head -100

综合日志分析脚本

#!/bin/bash
# analyze_log.sh - 快速分析日志文件
# 用法: bash analyze_log.sh <日志文件> [错误关键词]

LOG_FILE="${1:-/var/log/etl/etl_$(date +%Y%m%d).log}"
KEYWORD="${2:-ERROR}"
OUTPUT_FILE="/tmp/log_analysis_$(date +%Y%m%d_%H%M%S).txt"

if [ ! -f "$LOG_FILE" ]; then
    echo "日志文件不存在: $LOG_FILE"
    exit 1
fi

{
    echo "========================================"
    echo "日志分析报告"
    echo "文件: $LOG_FILE"
    echo "分析时间: $(date '+%Y-%m-%d %H:%M:%S')"
    echo "========================================"

    echo ""
    echo "【文件基本信息】"
    echo "文件大小: $(du -sh "$LOG_FILE" | cut -f1)"
    echo "总行数: $(wc -l < "$LOG_FILE")"
    echo "时间范围:"
    echo "  最早: $(head -1 "$LOG_FILE" | awk '{print $1, $2}')"
    echo "  最晚: $(tail -1 "$LOG_FILE" | awk '{print $1, $2}')"

    echo ""
    echo "【错误统计（关键词: $KEYWORD）】"
    error_count=$(grep -c "$KEYWORD" "$LOG_FILE")
grep_status=$?
if [ "$grep_status" -gt 1 ]; then
    echo "日志搜索失败: $LOG_FILE" >&2
    exit "$grep_status"
fi
    echo "错误总数: $error_count"

    echo ""
    echo "【按小时分布的错误数量】"
    grep "$KEYWORD" "$LOG_FILE" \
        | awk '{print substr($2,1,2) ":00"}' \
        | sort | uniq -c \
        | awk '{printf "  %s时 %d次\n", $2, $1}'

    echo ""
    echo "【最近 10 条错误】"
    grep "$KEYWORD" "$LOG_FILE" | tail -10

    echo ""
    echo "【ERROR 类型分布（如果有模块信息）】"
    grep "$KEYWORD" "$LOG_FILE" \
        | awk '{print $4}' \
        | sort | uniq -c | sort -rn | head -10

} | tee "$OUTPUT_FILE"

echo ""
echo "报告已保存到: $OUTPUT_FILE"

# 使用
bash analyze_log.sh /var/log/etl/etl_20260318.log ERROR
bash analyze_log.sh /var/log/app/app.log "Exception"

场景二：大文件处理

场景 2a：拆分大 CSV 文件

# 场景：有一个 10GB 的 CSV 文件，需要拆分成小块并行处理

FILE="/data/raw/huge_orders.csv"
OUTPUT_DIR="/data/split/"
LINES_PER_FILE=1000000  # 每个文件 100 万行

mkdir -p "$OUTPUT_DIR"

# 保存表头
HEADER=$(head -1 "$FILE")

# 拆分文件（去掉表头，按行数拆分）
tail -n +2 "$FILE" | split -l "$LINES_PER_FILE" - "${OUTPUT_DIR}/part_"

# 给每个分片加上表头
for part in "${OUTPUT_DIR}"/part_*; do
    # 临时文件
    tmp="${part}.tmp"
    echo "$HEADER" > "$tmp"
    cat "$part" >> "$tmp"
    mv "$tmp" "${part}.csv"
    rm "$part"
done

echo "拆分完成，共生成 $(ls "${OUTPUT_DIR}"/*.csv | wc -l) 个文件"
ls -lh "${OUTPUT_DIR}"/*.csv

场景 2b：CSV 数据清洗

#!/bin/bash
# clean_csv.sh - CSV 数据清洗脚本
# 功能：
#   1. 去掉 Windows 换行符
#   2. 去掉首尾空格
#   3. 删除空行
#   4. 删除仅由逗号组成的行
# 注意：本脚本不检查关键字段，也不转换日期格式

INPUT_FILE="$1"
OUTPUT_FILE="${INPUT_FILE%.csv}_clean.csv"

if [ -z "$INPUT_FILE" ]; then
    echo "用法: bash clean_csv.sh <输入文件.csv>"
    exit 1
fi

echo "开始清洗: $INPUT_FILE"
echo "输入行数: $(wc -l < "$INPUT_FILE")"

# 清洗流程
# 仅适用于不含字段内逗号或嵌入换行的简单 CSV
cat "$INPUT_FILE" \
    | tr -d '\r' \
    | sed 's/^[[:space:]]*//;s/[[:space:]]*$//' \
    | grep -v '^$' \
    | grep -v '^,\+$' \
    > "$OUTPUT_FILE"

echo "输出行数: $(wc -l < "$OUTPUT_FILE")"
echo "清洗完成: $OUTPUT_FILE"

# 显示前 5 行验证
echo ""
echo "清洗后的前 5 行："
head -5 "$OUTPUT_FILE"

场景 2c：多文件合并与去重

#!/bin/bash
# merge_deduplicate.sh - 合并多个 CSV 并去重

INPUT_DIR="/data/raw/orders/"
OUTPUT_FILE="/data/processed/orders_merged.csv"
KEY_COLUMN=1   # 以第 1 列（order_id）去重

echo "查找输入文件..."
FILES=(${INPUT_DIR}orders_*.csv)
echo "找到 ${#FILES[@]} 个文件"

# 取第一个文件的表头
echo "保存表头..."
head -1 "${FILES[0]}" > "$OUTPUT_FILE"

# 合并所有文件（跳过表头）
echo "合并文件..."
for file in "${FILES[@]}"; do
    tail -n +2 "$file"
done >> "${OUTPUT_FILE}.tmp"

# 按第 1 列去重（保留第一次出现的）
echo "去重处理..."
sort -s -t',' -k${KEY_COLUMN},${KEY_COLUMN} "${OUTPUT_FILE}.tmp" \
    | awk -F',' -v col="$KEY_COLUMN" \
        'seen[$col]++ == 0 {print}' \
    >> "$OUTPUT_FILE"

rm "${OUTPUT_FILE}.tmp"

# 统计结果
input_total=$(for f in "${FILES[@]}"; do tail -n +2 "$f"; done | wc -l)
output_total=$(($(wc -l < "$OUTPUT_FILE") - 1))
duplicates=$((input_total - output_total))

echo ""
echo "合并结果:"
echo "  输入总行数: $input_total"
echo "  输出行数: $output_total"
echo "  去除重复: $duplicates"
echo "  输出文件: $OUTPUT_FILE ($(du -sh "$OUTPUT_FILE" | cut -f1))"

场景三：服务器环境初始化

#!/bin/bash
# server_init.sh - 新数据服务器初始化脚本
# 功能：配置基础环境、安装常用工具、创建目录结构

set -e

# 颜色定义
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'  # 恢复默认颜色

log() { echo -e "${GREEN}[$(date '+%H:%M:%S')]${NC} $1"; }
warn() { echo -e "${YELLOW}[WARN]${NC} $1"; }
error() { echo -e "${RED}[ERROR]${NC} $1"; exit 1; }

DATA_USER="alice"
DATA_GROUP="data"

log "====== 开始初始化服务器 ======"

# ---- 1. 系统更新 ----
log "1. 更新系统..."
if command -v yum &>/dev/null; then
    sudo yum update -y
    sudo yum install -y epel-release
    sudo yum install -y vim git wget curl jq tmux htop tree lsof \
                        python3 python3-pip \
                        java-1.8.0-openjdk java-1.8.0-openjdk-devel \
                        mysql
    PKG_MANAGER="yum"
else
    sudo apt update && sudo apt upgrade -y
    sudo apt install -y vim git wget curl jq tmux htop tree lsof \
                        python3 python3-pip \
                        default-jdk \
                        mysql-client
    PKG_MANAGER="apt"
fi
log "系统更新完成"

# ---- 2. 创建用户和目录 ----
log "2. 创建数据目录结构..."

# 创建数据目录
sudo mkdir -p /data/{raw,processed,output,archive,logs,tmp}
sudo mkdir -p /data/warehouse/{ods,dwd,dws,ads}
sudo mkdir -p /var/log/etl

# 修改目录权限（假设数据用户是 alice）
sudo chown -R ${DATA_USER}:${DATA_GROUP} /data/ /var/log/etl/ 2>/dev/null || \
    warn "设置目录权限失败，请手动执行"

log "目录创建完成"
tree /data/

# ---- 3. 配置 SSH ----
log "3. 配置 SSH..."
mkdir -p ~/.ssh
chmod 700 ~/.ssh
cat >> ~/.ssh/config << 'EOF'
# 防止 SSH 长时间不操作断线
Host *
    ServerAliveInterval 60
    ServerAliveCountMax 3
EOF

# ---- 4. 安装 Miniconda ----
log "4. 安装 Miniconda..."
if ! command -v conda &>/dev/null; then
    CONDA_URL="https://repo.anaconda.com/miniconda/Miniconda3-latest-Linux-x86_64.sh"
    wget -q "$CONDA_URL" -O /tmp/miniconda.sh
    bash /tmp/miniconda.sh -b -p ~/miniconda3
    ~/miniconda3/bin/conda init bash
    rm /tmp/miniconda.sh
    log "Miniconda 安装完成"
else
    log "Conda 已安装，跳过"
fi

# ---- 5. 配置 .bashrc ----
log "5. 配置 Shell 环境..."
cat >> ~/.bashrc << 'EOF'

# ====== 数据工程环境配置 ======
# 别名
alias ll='ls -lh'
alias la='ls -lha'
alias ..='cd ..'
alias tailf='tail -f'
alias grep='grep --color=auto'

# 历史记录
export HISTSIZE=10000
export HISTFILESIZE=100000
export HISTTIMEFORMAT="%F %T "

# Java 环境（如果安装了 JDK）
if [ -d "/usr/java/jdk1.8.0_281" ]; then
    export JAVA_HOME=/usr/java/jdk1.8.0_281
    export PATH=$JAVA_HOME/bin:$PATH
fi

# 快速创建目录并进入
mkcd() { mkdir -p "$1" && cd "$1"; }
EOF

source ~/.bashrc

log "环境配置完成"

# ---- 6. 验证安装 ----
log "====== 安装验证 ======"
echo "操作系统: $(cat /etc/os-release | grep PRETTY_NAME | cut -d= -f2 | tr -d '"')"
echo "Python: $(python3 --version 2>&1)"
echo "Java: $(java -version 2>&1 | head -1)"
echo "Git: $(git --version)"
echo "curl: $(curl --version | head -1)"

log "====== 初始化完成 ======"

场景四：ETL 自动化脚本示例（需适配后部署）

#!/bin/bash
# etl_pipeline.sh - ETL 数据管道示例，非开箱即用的生产方案
# 功能：每日完整的数据同步和处理流程

set -euo pipefail

# ============================================================
# 配置
# ============================================================
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
DATE="${1:-$(date +%Y-%m-%d)}"
DATE_FMT="${DATE//-/}"  # 20260318 格式

# 目录配置
DATA_DIR="/data"
RAW_DIR="${DATA_DIR}/raw/${DATE}"
CLEAN_DIR="${DATA_DIR}/processed/${DATE}"
OUTPUT_DIR="${DATA_DIR}/output/${DATE}"
LOG_DIR="/var/log/etl"
LOG_FILE="${LOG_DIR}/etl_${DATE_FMT}.log"

# 数据库配置（从环境变量读取，不硬编码密码）
DB_HOST="${DB_HOST:-localhost}"
DB_PORT="${DB_PORT:-3306}"
DB_USER="${DB_USER:-etl_user}"
DB_NAME="${DB_NAME:-warehouse}"
DB_PASS="${DB_PASS:?请设置 DB_PASS 环境变量}"

# ============================================================
# 工具函数
# ============================================================
mkdir -p "$LOG_DIR"

log() {
    local level="$1" msg="$2"
    local ts=$(date '+%Y-%m-%d %H:%M:%S')
    echo "${ts} [${level}] ${msg}" | tee -a "$LOG_FILE"
}
info()  { log "INFO " "$1"; }
warn()  { log "WARN " "$1"; }
error() { log "ERROR" "$1"; exit 1; }

step_start() {
    info ">>> 开始步骤: $1"
    step_start_time=$(date +%s)
}

step_end() {
    local elapsed=$(($(date +%s) - step_start_time))
    info "<<< 步骤完成 (耗时 ${elapsed}s)"
}

# ============================================================
# 步骤定义
# ============================================================

# 步骤 1：检查前置条件
check_prerequisites() {
    step_start "前置条件检查"

    # 检查目录
    for dir in "$RAW_DIR" "$CLEAN_DIR" "$OUTPUT_DIR"; do
        mkdir -p "$dir"
    done

    # 检查数据库连接
    if ! mysql -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" -p"$DB_PASS" \
         -e "SELECT 1" "$DB_NAME" &>/dev/null; then
        error "数据库连接失败：${DB_HOST}:${DB_PORT}"
    fi
    info "数据库连接正常"

    step_end
}

# 步骤 2：数据采集（模拟：实际应该调用数据源 API 或同步工具）
collect_data() {
    step_start "数据采集"

    # 模拟从 API 获取订单数据
    info "从 API 采集 ${DATE} 的订单数据..."
    curl -s -f \
        -H "Authorization: Bearer ${API_TOKEN:-}" \
        "https://data-source.example.com/api/orders?date=${DATE}" \
        -o "${RAW_DIR}/orders.json" \
        || error "采集订单数据失败"

    # 转换 JSON 为 CSV（用 Python 处理 JSON 比 Shell 更稳）
    python3 << EOF
import json, csv

with open('${RAW_DIR}/orders.json') as f:
    data = json.load(f)

orders = data.get('orders', [])
if not orders:
    print("警告：无数据")
    exit(0)

with open('${RAW_DIR}/orders.csv', 'w', newline='') as f:
    writer = csv.DictWriter(f, fieldnames=orders[0].keys())
    writer.writeheader()
    writer.writerows(orders)

print(f"写入 {len(orders)} 条记录")
EOF

    local raw_count=$(wc -l < "${RAW_DIR}/orders.csv")
    info "采集完成：原始数据 $((raw_count-1)) 条"

    step_end
}

# 步骤 3：数据清洗
clean_data() {
    step_start "数据清洗"

    local raw_file="${RAW_DIR}/orders.csv"
    local clean_file="${CLEAN_DIR}/orders.csv"

    # 清洗逻辑
    head -1 "$raw_file" > "$clean_file"   # 保留表头
    # 仅适用于简单 CSV，且第 1、5、6 列依次为 order_id、金额、状态
    tail -n +2 "$raw_file" \
        | tr -d '\r' \
        | awk -F',' '$1 != "" && $5 > 0 && $6 ~ /^(paid|pending|shipped)$/ { print }' \
        >> "$clean_file"

    local raw_count=$(($(wc -l < "$raw_file") - 1))
    local clean_count=$(($(wc -l < "$clean_file") - 1))
    local dropped=$((raw_count - clean_count))

    info "清洗完成：保留 ${clean_count} 条，丢弃 ${dropped} 条无效数据"

    if [ "$clean_count" -eq 0 ]; then
        error "清洗后无有效数据，请检查原始数据"
    fi

    step_end
}

# 步骤 4：统计汇总
aggregate_data() {
    step_start "数据汇总"

    local clean_file="${CLEAN_DIR}/orders.csv"

    # 生成各维度汇总（awk 处理）
    awk -F',' 'NR>1 {
        total_orders++
        total_amount += $5
        city_orders[$3]++
        city_amount[$3] += $5
        status_count[$6]++
    }
    END {
        # 汇总到输出目录
        print "date,metric,value" > "'"${OUTPUT_DIR}/summary.csv"'"
        printf "%s,total_orders,%d\n", "'"${DATE}"'", total_orders >> "'"${OUTPUT_DIR}/summary.csv"'"
        printf "%s,total_amount,%.2f\n", "'"${DATE}"'", total_amount >> "'"${OUTPUT_DIR}/summary.csv"'"

        # 城市维度
        for (city in city_orders) {
            printf "%s,city_orders_%s,%d\n", "'"${DATE}"'", city, city_orders[city] >> "'"${OUTPUT_DIR}/summary.csv"'"
        }
    }' "$clean_file"

    info "汇总文件已生成：${OUTPUT_DIR}/summary.csv"
    step_end
}

# 步骤 5：写入数据库
load_to_db() {
    step_start "写入数据库"

    local summary_file="${OUTPUT_DIR}/summary.csv"

    # 写入 MySQL
    mysql -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" << SQL
LOAD DATA LOCAL INFILE '${summary_file}'
INTO TABLE daily_summary
FIELDS TERMINATED BY ','
LINES TERMINATED BY '\n'
IGNORE 1 ROWS
(date, metric, value);
SQL

    local loaded=$(mysql -h "$DB_HOST" -P "$DB_PORT" -u "$DB_USER" -p"$DB_PASS" "$DB_NAME" \
                    -se "SELECT COUNT(*) FROM daily_summary WHERE date='${DATE}'")
    info "数据库写入完成：${loaded} 条记录"

    step_end
}

# 步骤 6：发送完成通知（Webhook）
send_notification() {
    step_start "发送完成通知"

    local total_time=$(($(date +%s) - pipeline_start_time))
    local row_count=$(($(wc -l < "${CLEAN_DIR}/orders.csv") - 1))

    curl -s -X POST \
        -H "Content-Type: application/json" \
        -d "{
            \"pipeline\": \"daily_etl\",
            \"date\": \"${DATE}\",
            \"status\": \"success\",
            \"rows\": ${row_count},
            \"duration_seconds\": ${total_time}
        }" \
        "${WEBHOOK_URL:-https://hooks.example.com/notify}" || \
        warn "发送通知失败（不影响主流程）"

    step_end
}

# ============================================================
# 主流程
# ============================================================
main() {
    pipeline_start_time=$(date +%s)
    info "=============================="
    info "ETL 管道开始执行"
    info "处理日期: ${DATE}"
    info "日志文件: ${LOG_FILE}"
    info "=============================="

    check_prerequisites
    collect_data
    clean_data
    aggregate_data
    load_to_db
    send_notification

    local total_time=$(($(date +%s) - pipeline_start_time))
    info "=============================="
    info "ETL 管道执行完成"
    info "总耗时: ${total_time}s"
    info "=============================="
}

# 运行主流程
main "$@"

部署为定时任务：

# 给脚本添加执行权限
chmod +x /home/alice/scripts/etl_pipeline.sh

# 设置 crontab
crontab -e

# 每天凌晨 2:00 执行 ETL
0 2 * * * /home/alice/scripts/etl_pipeline.sh >> /var/log/etl/cron_$(date +\%Y\%m\%d).log 2>&1

场景五：Spark 作业提交与监控

#!/bin/bash
# submit_spark_job.sh - 提交 Spark 作业并监控

set -e

DATE="${1:-$(date +%Y-%m-%d)}"
JAR_FILE="/home/alice/jars/data-pipeline-1.0.jar"
MAIN_CLASS="com.company.DailyETL"
LOG_DIR="/var/log/spark"
LOG_FILE="${LOG_DIR}/spark_$(date +%Y%m%d_%H%M%S).log"

mkdir -p "$LOG_DIR"

echo "提交 Spark 作业..."
echo "日期: $DATE"
echo "JAR: $JAR_FILE"
echo "日志: $LOG_FILE"

# 在 tmux 中提交（这样可以随时查看，SSH 断线也不影响）
tmux new-session -d -s "spark_${DATE//-/}" \
    "spark-submit \
        --master yarn \
        --deploy-mode cluster \
        --driver-memory 4g \
        --executor-memory 8g \
        --executor-cores 4 \
        --num-executors 20 \
        --conf spark.sql.adaptive.enabled=true \
        --class ${MAIN_CLASS} \
        ${JAR_FILE} \
        --date ${DATE} \
        > ${LOG_FILE} 2>&1; echo '任务完成，退出码: '\$?"

echo ""
echo "作业已在 tmux 会话中启动"
echo "查看日志: tmux attach -t spark_${DATE//-/}"
echo "实时日志: tail -f ${LOG_FILE}"

# 等待作业启动，检查 YARN 作业 ID
sleep 10
APP_ID=$(grep "application_" "$LOG_FILE" | head -1 | grep -o 'application_[0-9_]*' | head -1)

if [ -n "$APP_ID" ]; then
    echo ""
    echo "YARN 作业 ID: $APP_ID"
    echo "查看作业状态: yarn application -status $APP_ID"
    echo "查看作业日志: yarn logs -applicationId $APP_ID"
fi

常用一行命令（One-liners）

# 找出占用磁盘最多的目录（TOP 10）
du -h --max-depth=2 /data/ 2>/dev/null | sort -rh | head -10

# 按日志前两个空白字段统计连续相同行数（不等同于实时按分钟统计）
tail -f app.log | awk '{print $1" "$2}' | uniq -c

# 找出最近 1 天修改的所有大文件（>100MB）
find / -mtime -1 -size +100M -type f 2>/dev/null

# 批量替换目录下所有文件中的某个字符串
find /config -name "*.xml" -exec sed -i 's/old_host/new_host/g' {} \;

# 监控日志，发现 ERROR 就响铃
tail -f app.log | grep --line-buffered "ERROR" | awk '{print "\a" $0}'

# 统计每个 IP 的访问量（访问日志）
awk '{print $1}' access.log | sort | uniq -c | sort -rn | head -20

# 快速测试某台机器的某个端口是否通
(echo >/dev/tcp/192.168.1.100/3306) 2>/dev/null && echo "开放" || echo "关闭"

# 批量创建测试数据文件
for i in $(seq 1 10); do
    echo "id,name,value" > /data/test/data_${i}.csv
    seq 1 1000 | awk -v i=$i '{printf "%d,name_%d_%d,%.2f\n", NR, i, NR, rand()*100}' >> /data/test/data_${i}.csv
done

排查案例：磁盘满导致 ETL 失败

# 收到告警：ETL 失败，错误信息：No space left on device

# 1. 查看磁盘使用情况
df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/sda1        50G   50G     0 100% /data

# 2. 找最占空间的目录
du -sh /data/* 2>/dev/null | sort -rh | head -10

# 3. 找大文件
find /data -size +1G -type f 2>/dev/null | xargs ls -lh | sort -k5 -rh

# 4. 找可以清理的过期日志
ls -lt /data/logs/ | head -20
# 发现 90 天前的日志还在

# 5. 清理
find /data/logs -name "*.log" -mtime +30 -exec rm {} \;
find /data/raw -name "*.csv" -mtime +90 -exec rm {} \;

# 6. 清理完验证
df -h
# Filesystem      Size  Used Avail Use% Mounted on
# /dev/sda1        50G   35G   15G  70% /data

# 7. 重跑 ETL
bash /home/alice/scripts/etl_pipeline.sh

# 8. 设置定期清理的 crontab，避免下次再发生
crontab -e
# 0 3 * * * find /data/logs -name "*.log" -mtime +30 -delete

下一步建议

学完本系列，推荐的进阶方向：

深入 Python：Shell 擅长文件操作和简单文本处理，复杂逻辑用 Python 更合适
大数据工具：Hadoop / Spark / Flink 的 Shell 脚本提交和管理
数据库操作：Shell 中调用 MySQL / Hive 命令行执行 SQL
监控与告警：结合 Shell 脚本和公司的监控系统（Prometheus/Grafana）

tip 最后的建议

Linux 技能需要在真实工作中积累。遇到不会的命令，不要绕过，而是搞清楚原理再往前走。每个在生产上排查过的问题、每个写过的脚本，都是真实的技能沉淀。

#Linux #Shell #实战 #ETL #数据工程 #自动化