Nền tảng lập trình & Công cụProgramming & Tooling Foundations

Thuộc bộ kiến thức Data Engineer Roadmap.

Tổng quan

Data engineering, xét cho cùng, là một ngành software engineering áp dụng vào việc di chuyển và định hình dữ liệu. Trước khi nói đến pipeline, warehouse hay orchestration tool, một data engineer cần một bộ kỹ năng lập trình nhỏ nhưng được chọn lọc kỹ, xuất hiện trong hầu như mọi tác vụ hàng ngày: viết SQL để query và transform dữ liệu ở quy mô lớn, viết Python để kết nối các hệ thống lại với nhau và xử lý dữ liệu bên ngoài database, sử dụng Linux thành thạo vì đó là nơi gần như toàn bộ hạ tầng dữ liệu vận hành, và dùng Git một cách kỷ luật vì code pipeline vẫn là code và xứng đáng được đối xử nghiêm túc như bất kỳ production software nào.

Note này cố tình mang tính nền tảng — đây là lớp mà mọi phần khác trong roadmap (modeling, warehousing, ETL/ELT, orchestration, streaming) được xây dựng lên trên. Có được sự thành thạo SQL và Python, thoải mái với shell Linux, và version control kỷ luật trước tiên, thì phần còn lại của roadmap sẽ chỉ là áp dụng những kỹ năng này vào các bài toán ngày càng lớn và chuyên biệt hơn, thay vì phải học lập trình từ đầu. Chúng ta cũng sẽ xem xét khi nào các ngôn ngữ khác (Scala, Java, Go) có ý nghĩa, tại sao khối lượng dữ liệu thay đổi cách ta suy luận về độ phức tạp thuật toán, và tại sao ngành công nghiệp ngày càng ưu tiên các công cụ declarative hơn là script imperative để mô tả pipeline.

Kiến thức nền tảng

SQL là kỹ năng cốt lõi không thể thương lượng

Nếu một data engineer chỉ được thành thạo một ngôn ngữ, đó phải là SQL. Mọi relational store và hầu hết analytical store — Postgres, MySQL, Snowflake, BigQuery, Redshift, Databricks SQL, DuckDB — đều nói SQL, và ngay cả các công cụ “NoSQL”-adjacent (Spark SQL, Trino, Presto, Athena) cũng hội tụ về SQL như query interface cho analytics. SQL là declarative: bạn mô tả kết quả mình muốn, còn query planner sẽ quyết định cách đạt được nó — đây chính là mức abstraction mà phần lớn công việc transform dữ liệu nên hoạt động ở đó.

Một bộ kỹ năng SQL ở mức production đi xa hơn nhiều so với SELECT ... WHERE:

• Join ở quy mô lớn — hiểu loại join (inner, left, full, semi, anti) cùng cardinality và phân bố của join key ảnh hưởng thế nào đến cả correctness lẫn chi phí. Một join tức thời trên 10 nghìn dòng có thể làm tràn bộ nhớ hoặc phải spill ra disk trên 10 tỷ dòng nếu key bị skew hoặc không được index/partition.
• Window functions — tính running total, ranking, moving average, và so sánh lag/lead mà không làm gộp các dòng lại qua GROUP BY. Đây là tính năng SQL mạnh nhất cho analytics vì nó cho phép giữ nguyên chi tiết cấp dòng (row-level) trong khi vẫn thêm ngữ cảnh aggregate.
• CTE (Common Table Expression) — mệnh đề WITH đặt tên cho các tập kết quả trung gian, giúp các bước transform nhiều tầng trở nên dễ đọc và dễ test thay vì một mớ subquery lồng nhau. Recursive CTE còn cho phép duyệt các cấu trúc phân cấp (org chart, bill-of-materials) hoàn toàn bằng SQL thuần.
• Kiến thức nền về query optimization — biết đọc plan EXPLAIN / EXPLAIN ANALYZE, biết khi nào một query sẽ dùng index thay vì full scan, hiểu partition pruning và predicate pushdown trong các engine columnar/distributed, và biết rằng SELECT * cùng DISTINCT/ORDER BY không cần thiết là những nguồn lãng phí công sức phổ biến, có thể tránh được.

Ví dụ thực hành: CTE + window function

Xét một bảng chứa doanh thu theo ngày của từng cửa hàng, và cần tính cho mỗi cửa hàng running total doanh thu cùng thứ hạng của từng ngày theo doanh thu:

WITH daily_revenue AS (
    SELECT
        store_id,
        order_date,
        SUM(order_amount) AS revenue
    FROM orders
    WHERE order_date >= DATE '2026-01-01'
    GROUP BY store_id, order_date
)
SELECT
    store_id,
    order_date,
    revenue,
    SUM(revenue) OVER (
        PARTITION BY store_id
        ORDER BY order_date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS running_total_revenue,
    RANK() OVER (
        PARTITION BY store_id
        ORDER BY revenue DESC
    ) AS revenue_rank_in_store
FROM daily_revenue
ORDER BY store_id, order_date;

CTE (daily_revenue) thực hiện phép aggregation một lần và đặt cho nó một cái tên dễ đọc; query bên ngoài sau đó áp dụng hai window function độc lập lên trên nó. SUM(...) OVER (PARTITION BY store_id ORDER BY order_date ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) tạo ra một running total được reset theo từng cửa hàng, và RANK() OVER (PARTITION BY store_id ORDER BY revenue DESC) xếp hạng doanh thu mỗi ngày trong phạm vi cửa hàng đó — tất cả mà không làm gộp mất grain theo ngày qua GROUP BY. Pattern này (CTE để định hình các dòng cơ sở, window function để thêm ngữ cảnh running/ranking) là một trong những building block phổ biến nhất trong SQL analytics cũng như trong model dbt.

Python cho data engineering

Python thống trị data engineering vì một lý do đơn giản: đây là ngôn ngữ chung xuyên suốt mọi lớp của modern data stack. Việc khám phá dữ liệu (exploration) và prototyping diễn ra trong pandas trên laptop; cùng một logic đó, khi vượt quá khả năng của một máy đơn, được viết lại (thường gần như 1:1) bằng PySpark để xử lý phân tán; và lớp orchestration lập lịch cũng như giám sát toàn bộ — Apache Airflow, Dagster, Prefect — bản thân nó cũng Python-native, nghĩa là pipeline được định nghĩa hoàn toàn bằng code Python (hoặc các hàm được decorate bằng Python), không phải một DSL độc quyền. Tính liên tục này — cùng một ngôn ngữ từ notebook khám phá đến DAG production — chính là điều khiến Python trở thành mô liên kết (connective tissue) của ngành, còn quan trọng hơn bất kỳ thư viện đơn lẻ nào.

Các thư viện chủ chốt mà data engineer thường xuyên dùng đến:

Thư viện	Mục đích
pandas	Thao tác dữ liệu dạng bảng trong bộ nhớ — exploration, transform quy mô nhỏ-vừa, join/aggregation nhanh
pyarrow	Định dạng columnar in-memory (Apache Arrow) và I/O Parquet/CSV nhanh; là nền tảng bên dưới pandas 2.x, tương tác với Spark, và DuckDB
requests	Gọi HTTP đến REST API — phần lớn logic “extract” trong ETL là requests.get() đến một API của vendor nào đó
sqlalchemy	Bộ công cụ SQL và ORM độc lập với database — kết nối và ghi dữ liệu vào relational database từ code Python
pyspark	API Python cho Apache Spark — xử lý phân tán khi dữ liệu không còn vừa trên một máy

Ví dụ thực hành: CSV → transform → database

Một script “extract, transform, load” nhỏ nhưng tiêu biểu — đọc một file CSV, làm sạch và tái định hình bằng pandas, rồi ghi vào một bảng Postgres bằng SQLAlchemy:

import pandas as pd
from sqlalchemy import create_engine

# Extract: read the raw CSV
df = pd.read_csv("daily_orders.csv", parse_dates=["order_date"])

# Transform: clean, cast, and derive columns
df = df.dropna(subset=["store_id", "order_amount"])
df["order_amount"] = df["order_amount"].astype("float64")
df["order_date"] = df["order_date"].dt.date
df["revenue_bucket"] = pd.cut(
    df["order_amount"],
    bins=[0, 50, 200, float("inf")],
    labels=["low", "medium", "high"],
)

# Load: write to a Postgres table, replacing existing data for idempotency
engine = create_engine("postgresql+psycopg2://etl_user:secret@localhost:5432/warehouse")
df.to_sql(
    "stg_orders",
    con=engine,
    schema="staging",
    if_exists="replace",
    index=False,
    chunksize=10_000,
    method="multi",
)

Ngay cả một script nhỏ như thế này cũng minh họa những mối quan tâm thực tế: chunksize và method="multi" gộp các lệnh insert thành batch thay vì từng dòng một (rất quan trọng khi có nhiều hơn vài nghìn dòng), if_exists="replace" khiến việc chạy lại (rerun) trở nên idempotent thay vì bị nhân đôi dữ liệu, và bước transform chính là nơi các quy tắc chất lượng dữ liệu (loại bỏ null, ép kiểu, phân nhóm) được áp dụng trước khi dữ liệu đến warehouse.

Các ngôn ngữ khác: Scala, Java, Go

Python không phải là ngôn ngữ duy nhất một data engineer sẽ gặp, và biết khi nào mỗi ngôn ngữ còn lại có ý nghĩa sẽ tiết kiệm thời gian:

• Scala — Apache Spark được viết bằng Scala, và Scala vẫn là ngôn ngữ “gốc” của Spark: API Scala thường nhận tính năng mới trước tiên, tránh được chi phí serialization Python-JVM (không cần round-trip qua py4j cho UDF), và là lựa chọn mặc định trên các team nặng về JVM, coi trọng static typing và throughput thô trên các job Spark rất lớn. Dùng Scala khi team đã có chuyên môn JVM sẵn, hoặc khi UDF trong một job PySpark trở thành điểm nghẽn hiệu năng.
• Java — xương sống truyền thống của hệ sinh thái big-data: Hadoop (HDFS, MapReduce, YARN) và phần lớn hệ sinh thái Kafka (bản thân Kafka, Kafka Connect, các thư viện client Kafka Streams) đều được viết bằng Java, và nhiều doanh nghiệp vẫn vận hành các dịch vụ dữ liệu và connector dựa trên Java. Dùng Java khi bảo trì hoặc mở rộng hạ tầng Hadoop/Kafka hiện có, hoặc viết custom Kafka Connect connector/consumer trong môi trường chuẩn hóa JVM.
• Go — ngày càng được dùng cho các công cụ dữ liệu hiệu năng cao, footprint thấp: CLI utility, connector tùy chỉnh, dịch vụ ingestion/streaming nhẹ, và các công cụ liên quan đến infrastructure (nhiều công cụ data/observability trong CNCF được viết bằng Go). Thời gian khởi động nhanh, binary tĩnh đơn lẻ, và các primitive concurrency mạnh (goroutine/channel) khiến Go hấp dẫn cho việc xây dựng các dịch vụ nhỏ, nhanh xung quanh pipeline — ví dụ một Kafka consumer phân phối dữ liệu đến nhiều sink — nơi mà chi phí runtime hay độ phức tạp đóng gói của Python sẽ là bất lợi.

Với hầu hết các vai trò data engineering, SQL và Python đã bao phủ hơn 90% công việc hàng ngày; Scala, Java, và Go trở nên liên quan khi stack cụ thể của team (Spark nặng, Hadoop/Kafka legacy, hoặc công cụ đòi hỏi hiệu năng cao) yêu cầu.

Kiến thức Linux nền tảng cho data engineer

Gần như toàn bộ hạ tầng dữ liệu — database, cụm Spark/Hadoop, scheduler Airflow, broker Kafka, VM và container trên cloud — đều chạy trên Linux. Ngay cả khi laptop hàng ngày của bạn là macOS hay Windows, các server, container, và CI runner thực thi pipeline của bạn đều là Linux, nên sự thoải mái với shell không phải là tùy chọn.

Các công cụ thiết yếu dùng hàng ngày:

Lệnh / khái niệm	Dùng để làm gì
grep	Tìm kiếm text nhanh — tìm một chuỗi lỗi trên nhiều file log (grep -r "ERROR" /var/log/pipeline/)
awk	Xử lý text theo trường (field) — trích xuất/tổng hợp cột nhanh trên dữ liệu CSV hoặc dòng log (awk -F',' '{sum+=$3} END{print sum}' data.csv)
sed	Stream editing — thay thế và dọn dẹp tại chỗ (sed -i 's/NULL/\\N/g' export.csv)
cron	Lập lịch job theo thời gian (crontab -e) — cách lập lịch script nguyên thủy, vẫn phổ biến trước/song song với Airflow
systemd	Quản lý service (systemctl status, journalctl -u) — chạy một tiến trình dài hạn (một Kafka consumer, một API) như một service được quản lý, tự khởi động lại khi lỗi
Shell scripting (bash)	Glue code — nối chuỗi các công cụ CLI, kiểm tra exit code, bọc một job Python với các bước trước/sau

Một ví dụ nhanh kết hợp vài công cụ trên — đếm số dòng lỗi theo từng giờ từ một file log:

grep "ERROR" pipeline.log \
  | awk '{print $1, $2}' \
  | cut -c1-13 \
  | sort \
  | uniq -c \
  | sort -rn

Kiểu one-liner này — ghép nối vài công cụ nhỏ, có thể kết hợp (composable) với nhau — thường nhanh hơn viết một script Python cho việc kiểm tra log hoặc dữ liệu nhanh, ad hoc, và phản xạ này chuyển trực tiếp sang việc debug pipeline chạy trên server từ xa nơi không có sẵn một IDE đầy đủ.

Git và GitHub cho code pipeline

Code pipeline — các phép transform SQL, model dbt, DAG Airflow, job Spark, infrastructure-as-code — là production software và xứng đáng có cùng kỷ luật version control như bất kỳ codebase ứng dụng nào, chứ không phải các file “final_v2.sql” tùy tiện. Git theo dõi mọi thay đổi với lịch sử đầy đủ, cho phép làm việc song song qua branch, và (kết hợp với GitHub/GitLab) cung cấp pull request để review code trước khi bất cứ thứ gì đến được production.

Một vài điểm đặc thù cho repo pipeline dữ liệu:

• Chiến lược branching — một mô hình trunk-based nhẹ nhàng hoặc GitHub-flow (feature branch tồn tại ngắn hạn tách từ main, merge qua PR) phù hợp với hầu hết data team hơn là các branch GitFlow tồn tại lâu dài, vì các thay đổi pipeline (một cột mới, sửa một join, thay đổi schema) thường nhỏ và cần đến production nhanh chóng, an toàn.
• Văn hóa review PR — review một thay đổi SQL/dbt nghĩa là kiểm tra tác động lên query plan (join này có làm bùng nổ số dòng không?), khả năng tương thích ngược (thay đổi này có làm hỏng các consumer downstream của một bảng không?), và test (model dbt có test not_null/unique không, DAG có unit test cho hàm transform không?). Đối xử với thay đổi schema nghiêm túc như thay đổi API contract.
• CI cho pipeline — lint SQL, chạy dbt build --select state:modified+ với các model đã thay đổi, và chạy validate DAG Airflow (airflow dags list-import-errors) trong CI giúp bắt lỗi pipeline hỏng trước khi merge, thay vì lúc 3 giờ sáng ở production.

Để tìm hiểu sâu hơn về cơ chế Git — object model, chiến lược branching, kỷ luật commit, và thực hành code review — xem ../backend/en/04-version-control-and-collaboration.md; mọi nội dung ở đó áp dụng trực tiếp cho một repo pipeline.

Cấu trúc dữ liệu & thuật toán trong data engineering

Các buổi phỏng vấn software engineering nói chung thường nhấn mạnh phân tích Big-O như một mục tiêu tự thân; với data engineer, cùng một trực giác đó lại quan trọng vì một lý do rất thực tế: khối lượng dữ liệu thay đổi cách tính toán. Một vòng lặp lồng nhau O(n^2) chạy tức thời trên 1.000 dòng có thể trở thành một job chạy nhiều giờ (hoặc không thể chạy được) trên 100 triệu dòng, và khác biệt giữa cách tiếp cận O(n) và O(n log n) có thể là ranh giới giữa việc pipeline hoàn thành trong khung giờ đã lên lịch hay không hoàn thành được chút nào.

Các cấu trúc thường xuyên xuất hiện:

• Hash map — cơ chế đứng sau hash join (chiến lược join mặc định của hầu hết engine cho equality join), khử trùng lặp (deduplication), và groupby aggregation; hiểu rằng hash join có độ phức tạp O(n + m) so với O(n * m) của nested-loop join giải thích tại sao thứ tự join và lựa chọn key lại quan trọng đến vậy trên các bảng lớn.
• Cây (tree) — B-tree là nền tảng của hầu hết database index (lý do một lookup có index là O(log n), không phải O(n)); các sơ đồ partitioning dạng cây (phân cấp theo ngày/vùng, thư mục partition kiểu Hive) cho phép query engine cắt bỏ (prune) toàn bộ nhánh dữ liệu mà nó không cần quét.
• Cấu trúc đã sắp xếp / pattern merge — nhiều chiến lược shuffle và merge-sort join phân tán dựa vào việc sắp xếp rẻ hơn một phép tích Descartes đầy đủ; hiểu trực giác merge-sort giải thích tại sao sort-merge join được ưu tiên hơn hash join khi dữ liệu đã được partition/sort sẵn hoặc không vừa trong bộ nhớ.

Bài học thực tế: trước khi tối ưu code, hãy tự hỏi “độ phức tạp Big-O của thao tác này là gì, và nó scale như thế nào với khối lượng dữ liệu thực tế của chúng ta, chứ không phải một test fixture nhỏ?” Một join, một groupby, hay một full-table scan mà một senior data engineer theo bản năng sẽ kiểm tra lại kỹ ở quy mô lớn — chính là kiểu suy luận này.

Cách tiếp cận declarative và imperative

Một chủ đề lặp lại xuyên suốt modern data stack là sự ưu tiên các công cụ declarative hơn imperative ở bất cứ đâu công việc cho phép:

Cách tiếp cận	Ví dụ	Bạn mô tả…
Declarative	SQL, dbt, Terraform	Trạng thái đích mong muốn; công cụ tự tìm cách đạt được nó
Imperative	Script Python thô, logic DAG Airflow viết tay	Chuỗi bước chính xác cần thực thi

SQL khai báo hình dạng của kết quả và để lại chiến lược thực thi cho query planner. dbt khai báo SQL của một model cùng các dependency của nó (qua ref()), và dbt sẽ tự tìm ra thứ tự thực thi (DAG) và, ở mỗi lần chạy, chỉ cần áp dụng phần chênh lệch. Terraform khai báo trạng thái infrastructure mong muốn, và terraform plan/apply tính toán rồi áp dụng phần diff. Đây là lý do các công cụ declarative thường idempotent theo mặc định — chạy lại khi không có gì thay đổi là một no-op — và tại sao chúng dễ suy luận, dễ review, dễ rollback hơn: phần chênh lệch giữa hai trạng thái của file mô tả trạng thái mong muốn chính là thay đổi, không có sự mơ hồ ẩn giấu kiểu “lần chạy trước để lại gì”.

Code imperative (một script Python, hay logic task của một DAG Airflow) vẫn có vị trí thiết yếu — control flow của orchestration, gọi API bên ngoài, phân nhánh điều kiện phức tạp, và bất cứ điều gì một DSL declarative không thể diễn đạt gọn gàng — nhưng nó đẩy trách nhiệm về idempotency và error recovery sang cho kỹ sư. Một pipeline imperative viết tốt sẽ chủ động xây dựng tính idempotent (upsert thay vì insert, kiểm tra điều kiện tiên quyết, các pattern kiểu if_exists="replace" như trong ví dụ trước); một pipeline declarative có được phần lớn điều đó miễn phí. Xu hướng ngành ưu tiên SQL/dbt cho logic transform và Terraform cho infrastructure, trong khi giữ Python/Airflow ở lớp orchestration và glue mà chúng thực sự làm tốt, phản ánh chính sự đánh đổi này.

Khái niệm chính

Lựa chọn ngôn ngữ và công cụ

Bảng dưới đây tóm tắt nên dùng công cụ nào và khi nào, xuyên suốt mọi nội dung ở trên. Xem ./08-etl-elt-and-data-pipelines.md để biết các công cụ này kết hợp thành pipeline hoàn chỉnh ra sao, và ./09-workflow-orchestration.md để biết công cụ orchestration lập lịch và giám sát chúng thế nào.

Ngôn ngữ / công cụ	Trường hợp sử dụng chính trong data engineering	Khi nào nên dùng
SQL	Query và transform dữ liệu trong warehouse/lake; lớp transform của ELT	Luôn luôn — mặc định cho bất kỳ transform nào có thể diễn đạt bằng query
Python (pandas)	Exploration, prototyping, transform quy mô nhỏ-vừa trong bộ nhớ	Phân tích cục bộ, notebook, dữ liệu vừa với bộ nhớ trên một máy
Python (PySpark)	Xử lý phân tán dữ liệu quá lớn cho một máy	Transform batch quy mô nhiều GB/TB, join qua các dataset khổng lồ
Python (Airflow/Dagster/Prefect)	Orchestration — lập lịch, quản lý dependency, giám sát pipeline	Điều phối pipeline nhiều bước với dependency, retry, cảnh báo
dbt (SQL + Jinja)	Lớp transform declarative trên nền warehouse	Modeling và test logic transform với version control và CI
Scala	Job Spark hiệu năng cao, native trên các team nặng về JVM	Job nhiều UDF Spark mà chi phí serialization của PySpark gây hại
Java	Bảo trì/mở rộng Hadoop, Kafka Connect, tích hợp client Kafka	Hạ tầng big-data legacy, môi trường chuẩn hóa JVM
Go	Công cụ và dịch vụ dữ liệu hiệu năng cao, footprint thấp	Connector tùy chỉnh, dịch vụ streaming nhẹ, công cụ CLI
Terraform	Infrastructure as code declarative cho tài nguyên nền tảng dữ liệu	Cấp phát warehouse, cluster, IAM, storage bucket có thể tái tạo được
Bash / Linux CLI	Glue code, kiểm tra nhanh, lập lịch, quản lý service	Kiểm tra log/dữ liệu ad hoc, bọc job, lập lịch cron, service systemd
Git / GitHub	Version control và review cho tất cả những gì kể trên	Luôn luôn — mọi artifact pipeline thuộc về một repo

Best Practices

1. Học SQL thật sâu trước mọi thứ khác. Window function, CTE, và query plan mang lại giá trị trong hầu như mọi tác vụ bạn sẽ làm; hãy coi đây là kỹ năng cốt lõi, không phải phụ lục cho “lập trình thật sự”.
2. Prototype trong pandas, scale lên PySpark, giữ logic dễ nhận ra. Viết code pandas khám phá với tư duy hướng đến việc nó sẽ được chuyển sang Spark như thế nào giúp tránh phải viết lại đau đớn về sau.
3. Đẩy logic transform vào SQL/dbt declarative khi có thể. Dành Python/Airflow cho orchestration, control flow, và những gì SQL thực sự không diễn đạt được — đừng viết lại một JOIN trong vòng lặp Python.
4. Chủ động làm cho code imperative idempotent. Upsert thay vì insert, các pattern if_exists="replace"/MERGE, và kiểm tra điều kiện tiên quyết rõ ràng — giả định rằng mọi job sẽ được chạy lại, vì nó sẽ được chạy lại.
5. Đọc query plan trước khi tối ưu mù quáng. EXPLAIN ANALYZE (hoặc query plan trong Spark UI) cho biết một join đang broadcast, shuffle, hay scan — đoán mò lãng phí thời gian kỹ thuật.
6. Đối xử với code pipeline như production code. PR, review, lint/test trong CI, và một chiến lược branching thật sự — một model SQL bị hỏng là một sự cố (incident), không phải một lỗi gõ phím.
7. Chủ động chọn Scala/Java/Go, không phải theo mặc định. Ghép ngôn ngữ với điểm nghẽn thực tế (hiệu năng UDF Spark, hạ tầng Kafka/Hadoop có sẵn, một dịch vụ throughput cao nhẹ) thay vì vì mới lạ.
8. Giữ nền tảng Linux luôn sắc bén. One-liner grep/awk/sed và sự thoải mái với systemd/cron tiết kiệm thời gian debug một pipeline từ xa nhiều hơn hẳn so với việc dùng một IDE đầy đủ.
9. Tư duy Big-O ở đúng quy mô dữ liệu thực tế. Test fixture 1.000 dòng sẽ không bộc lộ một join O(n^2) sụp đổ ở 100 triệu dòng — hãy suy luận về tốc độ tăng trưởng, không chỉ tính đúng đắn.

Tài liệu tham khảo

• roadmap.sh — Data Engineer
• PostgreSQL — Window Functions
• PostgreSQL — WITH Queries (CTEs)
• pandas documentation
• Apache Arrow / PyArrow documentation
• Apache Spark — PySpark documentation
• Apache Airflow documentation
• dbt documentation
• ../backend/en/04-version-control-and-collaboration.md

WITH daily_revenue AS (
    SELECT
        store_id,
        order_date,
        SUM(order_amount) AS revenue
    FROM orders
    WHERE order_date >= DATE '2026-01-01'
    GROUP BY store_id, order_date
)
SELECT
    store_id,
    order_date,
    revenue,
    SUM(revenue) OVER (
        PARTITION BY store_id
        ORDER BY order_date
        ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW
    ) AS running_total_revenue,
    RANK() OVER (
        PARTITION BY store_id
        ORDER BY revenue DESC
    ) AS revenue_rank_in_store
FROM daily_revenue
ORDER BY store_id, order_date;

import pandas as pd
from sqlalchemy import create_engine

# Extract: read the raw CSV
df = pd.read_csv("daily_orders.csv", parse_dates=["order_date"])

# Transform: clean, cast, and derive columns
df = df.dropna(subset=["store_id", "order_amount"])
df["order_amount"] = df["order_amount"].astype("float64")
df["order_date"] = df["order_date"].dt.date
df["revenue_bucket"] = pd.cut(
    df["order_amount"],
    bins=[0, 50, 200, float("inf")],
    labels=["low", "medium", "high"],
)

# Load: write to a Postgres table, replacing existing data for idempotency
engine = create_engine("postgresql+psycopg2://etl_user:secret@localhost:5432/warehouse")
df.to_sql(
    "stg_orders",
    con=engine,
    schema="staging",
    if_exists="replace",
    index=False,
    chunksize=10_000,
    method="multi",
)

grep "ERROR" pipeline.log \
  | awk '{print $1, $2}' \
  | cut -c1-13 \
  | sort \
  | uniq -c \
  | sort -rn