Linux cho DE 8 — Shell scripting tự động hoá pipeline

Vì sao data engineer vẫn cần viết bash tử tế

Sau bảy bài của series, bạn đã thạo điều hướng file, xử lý văn bản, pipe/redirect, quyền hạn, tiến trình và lập lịch. Bài cuối này gom tất cả lại thành thứ chạy được: một script bash tự động hoá một đầu việc thật. Trong một team dữ liệu ngân hàng như NCB, bash không biến mất dù đã có Airflow — nó là chất keo dán giữa các mảnh: một BashOperator gọi script tải file SFTP từ đối tác, một cron job dọn log, một entrypoint container kiểm tra cấu hình trước khi khởi động, một hook CI/CD build và deploy.

Vấn đề là 90% script bash ngoài đời được viết ẩu: không quote biến, không kiểm tra lỗi, rm -rf $DIR với $DIR rỗng, chạy nửa chừng thì hỏng và để lại đống rác. Với dữ liệu ngân hàng, một script "gần đúng" có thể load nhầm file cũ vào warehouse, hoặc xoá file nguồn trước khi load xong. Bài này dạy cách viết bash phòng thủ: chạy đúng thì tốt, chạy sai thì dừng ngay và báo, không bao giờ để lại trạng thái nửa vời.

Bài nối tiếp Linux 6 — Tiến trình & job và Linux 7 — Lập lịch & systemd. Khi script phức tạp lên, bạn sẽ thấy ranh giới để chuyển sang Python cho DE và Airflow — phần cuối bài bàn kỹ chuyện này.

Khung xương một script: shebang và quyền thực thi

Mọi script bắt đầu bằng shebang — dòng đầu tiên báo cho kernel biết dùng trình thông dịch nào:

#!/usr/bin/env bash

Dùng #!/usr/bin/env bash thay vì #!/bin/bash cứng vì env sẽ tìm bash trong PATH. Trên nhiều hệ thống (macOS, một số container), bash mới không nằm ở /bin. Đừng dùng #!/bin/sh nếu bạn xài tính năng của bash (mảng, [[ ]], local) — trên Debian/Ubuntu /bin/sh là dash, không có những thứ đó.

Một quy ước nhỏ nhưng đáng giá: đặt tên file .sh để người khác nhận ra đó là shell script, và bắt đầu bằng một khối comment mô tả script làm gì, cần biến môi trường nào, chạy ra sao. Trong môi trường ngân hàng nơi script sống nhiều năm và người viết đã chuyển team, một header rõ ràng tiết kiệm hàng giờ đọc lại.

Cấp quyền thực thi rồi chạy trực tiếp:

chmod +x load_pipeline.sh
./load_pipeline.sh          # kernel đọc shebang, gọi bash
# hoặc gọi tường minh, bỏ qua cả shebang lẫn quyền x:
bash load_pipeline.sh

Biến và quoting: nguồn gốc của 90% bug

Bash không có kiểu dữ liệu — mọi thứ là chuỗi. Gán biến không có khoảng trắng quanh dấu =:

name="customers"            # đúng
count=42
name = "customers"          # SAI: bash tưởng 'name' là lệnh

Luật vàng: luôn bọc biến trong dấu nháy kép "$var". Không quote thì bash tách chuỗi theo khoảng trắng (word splitting) và bung ký tự đại diện (globbing) — thảm hoạ khi tên file có dấu cách hoặc biến rỗng:

file="báo cáo tháng 6.csv"
rm $file       # SAI: xoá 3 file "báo", "cáo", "tháng"... rồi "6.csv"
rm "$file"     # đúng: xoá đúng 1 file

Giá trị mặc định với ${var:-default} — cực hữu ích để cho phép override qua biến môi trường:

DB_HOST="${DB_HOST:-localhost}"       # nếu DB_HOST chưa set/rỗng -> localhost
RETRIES="${RETRIES:-3}"
OUT_DIR="${OUT_DIR:?phải truyền OUT_DIR}"   # ':?' báo lỗi và thoát nếu chưa set

Mảng cho danh sách phần tử (đường dẫn, tên bảng):

tables=(customers accounts transactions)
echo "${tables[0]}"              # phần tử đầu: customers
echo "${tables[@]}"             # tất cả phần tử, mỗi phần tử là 1 word
echo "${#tables[@]}"            # số phần tử: 3
for t in "${tables[@]}"; do     # LUÔN "${arr[@]}" để giữ nguyên phần tử
  echo "load bảng $t"
done

Ghi nhớ: "${arr[@]}" (có nháy) giữ mỗi phần tử thành một từ riêng; "${arr[*]}" gộp tất cả thành một chuỗi ngăn cách bởi ký tự đầu của IFS.

Tham số dòng lệnh: từ $1 đến getopts

Tham số vị trí truy cập qua $1, $2...; $0 là tên script. Vài biến đặc biệt:

echo "Script $0 nhận $# tham số"
for arg in "$@"; do echo "arg: $arg"; done

Với script nghiêm túc có nhiều tuỳ chọn, dùng getopts để parse cờ -d, -t, -v:

usage() { echo "Dùng: $0 -d YYYY-MM-DD -t bảng [-v]" >&2; exit 2; }

verbose=0
while getopts ":d:t:v" opt; do
  case "$opt" in
    d) run_date="$OPTARG" ;;      # dấu ':' sau chữ nghĩa là cờ cần giá trị
    t) table="$OPTARG" ;;
    v) verbose=1 ;;
    \?) echo "Cờ lạ: -$OPTARG" >&2; usage ;;
    :)  echo "Cờ -$OPTARG cần giá trị" >&2; usage ;;
  esac
done
: "${run_date:?cần -d}" "${table:?cần -t}"    # bắt buộc phải có

Điều kiện: if, [[ ]], test và &&/||

Bash quyết định "đúng/sai" dựa trên exit code: 0 là thành công (true), khác 0 là thất bại (false) — ngược với trực giác của lập trình viên quen 1 = true.

Ưu tiên [[ ... ]] (bash keyword, an toàn hơn) thay cho [ ... ] (lệnh test cũ):

if [[ -f "$file" ]]; then echo "file tồn tại"; fi        # -f: là file thường
if [[ -d "$dir" ]]; then echo "là thư mục"; fi           # -d: là thư mục
if [[ -s "$file" ]]; then echo "file không rỗng"; fi     # -s: kích thước > 0
if [[ -z "$var" ]]; then echo "biến rỗng"; fi            # -z: chuỗi rỗng
if [[ -n "$var" ]]; then echo "biến có nội dung"; fi     # -n: khác rỗng

So sánh số và chuỗi dùng toán tử khác nhau — đây là chỗ hay nhầm:

# Số: -eq -ne -lt -le -gt -ge
if (( count > 100 )); then echo "nhiều dòng"; fi          # (( )) cho số học
if [[ "$rows" -lt 1 ]]; then echo "không có dòng nào"; fi

# Chuỗi: =  !=  <  >
if [[ "$env" == "prod" ]]; then echo "môi trường thật"; fi
if [[ "$name" == cust* ]]; then echo "khớp mẫu"; fi       # == hỗ trợ glob

Toán tử && (và) / || (hoặc) nối lệnh theo exit code — viết kiểm tra gọn:

mkdir -p "$OUT_DIR" && echo "đã tạo thư mục"      # chỉ echo nếu mkdir thành công
command -v psql >/dev/null || { echo "chưa cài psql" >&2; exit 1; }

Vòng lặp: for, while read, until

for duyệt danh sách:

for t in "${tables[@]}"; do echo "xử lý $t"; done
for i in $(seq 1 5); do echo "lần $i"; done

while read là cách chuẩn để đọc file theo dòng — không dùng for line in $(cat file) (sẽ vỡ khi dòng có khoảng trắng):

while IFS= read -r line; do        # IFS= giữ nguyên khoảng trắng đầu/cuối
  echo "dòng: $line"               # -r ngăn bash diễn giải dấu \
done < "$input_file"

until lặp cho tới khi điều kiện đúng — hợp cho chờ dịch vụ sẵn sàng:

attempt=0
until pg_isready -h "$DB_HOST" -q; do
  attempt=$((attempt + 1))
  (( attempt >= 30 )) && { echo "DB không lên sau 30 lần thử" >&2; exit 1; }
  sleep 2
done

Hàm, exit code và $?

Hàm gom logic lặp lại. Dùng local cho biến nội bộ để không rò rỉ ra global:

row_count() {
  local file="$1"
  wc -l < "$file"          # in kết quả ra stdout để hàm gọi "bắt" lại
}

n=$(row_count "$OUT_DIR/data.csv")

return trả về exit code (0-255), không phải giá trị. Muốn trả dữ liệu thì echo ra stdout rồi bắt bằng $(...). Biến $? giữ exit code của lệnh vừa chạy:

check_not_empty() {
  [[ -s "$1" ]]            # exit code của [[ ]] chính là return của hàm
}

if check_not_empty "$file"; then
  echo "OK ($?)"          # $? = 0
else
  echo "file rỗng, code=$?"
fi

An toàn: bộ ba set -euo pipefail

Đây là dòng quan trọng nhất trong bài. Đặt ngay sau shebang để bash chuyển từ "cố gắng chạy tiếp" sang "hỏng là dừng ngay":

#!/usr/bin/env bash
set -euo pipefail

Giải thích từng cờ:

• set -e (errexit): script dừng ngay khi một lệnh trả exit code khác 0 (mà không được xử lý bởi if/||). Không có nó, script chạy cd /khong/ton/tai rồi vẫn rm -rf * ở thư mục hiện tại — thảm hoạ.
• set -u (nounset): dùng biến chưa khai báo là lỗi và dừng. Bắt được lỗi gõ nhầm tên biến ($OUT_DIT thay vì $OUT_DIR) và ngăn rm -rf "$DIR/" biến thành rm -rf "/" khi DIR rỗng.
• set -o pipefail: mặc định exit code của pipe là của lệnh cuối. Với curl url | gunzip | psql, nếu curl hỏng nhưng psql thành công thì pipe vẫn báo 0 — che giấu lỗi. pipefail khiến pipe trả code của lệnh đầu tiên thất bại.

Vài lưu ý thực chiến: với set -e, khi bạn chủ đích cho phép một lệnh có thể fail, thêm || true hoặc bọc trong if. Và luôn dùng ${var:-} khi cần đọc biến có thể chưa set dưới set -u.

IFS (Internal Field Separator) kiểm soát bash tách từ như thế nào. Một khuôn mẫu an toàn phổ biến đặt IFS chỉ còn newline và tab, loại bỏ dấu cách khỏi việc tách từ:

IFS=$'\n\t'

trap: dọn dẹp và bắt lỗi

trap 'lệnh' TÍN_HIỆU chạy lệnh khi script nhận tín hiệu. Hai cái quan trọng nhất: EXIT (chạy mọi khi script kết thúc, dù thành công hay thất bại) và ERR (chạy khi có lệnh lỗi dưới set -e).

tmpdir="$(mktemp -d)"                        # thư mục tạm an toàn

cleanup() {
  rc=$?                                       # giữ exit code gốc
  rm -rf "$tmpdir"                            # luôn dọn rác
  (( rc != 0 )) && echo "Script thất bại, code=$rc" >&2
  return $rc
}
trap cleanup EXIT

trap 'echo "Lỗi ở dòng $LINENO" >&2' ERR      # in số dòng gây lỗi

Nhờ trap ... EXIT, dù script dừng ở giữa vì set -e, thư mục tạm vẫn được xoá. Đây là cách đảm bảo không để lại trạng thái nửa vời — nền tảng của tính idempotent. Lưu ý thứ tự: trap phải đặt sau khi biến tmpdir đã tồn tại, và trong cleanup ta bắt rc=$? ở dòng đầu tiên để giữ nguyên exit code gốc trước khi bất kỳ lệnh nào trong hàm ghi đè nó. Bỏ qua chi tiết này thì script luôn thoát với code 0 dù thực chất đã lỗi — một cái bẫy im lặng mà Airflow sẽ tưởng task thành công.

Kiểm tra tiền đề: fail nhanh, fail rõ

Trước khi làm việc thật, kiểm tra mọi giả định. Fail ngay từ đầu tốt hơn fail giữa chừng:

require_cmd() {
  command -v "$1" >/dev/null 2>&1 || {
    echo "Thiếu lệnh bắt buộc: $1" >&2; exit 127;
  }
}
require_cmd curl
require_cmd psql
require_cmd sha256sum

[[ -n "${DB_URL:-}" ]] || { echo "Chưa set DB_URL" >&2; exit 1; }
[[ -w "$OUT_DIR" ]]     || { echo "Không ghi được vào $OUT_DIR" >&2; exit 1; }

command -v là cách chuẩn kiểm tra lệnh có tồn tại (thay which, vốn không phải chuẩn POSIX). Triết lý ở đây: script nên kiểm tra mọi giả định của mình ngay đầu — lệnh cần có, biến môi trường cần set, thư mục cần ghi được, dịch vụ cần lên. Fail ở giây đầu tiên với thông điệp "thiếu psql" dễ chẩn đoán hơn nhiều so với fail ở giây thứ 300 giữa lúc load, khi đã tải xong file 2GB và không rõ trạng thái DB.

Logging: nói ra stderr, phân mức

Quy tắc: thông điệp log/lỗi ra stderr (>&2), chỉ dữ liệu "kết quả thật" ra stdout. Nhờ vậy script có thể được đưa vào pipe mà log không lẫn vào dữ liệu.

log()  { echo "[$(date +'%F %T')] [INFO]  $*" >&2; }
warn() { echo "[$(date +'%F %T')] [WARN]  $*" >&2; }
err()  { echo "[$(date +'%F %T')] [ERROR] $*" >&2; }

notify() {                        # gửi cảnh báo ra kênh chat nội bộ (minh hoạ)
  local status="$1" msg="$2"
  log "$status: $msg"
  # curl -s -X POST "$WEBHOOK_URL" --data "{\"text\":\"$status: $msg\"}" >/dev/null || true
}

Script pipeline hoàn chỉnh

Ghép tất cả lại thành một pipeline thực tế: tải file nguồn → kiểm tra checksum & không rỗng → load vào PostgreSQL → kiểm tra số dòng → thông báo. Có retry, có trap dọn dẹp, và idempotent (chạy lại nhiều lần cho cùng kết quả).

Lưu ý: script gọi psql để minh hoạ load dữ liệu, nhưng đây là script bash, không phải câu SQL trong sandbox — không đánh dấu chạy được.

#!/usr/bin/env bash
set -euo pipefail
IFS=$'\n\t'

# ---- Cấu hình (cho phép override qua env) ----
RUN_DATE="${1:-$(date +%F)}"
SRC_URL="${SRC_URL:?cần SRC_URL}"
DB_URL="${DB_URL:?cần DB_URL}"
OUT_DIR="${OUT_DIR:-/data/staging}"
RETRIES="${RETRIES:-3}"
TABLE="txn_daily"

tmpdir="$(mktemp -d)"
csv="$tmpdir/txn_${RUN_DATE}.csv"

log()  { echo "[$(date +'%F %T')] [INFO]  $*" >&2; }
err()  { echo "[$(date +'%F %T')] [ERROR] $*" >&2; }
notify() { log "NOTIFY $1: $2"; }   # thay bằng webhook thật khi triển khai

cleanup() {
  rc=$?
  rm -rf "$tmpdir"
  if (( rc != 0 )); then notify "FAILED" "pipeline $RUN_DATE lỗi (code=$rc)"; fi
  return $rc
}
trap cleanup EXIT
trap 'err "lỗi ở dòng $LINENO"' ERR

# ---- Tiền đề ----
for c in curl sha256sum psql; do
  command -v "$c" >/dev/null || { err "thiếu lệnh $c"; exit 127; }
done
mkdir -p "$OUT_DIR"

# ---- Idempotent: đã load ngày này thì bỏ qua ----
marker="$OUT_DIR/.done_${RUN_DATE}"
if [[ -f "$marker" ]]; then
  log "ngày $RUN_DATE đã load rồi, bỏ qua"
  exit 0
fi

# ---- Tải file có retry với backoff ----
download() {
  local attempt=1
  until curl -fsSL "$SRC_URL" -o "$csv"; do
    (( attempt >= RETRIES )) && { err "tải thất bại sau $RETRIES lần"; return 1; }
    log "tải lỗi, thử lại lần $((attempt+1)) sau $((attempt*5))s"
    sleep $((attempt * 5)); attempt=$((attempt + 1))
  done
}
download

# ---- Kiểm tra checksum & không rỗng ----
[[ -s "$csv" ]] || { err "file tải về rỗng"; exit 1; }
if [[ -n "${SRC_SHA256:-}" ]]; then
  echo "${SRC_SHA256}  ${csv}" | sha256sum -c - || { err "checksum sai"; exit 1; }
fi

# ---- Load vào DB trong 1 transaction (an toàn khi chạy lại) ----
rows_src=$(($(wc -l < "$csv") - 1))     # trừ dòng header
log "chuẩn bị load $rows_src dòng vào $TABLE"

psql "$DB_URL" -v ON_ERROR_STOP=1 <<SQL
BEGIN;
DELETE FROM ${TABLE} WHERE run_date = '${RUN_DATE}';   -- xoá dữ liệu ngày này nếu có
\copy ${TABLE} FROM '${csv}' WITH (FORMAT csv, HEADER true);
COMMIT;
SQL

# ---- Kiểm tra số dòng thực tế trong DB ----
rows_db=$(psql "$DB_URL" -tAc \
  "SELECT count(*) FROM ${TABLE} WHERE run_date='${RUN_DATE}'")
if [[ "$rows_db" -ne "$rows_src" ]]; then
  err "lệch số dòng: nguồn=$rows_src, db=$rows_db"; exit 1
fi

# ---- Đánh dấu hoàn tất & thông báo ----
touch "$marker"
notify "SUCCESS" "load $rows_db dòng vào $TABLE cho ngày $RUN_DATE"
log "hoàn tất"

Vài điểm thiết kế đáng chú ý:

• Idempotent hai lớp: marker file bỏ qua cả lần chạy nếu đã xong; và DELETE trước \copy khiến việc load lại một ngày không nhân đôi dữ liệu.
• Transaction: bọc DELETE/\copy trong BEGIN/COMMIT + ON_ERROR_STOP=1 — hỏng giữa chừng thì rollback, DB không rơi vào trạng thái nửa vời.
• Kiểm tra sau load: đếm dòng thực tế trong DB rồi đối chiếu với nguồn, phòng trường hợp load thiếu âm thầm.
• Retry có backoff: chỉ retry ở bước tải mạng, không retry bước load DB (tránh nhân đôi tác dụng phụ).

Luồng script với nhánh lỗi

Debug và kiểm tra chất lượng

• bash -x script.sh: chạy ở chế độ trace, in từng lệnh sau khi bung biến. Cách nhanh nhất để thấy "bash thật sự chạy gì".
• set -x / set +x: bật/tắt trace chỉ cho một đoạn nghi ngờ, tránh ngập log.
• shellcheck script.sh: linter tĩnh — bắt lỗi quoting, biến chưa dùng, so sánh sai, glob không mong muốn. Chạy shellcheck cho mọi script trước khi merge là kỷ luật đáng có; nó tìm ra phần lớn bug mà mắt thường bỏ sót. Có thể tích hợp vào CI (xem CI/CD).
• Thêm PS4='+ ${BASH_SOURCE}:${LINENO}: ' trước set -x để trace kèm tên file và số dòng.

Best practice và khi nào nên rời bash

Tóm tắt kỷ luật viết bash chắc chắn:

• Luôn set -euo pipefail và IFS=$'\n\t' ở đầu.
• Luôn quote "$var", "${arr[@]}"; dùng ${var:-default} cho biến tuỳ chọn.
• trap cleanup EXIT để không bao giờ để lại rác.
• Kiểm tra tiền đề, fail nhanh với thông điệp rõ ra stderr.
• Thiết kế idempotent; kiểm tra kết quả sau mỗi bước quan trọng.
• Chạy shellcheck trước khi đưa vào production.

Khi nào nên chuyển sang Python/Airflow? Bash tuyệt cho việc gọi lệnh, di chuyển file, glue ngắn (< ~100 dòng). Hãy cân nhắc rời bỏ khi:

Nguyên tắc thực dụng: khi script bash bắt đầu cần cấu trúc dữ liệu, unit test, hoặc điều phối nhiều task, đó là lúc bash đã chạm trần. Giữ bash làm lớp mỏng gọi công cụ; đẩy logic vào Python; đẩy điều phối vào Airflow.

Use case thực tế

Đối tác thẻ gửi file giao dịch hằng ngày qua SFTP; team DE của NCB cần tải, kiểm tra và nạp vào bảng staging trước 6h sáng để bộ phận đối soát bắt đầu làm việc. Trước đây một analyst tải tay rồi \copy — có hôm quên, có hôm load nhầm file hôm trước, một lần load hai lần khiến số liệu nhân đôi và bộ phận đối soát báo lệch 12.480 giao dịch.

Giải pháp: gói toàn bộ thành một script như ở trên, đặt trong BashOperator của Airflow chạy 5h sáng.

• Idempotent nhờ marker file + DELETE theo run_date: Airflow tự động retry task 3 lần, chạy lại không nhân đôi dữ liệu.
• Kiểm tra tiền đề & không rỗng: hôm đối tác gửi file 0 byte, script fail ở bước checksum, gửi cảnh báo Slack lúc 5h03 thay vì để analyst phát hiện lúc 8h.
• Đối chiếu số dòng: bắt được một lần file bị cắt cụt do lỗi truyền — nguồn khai 40.000 dòng nhưng DB chỉ nhận 31.204, script dừng và báo ngay.
• trap dọn tmp: file tạm luôn được xoá kể cả khi fail, tránh đầy đĩa /data/staging.

Kết quả: sự cố nạp trùng và nạp thiếu về 0 sau ba tháng; thời gian sẵn sàng dữ liệu ổn định trước 5h30 sáng.

Ghi nhớ

• Mở đầu mọi script: #!/usr/bin/env bash + chmod +x, rồi set -euo pipefail và IFS=$'\n\t'.
• set -e dừng khi lỗi, set -u bắt biến chưa set, pipefail bắt lỗi trong pipe — ba cờ này biến bash "ẩu" thành "phòng thủ".
• Luôn quote "$var" và "${arr[@]}"; dùng ${var:-default} / ${var:?msg} cho biến tuỳ chọn/bắt buộc.
• Số dùng (( )) và -eq/-lt/...; chuỗi dùng [[ ]] với ==/!=. Đừng lẫn lộn hai bộ toán tử.
• Đọc file theo dòng bằng while IFS= read -r line; do ...; done < file, không dùng for trên cat.
• trap cleanup EXIT đảm bảo dọn dẹp và không để lại trạng thái nửa vời; trap ... ERR giúp lộ dòng gây lỗi.
• Thiết kế idempotent (marker file, DELETE-then-load trong transaction) + retry ở bước mạng + kiểm tra số dòng sau load.
• Log ra stderr >&2, giữ stdout cho dữ liệu thật; luôn chạy shellcheck và bash -x khi debug.
• Khi cần cấu trúc dữ liệu, test, hay điều phối nhiều task có retry/SLA — chuyển sang Python / Airflow.