Tạo Datalake của Bạn với MINIO, TRINO và GRAFANA

Chào mừng bạn đến với bài viết về cách tạo một Datalake riêng của bạn sử dụng MINIO, TRINO và GRAFANA. Bài viết này sẽ được chia thành ba phần để dễ theo dõi. Ở phần này, chúng ta sẽ tập trung vào MINIO, nền tảng của Datalake.

Giới thiệu về MinIO: Hệ thống lưu trữ tương thích với S3

MinIO là một hệ thống lưu trữ đối tượng mã nguồn mở, hiệu suất cao, hoàn toàn tương thích với API của Amazon S3. Thay vì lưu trữ dữ liệu dưới dạng tệp trong các thư mục, MinIO lưu trữ chúng dưới dạng đối tượng trong các bucket. Điều này khiến nó cực kỳ linh hoạt và có khả năng mở rộng, đặc biệt khi làm việc với các tập dữ liệu lớn.

MinIO có thể được sử dụng làm lớp lưu trữ cho một Datalake. Trên nền tảng này, bạn có thể kết nối với các công cụ như:

• Apache Spark/Dask cho xử lý dữ liệu lớn
• Presto/Trino/Athena cho truy vấn SQL
• Grafana/Superset cho trực quan hóa dữ liệu
• Airflow cho quản lý công việc

Thiết lập MinIO

Cài đặt MinIO

Để bắt đầu, bạn cần tải về MinIO:

wget https://dl.min.io/server/minio/release/linux-amd64/minio

Làm cho tệp này có thể thực thi:

chmod +x minio

Di chuyển tệp thực thi này vào thư mục có thể chạy từ bất kỳ đâu:

sudo mv minio /usr/local/bin

Xuất thông tin người dùng và mật khẩu để khởi động lại máy chủ sau khi cài đặt:

export MINIO_ROOT_USER=<tên-người-dùng-của-bạn>
export MINIO_ROOT_PASSWORD=<mật-khẩu-của-bạn>

Khởi động máy chủ:

minio server ~/minio-data --console-address ":9001"

Lưu ý rằng, nếu bạn muốn kết nối với MinIO, nó sẽ mở một API tại "localhost:9000", vì vậy chúng ta sử dụng ":9001" cho máy chủ. Bạn có thể truy cập giao diện web tại "localhost:9001".

Sau đó, đăng nhập vào máy chủ để có một bảng điều khiển trình duyệt như thế này.

Thêm dữ liệu vào máy chủ

Trong trường hợp này, chúng ta có thể tạo một DAG trong Airflow để tải dữ liệu cho chúng ta. Tôi đang sử dụng Coingecko để lấy dữ liệu về một số đồng coin (đầu tiên bạn cần lấy API key, thêm vào biến môi trường, sau đó tải nó như tôi đã làm).

Chúng ta sẽ biến đổi dữ liệu bằng pandas để đưa vào định dạng đúng và cuối cùng lưu vào MinIO dưới dạng tệp Parquet.

• Nếu bạn không quen với Airflow, bạn có thể chỉ cần tạo một tệp riêng biệt và chạy hàm "coin_vis_etl".*

• Lưu ý tôi đang sử dụng cổng 9000, không phải 9001.*

Mã nguồn cho DAG Airflow

from airflow import DAG
from airflow.operators.python import PythonOperator
import requests
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
import os
from dotenv import load_dotenv
from datetime import datetime

def coin_vis_etl():
    load_dotenv()

    MY_API_KEY = os.getenv("MY_API_KEY")
    crypto_list = ['bitcoin', 'solana', 'ethereum', 'hyperliquid', 'binancecoin']

    for crypto in crypto_list:
        # Tải dữ liệu từ API
        url = f"https://api.coingecko.com/api/v3/coins/{crypto}/market_chart?vs_currency=usd&days=30"
        headers = {
            'accepts':'application/json',
            'x-cg-demo-api-key': MY_API_KEY
        }
        response = requests.get(url, headers=headers)

        # Định nghĩa metadata tùy chỉnh
        custom_metadata = {
            b"source": b"coingecko API",
            b"coin_name": f"{crypto}".encode()
        }

        if response.status_code == 200:
            data = response.json()
            temp = pd.DataFrame(data['prices'], columns=[f"{crypto}_timestamps", f"{crypto}_prices"])

            # Thêm các cột vì thời gian là giống nhau
            temp[f"{crypto}_market_caps"]= [x[1] for x in data["market_caps"]]
            temp[f"{crypto}_total_volumes"] = [x[1] for x in data["total_volumes"]]

            # Chuyển đổi timestamp thành định dạng thực
            temp[f"{crypto}_timestamps"] = pd.to_datetime(temp[f"{crypto}_timestamps"], unit='ms')

            # Sử dụng pyarrow để lưu thành Parquet
            table = pa.Table.from_pandas(temp)

            # Thêm metadata vào bảng
            existing_metadata = table.schema.metadata or {}
            new_metadata = {**custom_metadata, **existing_metadata}
            table = table.replace_schema_metadata(new_metadata)

            # Cuối cùng ghi dữ liệu vào MinIO
            fs = pa.fs.S3FileSystem(
                access_key="eric",
                secret_key='eric1234',
                endpoint_override ="http://localhost:9000"
            )

            pq.write_table(
                table,
                f"coin-vis-automated/coin_vis_{crypto}.parquet",
                filesystem=fs
            )
            print(f'{crypto} đã xong')

        else:
            print(f"Lỗi API tại {crypto}, {response.content}")

with DAG(
    dag_id = 'coin_vis_etl',
    start_date = datetime(2025, 9, 15),
    schedule_interval= '@hourly',
    catchup= False
) as dag:
    runetl = PythonOperator(
        task_id = 'coin_vis_etl',
        python_callable = coin_vis_etl
    )

    runetl

Như vậy, bạn đã tạo thành công một kho lưu trữ đối tượng tương thích với S3 của riêng mình.

Thực hành tốt nhất

• Đảm bảo an ninh: Sử dụng các biến môi trường để lưu trữ thông tin nhạy cảm như API keys và mật khẩu.
• Sao lưu dữ liệu: Thường xuyên sao lưu dữ liệu từ MinIO để tránh mất mát.
• Giám sát hiệu suất: Sử dụng Grafana để theo dõi hiệu suất của Datalake.

Cạm bẫy thường gặp

• Lỗi kết nối: Đảm bảo cổng và địa chỉ API được cấu hình đúng.
• Dữ liệu không chính xác: Kiểm tra phản hồi từ API và xử lý lỗi hợp lý.

Mẹo tối ưu hiệu suất

• Sử dụng Parquet: Tệp Parquet giúp giảm kích thước lưu trữ và tăng tốc độ truy vấn.
• Chọn lựa công cụ phù hợp: Sử dụng Spark hoặc Dask cho khối lượng dữ liệu lớn.

Kết luận

Bài viết này đã hướng dẫn bạn cách thiết lập một Datalake sử dụng MINIO, TRINO và GRAFANA. Hy vọng bạn đã có được những thông tin hữu ích và có thể áp dụng trong dự án của mình. Nếu bạn có bất kỳ câu hỏi nào, hãy để lại ý kiến dưới bài viết này.

Câu hỏi thường gặp

1. MinIO có miễn phí không?
   Có, MinIO là phần mềm mã nguồn mở và miễn phí.
2. Tôi có thể sử dụng MinIO cho sản xuất không?
   Có, MinIO được thiết kế cho cả môi trường phát triển và sản xuất.
3. Có cần cài đặt thêm phần mềm nào không?
   Để sử dụng Datalake đầy đủ chức năng, bạn nên kết hợp với các công cụ như Trino và Grafana.