第 9 章：管道组合

管道是 Unix 的灵魂——每个工具做一件事，通过管道连接，就能完成复杂的任务。

9.1 管道基础

管道的工作原理

命令1 (stdout) → | → 命令2 (stdin) → | → 命令3 (stdin) → stdout
     生产者           消费者/生产者           消费者

核心管道工具

工具	功能	常用选项
`cat`	输出文件内容	`-n` 行号
`sort`	排序	`-n` 数值 `-r` 逆序 `-k` 指定列 `-u` 去重
`uniq`	去重（需排序）	`-c` 计数 `-d` 仅重复 `-u` 仅唯一
`head`	前 N 行	`-n 10`
`tail`	后 N 行	`-n 10` `-f` 实时
`cut`	截取字段	`-d` 分隔符 `-f` 字段 `-c` 字符
`tr`	替换/删除字符	`-d` 删除 `-s` 压缩
`wc`	计数	`-l` 行数 `-w` 词数 `-c` 字节数
`grep`	过滤	`-v` 反选 `-i` 忽略大小写 `-c` 计数
`tee`	输出到屏幕和文件	`tee file`
`paste`	合并行	`-d` 分隔符 `-s` 单行

9.2 经典管道模式

模式一：统计排序

# 统计文件中各单词出现频率
$ tr -s ' ' '\n' < file | sort | uniq -c | sort -rn | head -10

# 分解：
# tr -s ' ' '\n'  — 将空格替换为换行（一行一个单词）
# sort             — 排序（uniq 需要有序输入）
# uniq -c          — 统计每行出现次数
# sort -rn         — 按数量降序排列
# head -10         — 取前 10 个

模式二：去重计数

# 统计日志中独立 IP 数
$ awk '{print $1}' access.log | sort -u | wc -l

# 统计各 IP 出现次数
$ awk '{print $1}' access.log | sort | uniq -c | sort -rn

模式三：过滤转换

# 从配置文件中提取非注释的有效配置
$ grep -v '^#' config.txt | grep -v '^$' | sed 's/^[[:space:]]*//'

# 或用 awk 一步完成
$ awk '!/^#/ && NF>0 {gsub(/^[[:space:]]+/, ""); print}' config.txt

模式四：分步处理

# 多步数据处理
$ cat data.txt \
    | sed 's/[[:space:]]*$//' \
    | grep -v '^$' \
    | awk -F, '$3 > 100 {print $1, $3}' \
    | sort -k2 -rn \
    | head -20

模式五：条件执行

# 使用 tee 分支处理
$ cat access.log \
    | tee >(grep '404' > 404_errors.log) \
    | tee >(grep '500' > 500_errors.log) \
    | awk '$9 >= 400' > all_errors.log

# 使用 && 和 || 控制流程
$ grep 'error' logfile && echo "找到错误" || echo "没有错误"

9.3 数据清洗管道

🏢 场景一：清洗用户数据

cat > dirty_data.csv << 'EOF'
  Alice , Engineering, 15000  
bob,Marketing,  12000
CAROL,Engineering,16000
Dave,Sales,11000
,Marketing,13000
Eve,Engineering,
EOF

# 清洗管道
$ cat dirty_data.csv \
    | sed 's/^[[:space:]]*//' \
    | sed 's/[[:space:]]*$//' \
    | grep -v '^,' \
    | grep -v ',,$' \
    | awk -F, '{
        # 统一大小写（首字母大写）
        name = $1
        gsub(/[a-z]/, "", substr(name,1,1))  
        
        # 清理数字
        gsub(/[[:space:]]/, "", $3)
        
        if (name != "" && $3 != "")
            printf "%s,%s,%s\n", name, $2, $3
    }'

🏢 场景二：日志清洗

# 清洗日志：标准化时间格式、提取关键信息
$ cat raw.log \
    | sed 's/\r$//' \
    | awk '!seen[$0]++' \
    | sed -E 's/[[:space:]]+/ /g' \
    | awk '{
        # 标准化时间戳
        gsub(/\[/, "")
        gsub(/\]/, "")
        
        # 提取关键字段
        timestamp = $4
        level = $6
        message = substr($0, index($0, $7))
        
        printf "%s [%s] %s\n", timestamp, level, message
    }'

🏢 场景三：配置文件标准化

# 将不同格式的配置统一为 key=value
cat mixed.conf << 'EOF'
# Database Configuration
host     = localhost
port: 5432
database_name=myapp
  user = admin
password = secret123

# Server Settings
server.host: 0.0.0.0
server.port=8080
EOF

$ cat mixed.conf \
    | sed 's/^[[:space:]]*//' \
    | sed 's/[[:space:]]*$//' \
    | grep -v '^#' \
    | grep -v '^$' \
    | sed 's/[[:space:]]*[:=][[:space:]]*/=/' \
    | sed 's/\./_/g'

9.4 复杂管道构建

多工具协作示例

# 生成文件类型统计报告
$ find . -type f \
    | sed 's/.*\./\./' \
    | sort \
    | uniq -c \
    | sort -rn \
    | awk '{
        count = $1
        ext = $2
        total += count
        printf "%6d  %-10s", count, ext
        for (i=0; i<count/5; i++) printf "█"
        printf "\n"
    }'
    END { printf "\n总计: %d 个文件\n", total }

# 分析 HTTP 访问日志中的错误模式
$ cat access.log \
    | awk '$9 >= 400' \
    | awk '{print $7, $9}' \
    | sort \
    | uniq -c \
    | sort -rn \
    | head -20 \
    | awk '{
        count = $1
        path = $2
        status = $3
        bar = ""
        for (i=0; i<count/2; i++) bar = bar "█"
        printf "%4d  %-40s %s %s\n", count, path, status, bar
    }'

使用 tee 进行分支处理

# 将日志同时按级别分类
$ cat app.log \
    | tee >(grep -i 'error' > errors.log) \
    | tee >(grep -i 'warn' > warnings.log) \
    | tee >(grep -i 'info' > info.log) \
    | grep -i 'debug' > debug.log

# 带统计的处理
$ cat data.csv \
    | tee >(wc -l > /tmp/total_count.txt) \
    | awk -F, '$3 > 100' \
    | tee >(wc -l > /tmp/filtered_count.txt) \
    | sort -t, -k3 -rn

管道中的错误处理

# 使用 pipefail 捕获管道中的错误
$ set -o pipefail
$ command1 | command2 | command3
$ echo $?  # 如果任何命令失败，返回非零值

# 检查每个步骤
$ cat file.txt | grep 'pattern' | awk '{print $1}'
$ # 如果 file.txt 不存在，cat 会报错
$ # 建议先检查文件存在
$ [ -f file.txt ] && cat file.txt | grep 'pattern' | awk '{print $1}'

# 使用临时文件避免管道错误
$ tmpfile=$(mktemp)
$ grep 'pattern' file.txt > "$tmpfile"
$ if [ -s "$tmpfile" ]; then
    awk '{print $1}' "$tmpfile"
fi
$ rm "$tmpfile"

9.5 性能优化

管道性能原则

原则	说明	示例
先过滤再处理	尽早减少数据量	`grep ’error’ log
避免不必要的 cat	直接将文件名传给工具	`awk '{print}' file` 而不是 `cat file
用 awk 替代多次调用	单个 awk 替代多个管道	一个 awk 程序替代 grep + sed + awk
避免无用的排序	只在需要时排序	使用 `-u` 而不是 `sort
使用 LC_ALL=C	加速字符处理	`LC_ALL=C sort file`

# ❌ 慢：多次进程启动
$ cat file | grep 'error' | awk '{print $2}' | sort | uniq -c | sort -rn

# ✅ 快：尽量用一个 awk
$ awk '/error/ {count[$2]++} END {for (k in count) print count[k], k}' file | sort -rn

# ✅ 更快：设置 locale
$ LC_ALL=C awk '/error/ {count[$2]++} END {for (k in count) print count[k], k}' file | sort -rn

大文件处理

# 处理大文件时，尽早过滤
# ❌ 处理所有行再筛选
$ awk '{print $1, $9}' huge.log | grep '404'

# ✅ 先筛选再提取
$ grep '404' huge.log | awk '{print $1, $9}'

# 使用 head 限制输出进行测试
$ head -1000 huge.log | awk '{count[$1]++} END {for (k in count) print count[k], k}' | sort -rn

并行处理

# 使用 xargs 并行处理
$ find . -name "*.log" | xargs -P 4 -I {} sh -c 'grep -c "error" {}'

# 使用 GNU parallel
$ find . -name "*.log" | parallel 'grep -c "error" {}'

# 分片处理大文件
$ split -l 1000000 huge.txt chunk_
$ for f in chunk_*; do
    awk '{count[$1]++} END {for (k in count) print count[k], k}' "$f"
done | awk '{count[$2]+=$1} END {for (k in count) print count[k], k}' | sort -rn
$ rm chunk_*

9.6 实用管道组合

系统管理

# 找出占用磁盘最多的文件
$ find / -type f -exec du -h {} + 2>/dev/null | sort -rh | head -20

# 找出最大的目录
$ du -sh */ 2>/dev/null | sort -rh | head -10

# 统计各类型文件的数量和大小
$ find . -type f -printf '%s %f\n' | awk '{
    split($2, a, ".")
    ext = a[length(a)]
    count[ext]++
    size[ext] += $1
} END {
    for (e in count)
        printf "%8d %10.2f MB  .%s\n", count[e], size[e]/1048576, e
}' | sort -k2 -rn

# 查看当前连接数最多的 IP
$ netstat -an | awk '/ESTABLISHED/ {split($5, a, ":"); count[a[1]]++} END {for (ip in count) print count[ip], ip}' | sort -rn | head -10

# 监控日志文件增长速度
$ tail -f access.log | pv -l -i 5 -r > /dev/null

文本处理

# 统计代码行数（按语言分类）
$ find . -type f \( -name "*.py" -o -name "*.js" -o -name "*.sh" -o -name "*.awk" \) \
    | while read f; do
        ext="${f##*.}"
        lines=$(wc -l < "$f")
        echo "$ext $lines"
    done \
    | awk '{count[$1]+=$2; files[$1]++} END {for (e in count) printf "%-10s %6d 文件  %8d 行\n", e, files[e], count[e]}' \
    | sort -k3 -rn

# 找出重复行
$ sort file | uniq -d

# 找出两个文件的差异行
$ sort file1 file2 | uniq -u

# 合并两个文件（去重）
$ sort -u file1 file2 > merged.txt

数据转换

# CSV 转 JSON（简单版）
$ awk -F, '
NR==1 { split($0, headers, ","); next }
{
    printf "{"
    for (i=1; i<=NF; i++) {
        printf "\"%s\":\"%s\"", headers[i], $i
        if (i<NF) printf ","
    }
    printf "}\n"
}' data.csv

# JSON Lines 转 CSV（简单版）
$ sed -E 's/\{|\}//g; s/","/,/g; s/"//g' jsonl.txt

# 键值对文件转表格
$ awk -F= '{
    if (prev != FILENAME) { print "--- " FILENAME " ---"; prev = FILENAME }
    printf "%-20s = %s\n", $1, $2
}' *.conf

9.7 调试管道

分步调试

# 每个步骤后检查输出
$ cat data.txt > /tmp/step1.txt
$ grep 'pattern' /tmp/step1.txt > /tmp/step2.txt
$ awk '{print $2}' /tmp/step2.txt > /tmp/step3.txt
$ sort /tmp/step3.txt > /tmp/step4.txt

# 使用 tee 查看中间结果
$ cat data.txt \
    | tee /tmp/debug1.txt \
    | grep 'pattern' \
    | tee /tmp/debug2.txt \
    | awk '{print $2}' \
    | tee /tmp/debug3.txt \
    | sort

# 查看每一步的输出行数
$ cat data.txt \
    | (n=$(wc -l); echo "step0: $n 行" >&2; cat) \
    | grep 'pattern' \
    | (n=$(wc -l); echo "step1: $n 行" >&2; cat) \
    | awk '{print $2}' \
    | (n=$(wc -l); echo "step2: $n 行" >&2; cat) \
    | sort

常见管道错误

错误	原因	解决方案
`Broken pipe`	上游命令提前终止	使用 `head` 时注意
没有输出	上游命令没有输出	检查上游命令
输出不完整	`sort` 需要等所有输入	大数据量时考虑 `sort --parallel`
编码问题	文件编码不一致	`file` 查看编码，`iconv` 转换

9.8 速查：管道模板

# 统计排序
command | sort | uniq -c | sort -rn | head -N

# 提取去重
command | awk '{print $N}' | sort -u

# 过滤转换
command | grep 'pattern' | sed 's/old/new/g' | awk '{print}'

# 条件统计
command | awk 'condition {count[$key]++} END {for (k in count) print k, count[k]}'

# 分类输出
command | tee >(filter1 > file1) | tee >(filter2 > file2) | filter3 > file3

# 并行处理
find . -name "*.log" | xargs -P 4 -I {} command {}

# 格式化输出
command | awk '{printf "%-20s %10d\n", $1, $2}' | column -t

扩展阅读

The Art of Unix Programming — Eric Raymond
Unix Pipes Power
GNU Parallel Tutorial

下一章：第 10 章：系统管理 — 配置文件修改、批量操作、监控脚本。

强曰为道

第 9 章：管道组合

第 9 章：管道组合

9.1 管道基础

管道的工作原理

核心管道工具

9.2 经典管道模式

模式一：统计排序

模式二：去重计数

模式三：过滤转换

模式四：分步处理

模式五：条件执行

9.3 数据清洗管道

🏢 场景一：清洗用户数据

🏢 场景二：日志清洗

🏢 场景三：配置文件标准化

9.4 复杂管道构建

多工具协作示例

使用 tee 进行分支处理

管道中的错误处理

9.5 性能优化

管道性能原则

大文件处理

并行处理

9.6 实用管道组合

系统管理

文本处理

数据转换

9.7 调试管道

分步调试

常见管道错误

9.8 速查：管道模板

扩展阅读