Automation cho Data Cleansing: Workflow tự động chuẩn hóa định dạng và Deduplication

Tóm tắt nội dung chính
– Vấn đề thực tế: Dữ liệu rối, định dạng không đồng nhất, bản ghi trùng lặp gây mất thời gian và sai sót trong báo cáo.
– Giải pháp: Xây dựng workflow automation cho việc chuẩn hoá định dạng và deduplication bằng các công cụ ETL/No‑Code (nếu cần) hoặc script Python/SQL.
– Quy trình chi tiết: Thu thập nguồn, chuẩn hoá, phát hiện trùng lặp, hợp nhất, lưu trữ kết quả.
– Template: Một mẫu quy trình có thể copy‑paste cho hầu hết các dự án.
– Lỗi phổ biến & cách sửa: Nhầm lẫn key, không xử lý null, thời gian chạy quá lâu.
– Scale: Sử dụng batch processing, phân tán (Spark, Airflow) và queue (Kafka).
– Chi phí thực tế: Từ 5 tr – 30 tr/tháng tùy vào khối lượng và công cụ.
– Số liệu trước‑sau: Thời gian xử lý giảm 70 %, chi phí nhân công giảm 60 %, độ chính xác tăng lên 98 %.
– FAQ: Các câu hỏi thường gặp về định dạng, deduplication, bảo mật.
– Giờ tới lượt bạn: Áp dụng ngay workflow mẫu, đo lường KPI, tối ưu hoá liên tục.

1. Vấn đề thật mà mình và khách hay gặp mỗi ngày

Trong môi trường doanh nghiệp Việt, dữ liệu thường xuất hiện ở ba dạng chính:

⚡ Lưu ý: Khi dữ liệu không đồng nhất, mỗi lần báo cáo sẽ mất 2‑3 giờ để “điều chỉnh tay”. Đối với một công ty vừa và nhỏ (SME) với 5 người phân tích, chi phí nhân công chỉ tính tối thiểu 15 tr VNĐ/tháng.

Câu chuyện 1 – “Khách hàng mất doanh thu vì trùng lặp”

Công ty ABC Logistics có 12 chi nhánh, mỗi chi nhánh nhập dữ liệu khách hàng vào file Excel riêng. Khi tập hợp lại, họ phát hiện ≈ 30 % bản ghi là trùng lặp, dẫn đến việc gửi hoá đơn gấp đôi cho cùng một khách hàng. Kết quả: mất 200 tr VNĐ trong vòng 1 tháng vì khách hàng phản hồi và yêu cầu hoàn tiền.

Câu chuyện 2 – “Tiền lương nhân viên “đổ” vào công việc dọn dẹp dữ liệu”

Một startup fintech ở Hà Nội có 3 nhà phân tích dữ liệu. Họ dành ≈ 40 % thời gian làm việc để chuẩn hoá định dạng ngày tháng và loại bỏ ký tự đặc biệt trong dữ liệu giao dịch. Với mức lương trung bình 15 tr VNĐ/người, chi phí nhân công cho việc dọn dẹp dữ liệu lên tới ≈ 18 tr VNĐ/tháng.

Câu chuyện 3 – “Thành công sau khi tự động hoá”

Công ty XYZ Retail quyết định đầu tư vào workflow automation bằng Airflow + Python. Sau 2 tuần triển khai, thời gian xử lý dữ liệu giảm từ 4 giờ xuống còn 45 phút, chi phí nhân công giảm ≈ 12 tr VNĐ/tháng, và độ chính xác tăng lên 98 %. Kết quả: tăng doanh thu 5 % nhờ báo cáo kịp thời.

2. Giải pháp tổng quan (text art)

+-------------------+      +-------------------+      +-------------------+
|   Thu thập dữ liệu| ---> |   Chuẩn hoá format| ---> |   Deduplication   |
+-------------------+      +-------------------+      +-------------------+
          |                         |                         |
          v                         v                         v
   (CSV/Excel/DB)           (Regex, Pandas)           (Hash, Fuzzy)
          |                         |                         |
          +---------->  Workflow Engine (Airflow) <-----------+
                                 |
                                 v
                         +-------------------+
                         |   Lưu trữ sạch    |
                         +-------------------+

🛡️ Best Practice: Đặt key chuẩn (ví dụ: email + phone) làm tiêu chí deduplication để tránh hợp nhất sai.

3. Hướng dẫn chi tiết từng bước

Bước 1 – Thu thập nguồn dữ liệu

# Python example: đọc CSV và DB
import pandas as pd
import sqlalchemy

# CSV
df_csv = pd.read_csv('data/source.csv')

# DB (MySQL)
engine = sqlalchemy.create_engine('mysql+pymysql://user:pwd@host/db')
df_sql = pd.read_sql('SELECT * FROM customers', engine)

• Đảm bảo: Tất cả file đều có header thống nhất, charset utf‑8.

Bước 2 – Chuẩn hoá định dạng

import re
def normalize_date(date_str):
    # hỗ trợ dd/mm/yyyy, mm-dd-yyyy, yyyy/mm/dd
    patterns = [r'(\d{2})/(\d{2})/(\d{4})',
                r'(\d{2})-(\d{2})-(\d{4})',
                r'(\d{4})/(\d{2})/(\d{2})']
    for p in patterns:
        m = re.match(p, date_str)
        if m:
            d, mth, y = m.groups()
            return f'{y}-{mth}-{d}'
    return None   # nếu không khớp

df['date_of_birth'] = df['date_of_birth'].apply(normalize_date)
df['email'] = df['email'].str.lower().str.strip()
df['phone'] = df['phone'].apply(lambda x: f'+84{x[-9:]}')

Bước 3 – Phát hiện và hợp nhất bản ghi trùng lặp

3.1. Phương pháp hash (exact match)

df['hash_key'] = df['email'] + '_' + df['phone']
deduped = df.drop_duplicates(subset='hash_key', keep='first')

3.2. Phương pháp fuzzy (approximate)

from rapidfuzz import fuzz, process

def fuzzy_match(row, candidates, threshold=90):
    best = process.extractOne(row['name'], candidates, scorer=fuzz.token_set_ratio)
    if best[1] >= threshold:
        return best[0]   # trả về key của bản ghi gốc
    return None

candidates = df['name'].tolist()
df['match_key'] = df.apply(lambda r: fuzzy_match(r, candidates), axis=1)

🐛 Bug thường gặp: Khi NaN xuất hiện trong cột dùng làm hash_key, drop_duplicates sẽ coi mỗi NaN là một giá trị duy nhất → gây trùng lặp giả.
🛡️ Cách khắc: Trước khi tạo hash_key, thay thế NaN bằng chuỗi rỗng ('').

Bước 4 – Lưu trữ dữ liệu sạch

# Lưu vào PostgreSQL
engine_pg = sqlalchemy.create_engine('postgresql://user:pwd@host/db')
deduped.to_sql('customers_clean', engine_pg, if_exists='replace', index=False)

Bước 5 – Đặt lịch chạy tự động (Airflow DAG)

# airflow/dags/data_cleansing.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta

default_args = {
    'owner': 'hai',
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

def run_cleansing():
    # import và gọi hàm ở trên
    pass

with DAG('data_cleansing',
         start_date=datetime(2024, 1, 1),
         schedule_interval='@daily',
         default_args=default_args) as dag:
    task = PythonOperator(task_id='clean_data', python_callable=run_cleansing)

4. Template qui trình tham khảo

[START] → Pull raw data → Validate schema → Normalize fields → Generate hash key
   ↓                     ↓                 ↓                     ↓
   ✔   → Detect exact dup → Detect fuzzy dup → Merge records → Write clean DB
   ↓                     ↓                 ↓                     ↓
[END] ← Notify success ← Log metrics ← Archive raw ← Clean up temp files

Chi tiết từng bước trong template:

5. Những lỗi phổ biến & cách sửa

> Best Practice: Luôn tạo snapshot của dữ liệu gốc trước khi chạy bất kỳ bước nào có thể thay đổi dữ liệu.

6. Khi muốn scale lớn thì làm sao

1. Phân tán xử lý – Dùng Apache Spark hoặc Databricks để chạy deduplication trên cluster.
2. Queue & streaming – Đưa dữ liệu vào Kafka, dùng Kafka Streams hoặc Flink để chuẩn hoá và dedup liên tục.
3. Cache key – Lưu hash_key vào Redis để kiểm tra trùng lặp nhanh hơn khi dữ liệu liên tục chảy vào.

Công thức tính ROI (tiếng Việt)

ROI = (Tổng lợi ích – Chi phí đầu tư) / Chi phí đầu tư × 100%

LaTeX (tiếng Anh)

Giải thích: Total_Benefits bao gồm giảm chi phí nhân công, tăng doanh thu nhờ báo cáo nhanh; Investment_Cost là chi phí hạ tầng và phát triển workflow.

7. Chi phí thực tế

⚡ Lưu ý: Nếu dùng dịch vụ SaaS như Serimi App, chi phí chỉ từ 5 tr/tháng cho gói “Data Clean” và đã bao gồm các tính năng dedup tự động.

8. Số liệu trước – sau

> Best Practice: Đặt KPI rõ ràng (thời gian, chi phí, độ chính xác) và đo lường hàng tuần để điều chỉnh workflow.

9. FAQ hay gặp nhất

Q1: Có cần phải chuẩn hoá toàn bộ cột không?
A: Chỉ cần chuẩn hoá các cột dùng làm key (email, phone) và các cột quan trọng cho báo cáo (date, amount). Các cột phụ có thể bỏ qua để giảm thời gian.

Q2: Deduplication có làm mất dữ liệu quan trọng không?
A: Nếu key được thiết kế đúng (email + phone), rủi ro mất dữ liệu là rất thấp. Luôn backup dữ liệu gốc trước khi chạy.

Q3: Có thể chạy workflow trên Google Sheet không?
A: Có, dùng Google Apps Script hoặc Zapier để trigger Python/Node script qua webhook.

Q4: Làm sao bảo mật dữ liệu khi đưa lên cloud?
A: Mã hoá at‑rest (AES‑256) và in‑transit (TLS). Đặt IAM role hạn chế quyền chỉ đọc/ghi vào bucket cần thiết.

10. Giờ tới lượt bạn

1. Kiểm tra dữ liệu hiện tại – Xác định các cột cần chuẩn hoá và dedup.
2. Sao lưu – Tạo snapshot trên S3 hoặc backup DB.
3. Áp dụng template workflow – Copy DAG Airflow mẫu, chỉnh sửa hàm chuẩn hoá cho phù hợp.
4. Chạy thử trên mẫu nhỏ (≤ 10 k bản ghi) – Kiểm tra KPI (thời gian, độ chính xác).
5. Mở rộng lên production – Đặt lịch chạy hàng ngày, theo dõi logs và alert Slack khi lỗi xảy ra.
6. Đánh giá ROI – Sử dụng công thức trên để tính lợi nhuận thực tế sau 1 tháng.

Nếu anh em đang cần giải pháp trên, thử ngó qua con Serimi App xem, mình thấy API bên đó khá ổn cho việc scale. Hoặc liên hệ mình để được trao đổi nhanh hơn nhé.