Lập lịch job ETL (trích xuất-dữ liệu-chuyển đổi) từ SQL Server sang Excel bằng Python

1# Cài đặt các thư viện cần thiết
2# pip install pyodbc pandas openpyxl schedule

1import pandas as pd
2import pyodbc
3import os
4import datetime
5import time
6import schedule
7import logging
8from openpyxl import Workbook
9from openpyxl.styles import Font, Alignment, PatternFill
10from openpyxl.utils.dataframe import dataframe_to_rows

1def tao_ket_noi_sql_server():
2    """Tạo kết nối đến SQL Server"""
3    try:
4        conn = pyodbc.connect(
5            'DRIVER={SQL Server};'
6            'SERVER=ten_server;'
7            'DATABASE=ten_database;'
8            'UID=ten_dang_nhap;'
9            'PWD=mat_khau'
10        )
11        return conn
12    except Exception as e:
13        logging.error(f"Lỗi kết nối đến SQL Server: {str(e)}")
14        return None

1def trich_xuat_du_lieu(query, params=None):
2    """Trích xuất dữ liệu từ SQL Server sử dụng truy vấn"""
3    try:
4        conn = tao_ket_noi_sql_server()
5        if conn is None:
6            return None
7        
8        if params:
9            df = pd.read_sql(query, conn, params=params)
10        else:
11            df = pd.read_sql(query, conn)
12        
13        conn.close()
14        return df
15    except Exception as e:
16        logging.error(f"Lỗi trích xuất dữ liệu: {str(e)}")
17        return None

1def bien_doi_du_lieu(df):
2    """Xử lý và làm sạch dữ liệu"""
3    if df is None or df.empty:
4        return None
5    
6    try:
7        # Loại bỏ các dòng trùng lặp
8        df = df.drop_duplicates()
9        
10        # Làm sạch dữ liệu - ví dụ: điền giá trị thiếu
11        df = df.fillna({
12            'TenCot1': 'Không có dữ liệu',
13            'TenCot2': 0,
14            'TenCot3': df['TenCot3'].mean() if 'TenCot3' in df.columns else 0
15        })
16        
17        # Chuyển đổi kiểu dữ liệu nếu cần
18        if 'NgayThang' in df.columns:
19            df['NgayThang'] = pd.to_datetime(df['NgayThang'])
20        
21        # Tính toán các trường phụ nếu cần
22        if 'DoanhThu' in df.columns and 'ChiPhi' in df.columns:
23            df['LoiNhuan'] = df['DoanhThu'] - df['ChiPhi']
24            
25        # Sắp xếp dữ liệu
26        if 'NgayThang' in df.columns:
27            df = df.sort_values(by='NgayThang', ascending=False)
28            
29        return df
30    except Exception as e:
31        logging.error(f"Lỗi biến đổi dữ liệu: {str(e)}")
32        return df  # Trả về dữ liệu gốc nếu có lỗi

1def tao_file_excel(df, ten_file, ten_sheet="DuLieu"):
2    """Xuất dữ liệu ra file Excel với định dạng đẹp"""
3    if df is None or df.empty:
4        logging.warning("Không có dữ liệu để xuất ra Excel")
5        return False
6    
7    try:
8        # Tạo workbook mới
9        wb = Workbook()
10        ws = wb.active
11        ws.title = ten_sheet
12        
13        # Thêm tiêu đề
14        ws.append(['BÁO CÁO DỮ LIỆU'])
15        ws.append([f'Ngày tạo: {datetime.datetime.now().strftime("%d/%m/%Y %H:%M:%S")}'])
16        ws.append([])  # Dòng trống
17        
18        # Thêm dữ liệu từ DataFrame
19        rows = dataframe_to_rows(df, index=False, header=True)
20        for r_idx, row in enumerate(rows, 4):  # Bắt đầu từ dòng 4
21            ws.append(row)
22            
23        # Định dạng tiêu đề
24        ws.merge_cells('A1:' + chr(65 + len(df.columns) - 1) + '1')
25        cell = ws['A1']
26        cell.font = Font(size=16, bold=True)
27        cell.alignment = Alignment(horizontal='center')
28        
29        # Định dạng ngày tạo
30        ws.merge_cells('A2:' + chr(65 + len(df.columns) - 1) + '2')
31        cell = ws['A2']
32        cell.font = Font(italic=True)
33        cell.alignment = Alignment(horizontal='center')
34        
35        # Định dạng header
36        header_row = 4  # Dòng header
37        for col in range(1, len(df.columns) + 1):
38            cell = ws.cell(row=header_row, column=col)
39            cell.font = Font(bold=True)
40            cell.fill = PatternFill(start_color="DDDDDD", end_color="DDDDDD", fill_type="solid")
41            cell.alignment = Alignment(horizontal='center')
42            
43        # Tự động điều chỉnh độ rộng cột
44        for col in ws.columns:
45            max_length = 0
46            column = col[0].column_letter
47            for cell in col:
48                if cell.value:
49                    max_length = max(max_length, len(str(cell.value)))
50            adjusted_width = (max_length + 2) * 1.2
51            ws.column_dimensions[column].width = min(adjusted_width, 50)  # Giới hạn độ rộng tối đa
52            
53        # Lưu file
54        wb.save(ten_file)
55        logging.info(f"Đã xuất dữ liệu ra file Excel: {ten_file}")
56        return True
57    except Exception as e:
58        logging.error(f"Lỗi khi tạo file Excel: {str(e)}")
59        return False

1def etl_process(ten_file_excel=None):
2    """Thực hiện toàn bộ quy trình ETL"""
3    # Thiết lập logging
4    logging.basicConfig(
5        level=logging.INFO,
6        format='%(asctime)s - %(levelname)s - %(message)s',
7        filename='etl_log.log'
8    )
9    
10    # Nếu không cung cấp tên file, tạo tên file theo ngày hiện tại
11    if ten_file_excel is None:
12        ngay_hien_tai = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
13        ten_file_excel = f"BaoCao_{ngay_hien_tai}.xlsx"
14    
15    logging.info(f"Bắt đầu quy trình ETL, xuất ra file: {ten_file_excel}")
16    
17    # 1. Trích xuất dữ liệu
18    query = """
19    SELECT
20        p.ProductID,
21        p.Name AS TenSanPham,
22        p.ProductNumber,
23        pc.Name AS DanhMuc,
24        p.ListPrice AS GiaNiemYet,
25        p.StandardCost AS ChiPhi,
26        ISNULL(SUM(sod.OrderQty), 0) AS SoLuongBan,
27        ISNULL(SUM(sod.LineTotal), 0) AS DoanhThu
28    FROM
29        Production.Product p
30    LEFT JOIN
31        Sales.SalesOrderDetail sod ON p.ProductID = sod.ProductID
32    LEFT JOIN
33        Production.ProductSubcategory ps ON p.ProductSubcategoryID = ps.ProductSubcategoryID
34    LEFT JOIN
35        Production.ProductCategory pc ON ps.ProductCategoryID = pc.ProductCategoryID
36    WHERE
37        p.SellEndDate IS NULL
38    GROUP BY
39        p.ProductID, p.Name, p.ProductNumber, pc.Name, p.ListPrice, p.StandardCost
40    ORDER BY
41        DoanhThu DESC
42    """
43    
44    df = trich_xuat_du_lieu(query)
45    
46    if df is None:
47        logging.error("Không thể trích xuất dữ liệu từ SQL Server, quy trình ETL bị hủy")
48        return False
49    
50    # 2. Biến đổi dữ liệu
51    df_transformed = bien_doi_du_lieu(df)
52    
53    if df_transformed is None:
54        logging.error("Không thể biến đổi dữ liệu, quy trình ETL bị hủy")
55        return False
56    
57    # 3. Tạo file Excel
58    result = tao_file_excel(df_transformed, ten_file_excel, "BaoCaoSanPham")
59    
60    if result:
61        logging.info("Quy trình ETL hoàn tất thành công!")
62        return True
63    else:
64        logging.error("Quy trình ETL thất bại!")
65        return False

1def lap_lich_etl():
2    """Thiết lập lịch chạy tự động cho quy trình ETL"""
3    # Thiết lập báo cáo hàng ngày lúc 8 giờ sáng
4    schedule.every().day.at("08:00").do(etl_process, f"BaoCaoHangNgay_{datetime.datetime.now().strftime('%Y%m%d')}.xlsx")
5    
6    # Thiết lập báo cáo hàng tuần vào ngày thứ Hai
7    schedule.every().monday.at("07:00").do(etl_process, f"BaoCaoTuan_{datetime.datetime.now().strftime('%Y%m%d')}.xlsx")
8    
9    # Thiết lập báo cáo hàng tháng vào ngày đầu tiên của tháng
10    schedule.every().day.at("06:00").do(lambda: 
11        etl_process(f"BaoCaoThang_{datetime.datetime.now().strftime('%Y%m')}.xlsx") 
12        if datetime.datetime.now().day == 1 else None
13    )
14    
15    logging.info("Đã thiết lập lịch tự động chạy ETL")
16    
17    # Vòng lặp vô hạn để duy trì lịch
18    print("Chương trình lập lịch ETL đang chạy...")
19    print("Nhấn Ctrl+C để dừng")
20    
21    try:
22        while True:
23            schedule.run_pending()
24            time.sleep(60)  # Kiểm tra mỗi phút
25    except KeyboardInterrupt:
26        print("Đã dừng lập lịch ETL")
27        logging.info("Đã dừng lịch tự động chạy ETL")

1if __name__ == "__main__":
2    # Thiết lập logging
3    logging.basicConfig(
4        level=logging.INFO,
5        format='%(asctime)s - %(levelname)s - %(message)s',
6        handlers=[
7            logging.FileHandler("etl_scheduler.log"),
8            logging.StreamHandler()
9        ]
10    )
11    
12    # Bắt đầu lập lịch
13    lap_lich_etl()

1@echo off
2python D:\duong_dan\den\script\etl_scheduler.py

1sudo nano /etc/systemd/system/etl-scheduler.service

1[Unit]
2Description=ETL Scheduler Service
3After=network.target
4
5[Service]
6User=username
7WorkingDirectory=/duong/dan/den/thu/muc
8ExecStart=/usr/bin/python3 /duong/dan/den/script/etl_scheduler.py
9Restart=always
10
11[Install]
12WantedBy=multi-user.target

1sudo systemctl enable etl-scheduler.service
2sudo systemctl start etl-scheduler.service

1def gui_thong_bao(tieu_de, noi_dung):
2    """Gửi email thông báo"""
3    import smtplib
4    from email.mime.text import MIMEText
5    from email.mime.multipart import MIMEMultipart
6    
7    # Thông tin email
8    email_gui = "your_email@gmail.com"
9    mat_khau = "your_password"
10    email_nhan = "recipient@example.com"
11    
12    # Tạo tin nhắn
13    msg = MIMEMultipart()
14    msg['From'] = email_gui
15    msg['To'] = email_nhan
16    msg['Subject'] = tieu_de
17    
18    msg.attach(MIMEText(noi_dung, 'plain'))
19    
20    # Gửi email
21    try:
22        server = smtplib.SMTP('smtp.gmail.com', 587)
23        server.starttls()
24        server.login(email_gui, mat_khau)
25        text = msg.as_string()
26        server.sendmail(email_gui, email_nhan, text)
27        server.quit()
28        return True
29    except Exception as e:
30        logging.error(f"Lỗi gửi email: {str(e)}")
31        return False

1def etl_process_with_retry(ten_file_excel=None, so_lan_thu=3, thoi_gian_cho=300):
2    """Thực hiện ETL với cơ chế thử lại nếu thất bại"""
3    for lan_thu in range(so_lan_thu):
4        try:
5            result = etl_process(ten_file_excel)
6            if result:
7                return True
8                
9            logging.warning(f"ETL thất bại lần {lan_thu + 1}/{so_lan_thu}, chờ {thoi_gian_cho} giây trước khi thử lại...")
10            time.sleep(thoi_gian_cho)
11        except Exception as e:
12            logging.error(f"Lỗi trong quá trình ETL lần {lan_thu + 1}: {str(e)}")
13            time.sleep(thoi_gian_cho)
14            
15    # Nếu đã thử hết số lần mà vẫn thất bại, gửi thông báo
16    gui_thong_bao(
17        "ETL thất bại sau nhiều lần thử", 
18        f"Quy trình ETL không thể hoàn tất sau {so_lan_thu} lần thử. Vui lòng kiểm tra hệ thống."
19    )
20    return False