Giới thiệu

Dữ liệu hiện diện khắp nơi, nhưng để hiểu rõ nó cần phải thu thập, làm sạch, lưu trữ và phân tích một cách hiệu quả. Trong bài viết này, tôi sẽ hướng dẫn bạn cách xây dựng một Pipeline Phân Tích Cảm Xúc Reddit. Pipeline này sẽ lấy bài viết từ Reddit, phân tích cảm xúc của chúng, lưu kết quả vào PostgreSQL, và trực quan hóa thông tin trong Grafana. Tất cả được điều phối bằng Apache Airflow và container hóa bằng Docker Compose.

Nếu bạn muốn xem mã nguồn, hãy kiểm tra repository tại đây.

Chi tiết dự án

Trích xuất bài viết từ các subreddit cụ thể (sử dụng API praw).
Làm sạch và phân tích cảm xúc với VADER Sentiment Analyzer.
Tải dữ liệu đã xử lý vào Postgres.
Điều phối mọi thứ với Airflow DAGs.
Trực quan hóa thông tin trên Dashboard Grafana.
Nhận thông báo qua email khi DAG hoàn thành hoặc thất bại qua SMTP.
Giám sát các chỉ số và sức khỏe dịch vụ bằng Prometheus, StatsD và Grafana.

Dưới đây là sơ đồ kiến trúc của dự án:

Cài đặt dự án

Bước 1: Trích xuất dữ liệu từ Reddit

Sử dụng thư viện praw trong Python, tôi sẽ lấy 50 bài viết hàng đầu từ các subreddit như r/kenya và r/nairobi.

python Copy

@task
def extract_data():
    reddit = praw.Reddit(
        client_id=CLIENT_ID,
        client_secret=SECRET_KEY,
        user_agent='data pipeline by u/user'
    )

    limit = 50
    data = []
    subs = ["kenya", "nairobi"]
    for sub in subs:
        subreddit = reddit.subreddit(sub)
        for s in subreddit.top(time_filter='day', limit=limit):
            data.append({
                'id': s.id,
                'title': clean_text(s.title) if s.title else 'N/A',
                'text': clean_text(s.selftext) if s.selftext else 'N/A',
                'url': s.url,
                'score': s.score,
                'subreddit': sub,
                'time_created': s.created_utc
            })

    return data

Bước 2: Biến đổi và Phân tích

Các bài viết sẽ được làm sạch (xóa ký tự đặc biệt, lọc ra N/A) và được gửi vào VADER Sentiment Analyzer, trả về:

pos: Điểm tích cực
neu: Điểm trung tính
neg: Điểm tiêu cực
compound: Cảm xúc tổng thể

Tôi phân loại cảm xúc thành Tích Cực, Trung Tính, hoặc Tiêu Cực.

python Copy

@task
def analyze_text():
    """Lấy bài viết từ Postgres, phân tích cảm xúc và lưu kết quả vào bảng khác."""
    from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
    from utils.extract import get_sentiment

    engine = create_engine(DB_URL)
    analyzer = SentimentIntensityAnalyzer()

    try:
        df = pd.read_sql_table("reddit_posts", con=engine, schema="reddit")
        df["text"] = df["text"].replace("N/A", np.nan)
        df.dropna(subset=["text"], inplace=True)

        scores = df["text"].apply(lambda x: analyzer.polarity_scores(x))
        df["compound"] = scores.apply(lambda x: x["compound"])
        df["neg"] = scores.apply(lambda x: x["neg"])
        df["neu"] = scores.apply(lambda x: x["neu"])
        df["pos"] = scores.apply(lambda x: x["pos"])
        df["sentiment"] = df["compound"].apply(get_sentiment)

        df.to_sql(
            "reddit_sentiment_analysis",
            con=engine,
            schema="reddit",
            if_exists="append",
            index=False,
        )
        return "Tải dữ liệu thành công!"
    except Exception as e:
        raise RuntimeError(f"Lỗi khi phân tích hoặc lưu dữ liệu: {e}")

Bước 3: Tải vào PostgreSQL

Dữ liệu đã xử lý sẽ được lưu vào bảng reddit_sentiment_analysis, và dữ liệu phân tích sẽ được lưu vào bảng reddit_posts, giúp tách biệt các mối quan tâm trong kiến trúc medallion: layers cho dữ liệu thô, dữ liệu bán cấu trúc và layer cho dữ liệu đã được xử lý, phục vụ như "nguồn thông tin duy nhất" cho các quyết định và sẵn sàng cho các bảng điều khiển trong Grafana.

Bước 4: Điều phối bằng Airflow

Hai DAG điều phối quy trình làm việc:

reddit_dag.py: Trích xuất và lưu trữ bài viết thô.
analyze_dag.py: Chờ DAG đầu tiên hoàn thành, thực hiện phân tích cảm xúc và gửi thông báo qua email.

Bước 5: Trực quan hóa trong Grafana

Grafana truy vấn Postgres trực tiếp và trực quan hóa các chỉ số như xu hướng cảm xúc, so sánh subreddit và khối lượng bài viết. Nó cũng trực quan hóa các chỉ số của Airflow qua Prometheus và Grafana.

StatsD được tích hợp vào dự án này như một bộ thu thập chỉ số cho Airflow. Airflow phát ra các chỉ số như dag_processing.processes và executor.open_slots, ghi lại sức khỏe của scheduler và worker, thời gian xử lý DAG và chi tiết thực thi nhiệm vụ.

Prometheus được sử dụng làm hệ thống lưu trữ và truy vấn chỉ số. Dịch vụ Prometheus được cấu hình trong các tệp docker-compose.yml và prometheus.yml để thu thập các chỉ số do Airflow phát ra tại điểm cuối /metrics qua một Prometheus exporter.

Thêm Prometheus làm nguồn dữ liệu trên Grafana và trực quan hóa chỉ số:

Truy cập localhost:3000 để vào dịch vụ Grafana.
Sử dụng admin & admin làm thông tin đăng nhập. Bạn sẽ được yêu cầu thay đổi mật khẩu vì lý do bảo mật.
Thêm Prometheus làm nguồn dữ liệu. Nhấp vào 'Add Data Source' và chọn Prometheus.
Nhập một bảng điều khiển tùy chỉnh bằng cách nhấp vào Import Dashboard và dán nội dung của dashboard.json.

Những hiểu biết từ dự án này

Bảng điều khiển dưới đây trực quan hóa dữ liệu được tạo từ dự án dựa trên điểm số cảm xúc:

66% dữ liệu cho thấy cảm xúc tích cực, 23% tiêu cực và 7% trung tính.
r/Nairobi có nhiều bài viết tích cực hơn so với r/Kenya nhưng với số lượng bài viết cao hơn.
r/Nairobi có 659 bài viết so với 616 bài viết của r/Kenya.
Điểm cảm xúc cao nhất được ghi nhận vào ngày 15 tháng 9 năm 2025 (0.483) và thấp nhất vào ngày 12 tháng 9 năm 2025 (0.213).
Điểm upvote không ảnh hưởng đến điểm cảm xúc, vì vậy điểm upvote thấp không nhất thiết đồng nghĩa với cảm xúc thấp và ngược lại.

Kết luận

Dự án này nhấn mạnh cách mà các pipeline dữ liệu được thiết kế tốt có thể mở khóa những hiểu biết giá trị từ các nguồn dữ liệu phi cấu trúc như Reddit, đồng thời tuân thủ các thực tiễn tốt nhất trong kỹ thuật dữ liệu và phân tích.

Hãy thích, bình luận và chia sẻ bài viết này rộng rãi! Để hợp tác trong các dự án, vui lòng gửi email cho tôi tại denzelkinyua11@gmail.com hoặc truy cập các nền tảng mạng xã hội của tôi được liên kết trên trang GitHub của tôi.

Xây Dựng Pipeline Phân Tích Cảm Xúc Reddit với Python và PostgreSQL