Hướng dẫn tối ưu truy vấn trên Data Warehouse với Python (Google BigQuery & Snowflake)

import os
from google.cloud import bigquery

# Khởi tạo hằng số cấu hình dự án
PROJECT_ID = "your-gcp-project-id"
DATASET_ID = "sales_data"
TARGET_TABLE = "transactions"

def initialize_bigquery_client():
    """Khởi tạo BigQuery client dựa trên tệp xác thực"""
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = "path_to_service_account.json"
    client = bigquery.Client(project=PROJECT_ID)
    return client

def execute_optimized_query(client_instance):
    """Thực hiện truy vấn đã được tối ưu hóa bằng cách giới hạn cột và phân vùng"""
    query_string = f"""
        SELECT 
            transaction_date, 
            customer_id, 
            SUM(total_amount) AS total_spent
        FROM `{PROJECT_ID}.{DATASET_ID}.{TARGET_TABLE}`
        WHERE transaction_date >= '2026-01-01'
        GROUP BY transaction_date, customer_id
        ORDER BY total_spent DESC
        LIMIT 100
    """
    
    # Thực thi truy vấn
    query_job = client_instance.query(query_string)
    results_dataframe = query_job.to_dataframe()
    
    return results_dataframe

# Thực thi chương trình
# bq_client = initialize_bigquery_client()
# df_results = execute_optimized_query(bq_client)

def estimate_query_cost(client_instance, query_text):
    """Kiểm tra chi phí truy vấn trước khi thực thi thực tế (Dry Run)"""
    job_config = bigquery.QueryJobConfig(dry_run=True, use_query_cache=False)
    
    # Gửi job kiểm tra
    dry_run_job = client_instance.query(query_text, job_config=job_config)
    
    bytes_processed = dry_run_job.total_bytes_processed
    megabytes_processed = bytes_processed / (1024 * 1024)
    
    print(f"Ước tính dung lượng dữ liệu quét: {megabytes_processed:.2f} MB")
    return bytes_processed

# Đoạn mã mẫu sử dụng hàm kiểm tra
# sample_query = "SELECT * FROM `project.dataset.table`"
# estimate_query_cost(bq_client, sample_query)

def validate_sql_query(query_string):
    """Kiểm tra câu lệnh SQL có tối ưu hay không (có WHERE và không chứa SELECT *)"""
    query_upper = query_string.upper()
    
    # Kiểm tra xem có sử dụng SELECT * hay không
    has_select_all = "SELECT *" in query_upper or "SELECT  *" in query_upper
    
    # Kiểm tra xem có mệnh đề WHERE hay không
    has_where_clause = "WHERE" in query_upper
    
    warnings = []
    if has_select_all:
        warnings.append("Cảnh báo: Tránh sử dụng 'SELECT *' trên Data Warehouse để tiết kiệm chi phí quét dữ liệu.")
    if not has_where_clause:
        warnings.append("Cảnh báo: Truy vấn thiếu mệnh đề 'WHERE', điều này sẽ quét toàn bộ bảng dữ liệu.")
        
    return {
        "is_safe": len(warnings) == 0,
        "warnings": warnings
    }

# Kiểm tra thử nghiệm với một câu lệnh SQL chưa tối ưu
UNOPTIMIZED_QUERY = "SELECT * FROM `project.dataset.massive_table`"
validation_result = validate_sql_query(UNOPTIMIZED_QUERY)

print(f"Trạng thái câu lệnh an toàn: {validation_result['is_safe']}")
for warning_message in validation_result['warnings']:
    print(f"- {warning_message}")

import pandas as pd

def fetch_data_in_batches(client_instance, base_query, batch_size=1000):
    """Truy vấn dữ liệu từ Data Warehouse theo từng lô nhỏ để tối ưu bộ nhớ RAM"""
    offset = 0
    all_chunks = []
    
    while True:
        # Xây dựng câu lệnh truy vấn phân trang
        paginated_query = f"{base_query} LIMIT {batch_size} OFFSET {offset}"
        
        # Mô phỏng quá trình thực thi truy vấn trên BigQuery/Snowflake
        # query_job = client_instance.query(paginated_query)
        # current_chunk_df = query_job.to_dataframe()
        
        # Dữ liệu giả lập cho mục đích minh họa
        current_chunk_df = pd.DataFrame({"row_id": range(offset, offset + batch_size)})
        
        # Kiểm tra nếu không còn dữ liệu trả về thì dừng vòng lặp
        if current_chunk_df.empty:
            break
            
        all_chunks.append(current_chunk_df)
        offset += batch_size
        
        # Giới hạn demo chạy tối đa 3 lô để tránh vòng lặp vô hạn
        if offset >= 3000:
            break
            
    # Gộp các lô lại thành một DataFrame hoàn chỉnh
    final_dataframe = pd.concat(all_chunks, ignore_index=True)
    print(f"Đã xử lý và gộp thành công {len(all_chunks)} lô dữ liệu.")
    
    return final_dataframe

# Thực thi hàm mẫu (Truyền client thực tế vào thay vì None khi chạy thật)
# BASE_SQL_QUERY = "SELECT transaction_id, customer_id FROM sales_transactions"
# result_df = fetch_data_in_batches(None, BASE_SQL_QUERY, batch_size=1000)