Linux cho DE 4 — Pipe, redirection & kết hợp lệnh

Vì sao pipe & redirection là "chất keo" của tự động hoá

Ba bài trước cho bạn các công cụ: điều hướng file (Linux 2), và bộ ba grep/sed/awk để xử lý văn bản (Linux 3). Nhưng một data engineer hiếm khi chạy một lệnh đơn lẻ. Công việc thật là ghép chúng lại: lấy output của lệnh này làm input cho lệnh kia, tách log ra file riêng, biến danh sách file thành tham số cho một lệnh khác, tải một file rồi giải nén rồi nạp vào database — tất cả trong một dòng, hoặc một script chạy trong Airflow, cron, hay BashOperator.

Cơ chế cho phép ghép nối đó chính là pipe và redirection. Nếu grep/sed/awk là những viên gạch, thì pipe và redirection là vữa gắn kết chúng thành một bức tường — một pipeline ETL thực thụ. Không hiểu chắc phần này, bạn sẽ mãi copy-paste những dòng lệnh mà không biết vì sao lỗi 2>&1 lại đặt sau chứ không phải trước, hay vì sao ... | while read lại làm mất biến đếm.

Bài này mổ xẻ cơ chế luồng dữ liệu của shell: ba luồng chuẩn, các toán tử chuyển hướng, pipe, rồi tee, xargs, command/process substitution và here-doc. Đây là bài "nối mạch" của series — nắm xong bạn đã đủ nền để viết script tự động hoá thật sự ở Linux 8 — Shell scripting.

Ba luồng chuẩn: stdin (0), stdout (1), stderr (2)

Mỗi tiến trình (process) trên Linux khi khởi động được cấp sẵn ba kênh dữ liệu, gọi là standard streams, mỗi kênh có một số hiệu file descriptor (fd) cố định:

Điểm mấu chốt mà nhiều người bỏ qua: stdout và stderr là hai luồng TÁCH BIỆT. Mặc định cả hai cùng đổ ra màn hình nên trông như một, nhưng chúng độc lập. Tách biệt này cực kỳ hữu ích: bạn có thể giữ lại kết quả sạch (stdout) và ghi lỗi (stderr) ra chỗ khác — ví dụ pipeline ETL vẫn xuất dữ liệu bình thường qua stdout trong khi mọi cảnh báo được gom vào file log riêng để đội vận hành soi.

Vì sao lỗi lại đi qua kênh riêng (fd 2)? Vì nếu lỗi trộn lẫn vào stdout, khi bạn pipe stdout sang lệnh sau, các dòng lỗi sẽ "làm bẩn" dữ liệu. Tách stderr ra fd 2 giúp dữ liệu chảy qua pipe luôn sạch.

# ls in tên file ra stdout, in lỗi "không tồn tại" ra stderr
ls /etc /thu_muc_khong_ton_tai
# /etc/...            <- stdout
# ls: cannot access '/thu_muc_khong_ton_tai': No such file or directory  <- stderr

Redirection — chuyển hướng luồng vào/ra file

Redirection là đổi đích của một luồng: thay vì đọc từ / ghi ra terminal, ta trỏ nó tới file (hoặc file descriptor khác).

Ghi stdout ra file: > và >>

psql -c "SELECT ..." > ketqua.csv     # GHI ĐÈ: file cũ bị xoá sạch, tạo mới
echo "run at $(date)" >> job.log       # GHI THÊM (append) vào cuối file

• > — ghi đè. Nếu file đã có nội dung, nó bị cắt về rỗng (truncate) rồi mới ghi. Rất dễ mất dữ liệu do gõ nhầm. Thực chất > là viết tắt của 1> (chuyển hướng fd 1).
• >> — ghi thêm vào cuối, không xoá nội dung cũ. Dùng cho log tích luỹ.

Cảnh báo: > file xoá sạch file trước cả khi lệnh bên trái chạy. Câu kinh điển gây mất dữ liệu là sort file.txt > file.txt — file bị truncate về rỗng trước khi sort kịp đọc, kết quả là file trắng. Muốn sắp xếp tại chỗ phải ghi ra file tạm rồi đổi tên, hoặc dùng sort -o file.txt file.txt.

Đọc stdin từ file: <

psql -f query.sql                      # nhiều lệnh có cờ đọc file sẵn
grep 'ERROR' < app.log                 # tương đương grep 'ERROR' app.log
wc -l < data.csv                        # đưa file vào stdin của wc

< trỏ stdin tới file. Với nhiều lệnh, cmd < file và cmd file cho kết quả giống nhau, nhưng chúng khác nhau tinh tế: với <, lệnh nhận dữ liệu qua stdin (không biết tên file), nên wc -l < data.csv chỉ in số (không kèm tên file) — hữu ích khi cần đúng con số.

Chuyển hướng stderr: 2> và gộp 2>&1

Đây là phần hay nhầm nhất. Muốn chuyển hướng stderr phải ghi rõ số fd 2:

python etl.py > out.log 2> err.log      # stdout -> out.log, stderr -> err.log (tách riêng)
python etl.py 2> /dev/null               # vứt bỏ mọi lỗi, chỉ giữ stdout
python etl.py > all.log 2>&1             # GỘP cả hai vào all.log

Cú pháp 2>&1 đọc là: "trỏ fd 2 (stderr) tới cùng nơi mà fd 1 (stdout) đang trỏ". Ký hiệu &1 nghĩa là "file descriptor số 1", không phải file tên 1.

Thứ tự cực kỳ quan trọng. So sánh:

python etl.py > all.log 2>&1     # ĐÚNG: stdout tới all.log, rồi stderr theo stdout -> cả hai vào all.log
python etl.py 2>&1 > all.log     # SAI: stderr theo stdout (lúc này còn là màn hình), RỒI stdout mới đổi sang file

Shell xử lý redirection từ trái sang phải. Ở dòng thứ hai, khi gặp 2>&1 thì stdout vẫn đang trỏ ra màn hình, nên stderr bị trỏ ra màn hình; sau đó > all.log chỉ đổi stdout — kết quả là lỗi vẫn hiện ra màn hình. Quy tắc nhớ: đổi đích stdout trước, rồi mới 2>&1.

Bản rút gọn &> file (bash) gộp cả hai luồng vào một file trong một ký hiệu:

python etl.py &> all.log         # tương đương > all.log 2>&1 (chỉ bash/zsh)
python etl.py &>> all.log        # gộp cả hai, ghi THÊM

/dev/null — hố đen dữ liệu

/dev/null là một "file thiết bị" đặc biệt: ghi gì vào đó cũng biến mất, đọc từ đó luôn được rỗng. Nó là "thùng rác" của Unix, dùng để vứt output không cần:

command 2> /dev/null            # bỏ mọi lỗi/cảnh báo, chỉ giữ kết quả
command > /dev/null 2>&1        # chạy im lặng hoàn toàn (cron job thường dùng)
diff a.txt b.txt > /dev/null && echo "giong nhau"   # chỉ cần mã thoát, không cần nội dung

Bảng tổng kết các toán tử chuyển hướng:

Pipe | — nối stdout của lệnh này vào stdin của lệnh kia

Pipe là linh hồn của shell. Toán tử | lấy stdout của lệnh bên trái nối thẳng vào stdin của lệnh bên phải, không đi qua đĩa (dữ liệu chảy trong bộ nhớ, theo dòng). Các lệnh chạy song song: lệnh sau bắt đầu xử lý ngay khi lệnh trước nhả ra dòng đầu tiên, nên pipe xử lý được cả luồng dữ liệu khổng lồ hay vô hạn (tail -f).

cat access.log | grep 'ERROR' | awk '{print $1}' | sort | uniq -c | sort -rn

Đây chính là hiện thân của triết lý Unix: mỗi công cụ làm một việc thật tốt, ghép qua pipe để giải quyết việc lớn. cat đọc, grep lọc, awk tách cột, sort/uniq gộp đếm — không lệnh nào biết về lệnh khác, chúng chỉ nói chuyện qua luồng text.

Sơ đồ dưới minh hoạ luồng dữ liệu qua một pipeline: stdout của mỗi tầng nối vào stdin tầng sau, trong khi stderr của mọi tầng vẫn đi thẳng ra terminal (không chảy qua pipe):

Vài lưu ý thực chiến với pipe:

• Pipe chỉ chuyển stdout, KHÔNG chuyển stderr. Nếu muốn cả lỗi cũng chảy qua pipe (ví dụ để gom log), dùng |& (bash) hoặc 2>&1 |:

  python etl.py 2>&1 | tee run.log      # gom cả lỗi lẫn output cho tee
  

• Mã thoát của pipeline mặc định là mã thoát của lệnh cuối cùng, nên lỗi ở giữa dễ bị "nuốt". Trong script nên bật set -o pipefail để pipeline trả về mã lỗi nếu bất kỳ lệnh nào trong chuỗi hỏng.
• Cẩn thận với subshell: cat f | while read x; do n=$((n+1)); done — biến n bị mất sau vòng lặp vì while chạy trong subshell riêng. Cách né: dùng process substitution (bên dưới) hoặc for trên mảng.

tee — vừa in màn hình vừa ghi log

Đôi khi bạn muốn output vừa hiện lên màn hình để theo dõi, vừa lưu vào file để sau này soi lại. Pipe thông thường không làm được: ... | > file thì mất màn hình. tee giải bài toán này — nó đọc stdin, ghi ra file và đẩy tiếp ra stdout (như chữ T trong ống nước, tách dòng chảy làm hai):

python etl.py | tee run.log            # in ra màn hình VÀ ghi run.log
python etl.py 2>&1 | tee run.log       # gồm cả lỗi
python etl.py | tee -a run.log         # -a = append, không ghi đè
# Ghi ra nhiều file cùng lúc
generate_report | tee report_$(date +%F).txt summary.txt

Một mẹo hay: tee dùng để ghi file cần quyền cao khi bạn không muốn sudo cả pipeline:

echo "10.0.0.5 db.internal" | sudo tee -a /etc/hosts

Vì sudo echo ... >> /etc/hosts sẽ thất bại — redirection >> do shell thực hiện với quyền người dùng thường, không phải sudo. sudo tee thì đúng: chỉ tee chạy dưới quyền root và tự ghi file.

xargs — biến output thành tham số

Đây là công cụ hay bị hiểu nhầm nhất. Có một khác biệt then chốt:

• Pipe (|) đưa dữ liệu vào stdin của lệnh sau.
• Nhưng nhiều lệnh nhận đầu vào qua tham số dòng lệnh (argument), không đọc stdin. Ví dụ rm, cp, mkdir, gzip... — bạn không thể echo file.txt | rm, vì rm không đọc stdin.

xargs là cầu nối: nó đọc từ stdin rồi biến thành tham số cho lệnh bạn chỉ định.

# Tìm mọi file .log cũ hơn 30 ngày rồi XOÁ (rm nhận tham số qua xargs)
find /opt/airflow/logs -name '*.log' -mtime +30 | xargs rm

# Nén tất cả file .csv trong danh sách
ls *.csv | xargs gzip

# In số dòng của từng file .csv (tương đương wc -l *.csv nhưng qua stream)
find . -name '*.csv' | xargs wc -l

Các cờ quan trọng

-n N — nhóm N tham số cho mỗi lần gọi lệnh (mặc định xargs nhồi càng nhiều tham số càng tốt vào một lần gọi):

echo 1 2 3 4 5 | xargs -n 2 echo   # gọi echo 3 lần: "1 2", "3 4", "5"

-I {} — đặt chỗ chèn (placeholder) để dùng tham số ở giữa lệnh, không chỉ ở cuối:

# Đổi tên: thêm đuôi .bak cho mỗi file
ls *.conf | xargs -I {} cp {} {}.bak

# Với mỗi partition, chạy một lệnh load
cat partitions.txt | xargs -I {} psql -c "\\copy staging FROM '/data/{}.csv' CSV HEADER"

-P N — chạy song song N tiến trình. Đây là cách "đa luồng hoá" pipeline shell rất dễ, tận dụng nhiều CPU:

# Tải 100 URL, chạy 8 tiến trình song song
cat urls.txt | xargs -P 8 -n 1 wget -q

# Nén song song mọi file .csv bằng 4 luồng
find . -name '*.csv' | xargs -P 4 -n 1 gzip

-0 (số không) — an toàn với tên file có khoảng trắng. Mặc định xargs tách tham số theo khoảng trắng và xuống dòng, nên tên file như bao cao.csv sẽ bị tách nhầm thành hai. Kết hợp find -print0 (ngăn cách bằng ký tự NUL) với xargs -0 để xử lý an toàn mọi tên file:

# ĐÚNG kể cả tên file có dấu cách / ký tự lạ
find /data -name '*.csv' -print0 | xargs -0 rm

# Đếm dòng an toàn
find . -name '*.csv' -print0 | xargs -0 wc -l

Quy tắc vàng: hễ pipe từ find sang xargs, luôn dùng cặp -print0 / -0. Trong môi trường dữ liệu ngân hàng, file export thường có tên tiếng Việt hoặc chứa dấu cách — bỏ qua -0 là mầm mống của bug xoá nhầm file.

Command substitution $(...) — nhúng output vào lệnh

$(...) chạy lệnh bên trong, thay thế bằng stdout của nó (bỏ dấu xuống dòng cuối), rồi lồng vào chỗ đang gõ. Đây là cách "lấy kết quả một lệnh làm phần văn bản của lệnh khác":

# Đặt tên file theo ngày hiện tại
psql -c "SELECT ..." > export_$(date +%Y%m%d).csv

# Đếm số file rồi in ra
echo "Co $(ls *.csv | wc -l) file CSV can xu ly"

# Dùng kết quả làm biến
TODAY=$(date +%F)
ROW_COUNT=$(wc -l < data.csv)
echo "File co $ROW_COUNT dong, xu ly ngay $TODAY"

# Lồng nhiều lớp
tar -czf backup_$(date +%F)_$(hostname).tar.gz /data

Cú pháp cũ dùng backtick `...` cho kết quả tương tự nhưng khó lồng nhau và dễ nhầm, nên hãy luôn dùng $(...).

Process substitution <(...) — biến output thành một "file"

Khác với command substitution (chèn văn bản), process substitution <(...) biến output của một lệnh thành thứ trông như một file (thực chất là một đường dẫn /dev/fd/...), để đưa vào những lệnh đòi hỏi tên file chứ không đọc stdin. Cực hữu ích khi cần so sánh output của hai lệnh mà không tạo file tạm:

# So sánh danh sách bảng ở 2 database mà KHÔNG cần file tạm
diff <(psql db_prod -c '\dt') <(psql db_staging -c '\dt')

# So sánh cột id đã sort của 2 file, mỗi file được sort "on the fly"
comm -3 <(sort file_a.txt) <(sort file_b.txt)

# Né bẫy subshell: while read nhận từ process substitution, biến GIỮ được sau vòng lặp
count=0
while read line; do count=$((count+1)); done < <(grep ERROR app.log)
echo "So dong loi: $count"      # count vẫn còn giá trị

Dòng cuối là cách sửa lỗi "biến mất sau | while" đã nói ở phần pipe: vì while giờ chạy trong shell hiện tại (đọc từ < <(...)) chứ không trong subshell.

Here-doc <<EOF và here-string <<<

Khi cần nạp một khối văn bản nhiều dòng vào stdin của lệnh (thay vì đọc từ file), dùng here-document:

# Gửi nhiều câu SQL vào psql qua stdin
psql banking <<EOF
CREATE TEMP TABLE t AS SELECT 1;
SELECT count(*) FROM t;
EOF

# Tạo file cấu hình nhiều dòng
cat > /opt/app/config.yaml <<EOF
database:
  host: db.internal
  port: 5432
EOF

Mọi dòng giữa <<EOF và dòng chỉ chứa EOF được đẩy vào stdin. EOF chỉ là nhãn quy ước — dùng chữ gì cũng được. Hai biến thể quan trọng:

• <<'EOF' (đặt nhãn trong nháy đơn): tắt việc thay biến $VAR và $(...) bên trong — nội dung được giữ nguyên văn (literal). Dùng khi khối văn bản có $ mà bạn không muốn shell diễn giải.
• <<-EOF (thêm dấu -): cho phép thụt lề bằng TAB ở nhãn kết thúc và các dòng, để code trong script gọn đẹp.

# Có thay biến (không nháy)
NAME=NCB
cat <<EOF
Ngan hang: $NAME, ngay: $(date +%F)
EOF

# Giữ nguyên văn, KHÔNG thay biến (có nháy đơn)
cat <<'EOF'
Gia thanh la $PRICE va lenh la $(command)  # in ra y nguyen, khong dien giai
EOF

Here-string <<< là bản một dòng: đẩy một chuỗi thẳng vào stdin của lệnh:

grep 'ERROR' <<< "$log_line"           # đưa nội dung biến vào grep
bc <<< "1024 * 1024"                   # tính toán nhanh
awk -F',' '{print $2}' <<< "$csv_row"  # tách cột từ một chuỗi

Ghép lại: những pipeline dữ liệu thực tế

Giờ ta lắp mọi mảnh thành pipeline hoàn chỉnh — đúng kiểu bạn sẽ viết trong một BashOperator của Airflow hay một cron job.

1. Liệt kê file → lọc → xử lý → ghi kết quả. Tìm mọi file export CSV trong ngày, lọc ra file lớn hơn 0 byte, gộp lại (bỏ header lặp), ghi ra một file tổng, log lại quá trình:

find /data/export -name 'txn_*.csv' -size +0c -print0 \
  | xargs -0 cat \
  | awk 'NR==1 || $0 !~ /^txn_id,/' \
  | tee /data/merged/all_txn_$(date +%F).csv \
  | wc -l

2. Đếm dòng của nhiều CSV cùng lúc (kiểm tra số bản ghi từng file khớp kỳ vọng không):

# In số dòng từng file + tổng, an toàn với tên file có dấu cách
find /data/export -name '*.csv' -print0 | xargs -0 wc -l

# Chỉ tổng số dòng dữ liệu (trừ header mỗi file), dùng song song cho nhanh
find /data/export -name '*.csv' -print0 \
  | xargs -0 -P 4 -I {} sh -c 'echo $(($(wc -l < "{}") - 1))' \
  | awk '{s+=$1} END{print "Tong ban ghi:", s}'

3. Tải → giải nén → nạp vào database — pipeline ETL kinh điển, mỗi bước một công cụ:

# Tải file gz, giải nén ngay trong luồng, đẩy thẳng vào COPY của psql (không tạo file tạm)
curl -sS https://sftp.partner.vn/daily/txn_$(date +%Y%m%d).csv.gz \
  | gunzip \
  | psql banking -c "\\copy staging_txn FROM STDIN WITH (FORMAT csv, HEADER true)" \
  > load.log 2>&1

# Kiểm tra kết quả bằng mã thoát
if [ $? -eq 0 ]; then echo "Load OK"; else echo "Load LOI, xem load.log"; fi

Chú ý pipeline thứ 3: dữ liệu không bao giờ chạm đĩa dưới dạng file giải nén — curl tải, gunzip giải nén trong bộ nhớ, psql \copy FROM STDIN nạp thẳng. Với file vài GB, đây là khác biệt lớn về dung lượng đĩa tạm và tốc độ. Đây chính là "chất keo" pipe/redirection biến ba công cụ rời rạc thành một dòng ETL liền mạch.

Use case thực tế

Bối cảnh: mỗi đêm, đối tác gửi lên SFTP khoảng 40 file CSV giao dịch (txn_<mã_chi_nhánh>_YYYYMMDD.csv.gz), tổng ~6 GB sau giải nén. Nhiệm vụ của job đêm: tải tất cả, kiểm tra tính toàn vẹn (đếm dòng khớp file manifest), nạp vào bảng staging trong PostgreSQL, và ghi log đầy đủ để đội vận hành soi buổi sáng. Server chỉ có 4 vCPU và ổ đĩa tạm hạn chế nên không thể giải nén hết ra đĩa cùng lúc.

Bước 1 — Tải song song, ghi log tách stdout/stderr. Danh sách file nằm trong manifest.txt (mỗi dòng một tên file). Dùng xargs -P để tải 4 file cùng lúc:

DATE=$(date +%Y%m%d)
LOGDIR=/var/log/etl/$DATE
mkdir -p "$LOGDIR"

grep "_${DATE}\.csv\.gz$" manifest.txt \
  | xargs -P 4 -I {} sh -c 'curl -sS -o /data/in/{} https://sftp.partner.vn/daily/{}' \
  > "$LOGDIR/download.out" 2> "$LOGDIR/download.err"

Tách stdout/stderr ra hai file giúp buổi sáng chỉ cần wc -l download.err là biết ngay có lỗi tải nào không.

Bước 2 — Kiểm tra toàn vẹn: đếm dòng từng file, so với manifest. Manifest có cột filename,expected_rows. Ta đếm dòng thực tế (trừ header) rồi so:

find /data/in -name "*_${DATE}.csv.gz" -print0 \
  | xargs -0 -P 4 -I {} sh -c 'echo "$(basename {}) $(($(zcat "{}" | wc -l) - 1))"' \
  | tee "$LOGDIR/rowcounts.txt"
# rowcounts.txt: mỗi dòng "tên_file  số_dòng_thực_tế"

zcat giải nén ra stdout mà không tạo file — đúng ràng buộc đĩa hạn chế. tee vừa in để theo dõi vừa lưu để đối chiếu.

Bước 3 — Nạp vào staging, gom toàn bộ log. Với mỗi file, giải nén trong luồng và \copy thẳng vào PostgreSQL, gộp cả stdout lẫn stderr vào một log tổng có dấu thời gian:

find /data/in -name "*_${DATE}.csv.gz" -print0 \
  | xargs -0 -I {} sh -c 'echo "[$(date +%T)] load {}"; zcat "{}" | psql banking -c "\\copy staging_txn FROM STDIN WITH (FORMAT csv, HEADER true)"' \
  >> "$LOGDIR/load.log" 2>&1

Kết quả. Toàn bộ job gói trong ~15 dòng shell, chạy trong một BashOperator. Không script Python, không thư viện. Nhờ xargs -P 4, thời gian tải và nạp giảm gần 4 lần so với chạy tuần tự (từ ~40 phút xuống ~11 phút trong đo thực tế của đội). Nhờ zcat | psql theo luồng, đĩa tạm không bao giờ vượt kích thước một file .gz. Và nhờ tách > out 2> err cùng tee, sáng hôm sau chỉ cần liếc ba file log là biết đêm qua có sự cố gì. Đây là minh chứng vì sao pipe + redirection là kỹ năng nền của tự động hoá ETL — xem tiếp cách đóng gói thành script hoàn chỉnh ở Linux 8 — Shell scripting và cách đặt lịch ở Linux 7 — Cron & systemd.

Ghi nhớ

• Ba luồng chuẩn: stdin(0) đọc vào, stdout(1) kết quả, stderr(2) lỗi — stdout và stderr tách biệt, pipe chỉ mang stdout.
• Redirection: > ghi đè (nguy hiểm — truncate trước khi chạy), >> ghi thêm, < đọc file, 2> lỗi ra file, 2>&1 gộp lỗi theo stdout. Thứ tự > file 2>&1 mới đúng; đảo lại là sai. &> file là bản gộp rút gọn của bash.
• /dev/null là thùng rác: > /dev/null 2>&1 để chạy im lặng.
• Pipe | = triết lý Unix: công cụ nhỏ, ghép qua stdout→stdin, chạy song song theo dòng. Trong script bật set -o pipefail để không nuốt lỗi giữa chuỗi.
• tee = vừa in màn hình vừa ghi file (-a để append). sudo tee để ghi file cần quyền root trong pipeline.
• xargs biến stdin thành tham số; -n nhóm tham số, -I {} đặt chỗ chèn, -P N chạy song song. Luôn dùng find -print0 | xargs -0 để an toàn với tên file có dấu cách.
• $(...) chèn văn bản từ output; <(...) biến output thành file ảo (so sánh 2 lệnh, né bẫy subshell của while read).
• Here-doc <<EOF nạp khối nhiều dòng vào stdin; <<'EOF' giữ nguyên văn (không thay biến); here-string <<< đẩy một chuỗi vào stdin.