pip install luigi

pip install luigi[toml]  # TOML 설정 파일 지원
pip install luigi[prometheus]  # Prometheus 모니터링 지원
pip install luigi[all]  # 모든 추가 기능 설치

[core]
log_level = INFO
default-scheduler-host = localhost
default-scheduler-port = 8082

[scheduler]
record_task_history = True
remove_delay = 86400

[worker]
keep_alive = True
ping_interval = 20
wait_interval = 20
max_reschedules = 3

luigid

import luigi

class MyTask(luigi.Task):
    def requires(self):
        return SomeOtherTask()
        
    def output(self):
        return luigi.LocalTarget('output.txt')
        
    def run(self):
        # 실제 작업 수행
        with self.output().open('w') as f:
            f.write('작업 완료!')

# 로컬 파일 Target
local_target = luigi.LocalTarget('data/output.csv')

# S3 Target (boto3 필요)
s3_target = luigi.contrib.s3.S3Target('s3://my-bucket/path/to/file.csv')

# PostgreSQL Target
pg_target = luigi.contrib.postgres.PostgresTarget(
    host='localhost', 
    database='mydb',
    user='user', 
    password='pass', 
    table='mytable',
    update_id='task_id'
)

class ProcessData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    data_path = luigi.Parameter(default='data/')
    process_count = luigi.IntParameter(default=10)
    
    def run(self):
        print(f"처리 날짜: {self.date}")
        print(f"데이터 경로: {self.data_path}")
        print(f"처리 개수: {self.process_count}")

python -m luigi --module my_module ProcessData --date 2025-03-26 --data-path /data/new/ --process-count 20

import luigi
import pandas as pd
import requests
from datetime import date

# 1. 데이터 다운로드 작업
class DownloadData(luigi.Task):
    date = luigi.DateParameter(default=date.today())
    
    def output(self):
        return luigi.LocalTarget(f'data/raw/data_{self.date}.csv')
    
    def run(self):
        # 데이터 URL (예시)
        url = f'https://example.com/data/daily_{self.date}.csv'
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 데이터 다운로드
        print(f"데이터 다운로드 중... 📥")
        response = requests.get(url)
        
        # 파일로 저장
        with self.output().open('w') as f:
            f.write(response.text)
        
        print(f"다운로드 완료! 👍")

# 2. 데이터 처리 작업
class ProcessData(luigi.Task):
    date = luigi.DateParameter(default=date.today())
    min_value = luigi.IntParameter(default=0)
    
    def requires(self):
        return DownloadData(date=self.date)
    
    def output(self):
        return luigi.LocalTarget(f'data/processed/processed_data_{self.date}.csv')
    
    def run(self):
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 입력 데이터 읽기
        print(f"데이터 처리 중... ⚙️")
        with self.input().open('r') as f:
            df = pd.read_csv(f)
        
        # 데이터 처리 (예: 특정 값 이상인 행만 필터링)
        filtered_df = df[df['value'] > self.min_value]
        
        # 결과 저장
        with self.output().open('w') as f:
            filtered_df.to_csv(f, index=False)
        
        print(f"처리 완료! 🎉 {len(df)}행에서 {len(filtered_df)}행으로 필터링됨")

# 3. 결과 저장 작업
class GenerateReport(luigi.Task):
    date = luigi.DateParameter(default=date.today())
    min_value = luigi.IntParameter(default=0)
    
    def requires(self):
        return ProcessData(date=self.date, min_value=self.min_value)
    
    def output(self):
        return luigi.LocalTarget(f'data/reports/report_{self.date}.html')
    
    def run(self):
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 처리된 데이터 읽기
        print(f"리포트 생성 중... 📊")
        with self.input().open('r') as f:
            df = pd.read_csv(f)
        
        # 간단한 HTML 리포트 생성
        html_content = f"""
        <html>
        <head><title>데이터 리포트 {self.date}</title></head>
        <body>
            <h1>데이터 리포트 - {self.date}</h1>
            <p>총 {len(df)} 개의 레코드가 처리되었습니다.</p>
            <p>최소값: {df['value'].min()}, 최대값: {df['value'].max()}, 평균: {df['value'].mean():.2f}</p>
            <h2>데이터 미리보기</h2>
            {df.head(10).to_html()}
        </body>
        </html>
        """
        
        # 결과 저장
        with self.output().open('w') as f:
            f.write(html_content)
        
        print(f"리포트 생성 완료! 📑")

python -m luigi --module my_pipeline GenerateReport --date 2025-03-26 --min-value 10 --local-scheduler

class MyTask(luigi.Task):
    def requires(self):
        return PreviousTask()  # 단일 의존성
        
class AnotherTask(luigi.Task):
    def requires(self):
        return [Task1(), Task2(), Task3()]  # 다중 의존성

class ProcessMultipleDays(luigi.Task):
    start_date = luigi.DateParameter()
    end_date = luigi.DateParameter()
    
    def requires(self):
        # 날짜 범위에 대해 동적으로 작업 생성
        current_date = self.start_date
        tasks = []
        
        while current_date <= self.end_date:
            tasks.append(ProcessData(date=current_date))
            current_date += datetime.timedelta(days=1)
            
        return tasks

class ConditionalTask(luigi.Task):
    data_source = luigi.Parameter(default='local')
    
    def requires(self):
        if self.data_source == 'local':
            return LocalDataTask()
        elif self.data_source == 's3':
            return S3DataTask()
        elif self.data_source == 'database':
            return DatabaseTask()
        else:
            raise ValueError(f"지원하지 않는 데이터 소스: {self.data_source}")

from luigi.tools.deps import get_task_requires

# 의존성 트리 출력
def print_dependency_tree(task, indent=0):
    print(' ' * indent + f"- {task.__class__.__name__}")
    deps = get_task_requires(task)
    for dep in deps:
        print_dependency_tree(dep, indent + 2)

# 사용 예
my_task = MyComplexTask()
print_dependency_tree(my_task)

# 기본 설정으로 실행
luigid

# 포트 및 로그 레벨 지정
luigid --port 8082 --log-level INFO

# 백그라운드로 실행 (Linux/Mac)
nohup luigid > luigi.log 2>&1 &

# crontab -e 명령어로 cron 작업 편집

# 매일 오전 3시에 실행
0 3 * * * python -m luigi --module my_pipeline DailyReport --date $(date +\%Y-\%m-\%d)

# 매주 월요일 오전 7시에 실행
0 7 * * 1 python -m luigi --module my_pipeline WeeklyReport --date $(date +\%Y-\%m-\%d)

class DailyReport(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def output(self):
        return luigi.LocalTarget(f'reports/daily/report_{self.date}.html')
    
    def requires(self):
        return ProcessDailyData(date=self.date)
    
    def run(self):
        # 일일 리포트 생성 로직
        pass

class MonthlyReport(luigi.Task):
    month = luigi.MonthParameter(default=datetime.date.today().replace(day=1))
    
    def output(self):
        return luigi.LocalTarget(f'reports/monthly/report_{self.month}.html')
    
    def requires(self):
        # 해당 월의 모든 일일 리포트에 의존
        start_date = self.month
        end_date = (self.month + datetime.timedelta(days=32)).replace(day=1) - datetime.timedelta(days=1)
        
        return [DailyReport(date=start_date + datetime.timedelta(days=i)) 
                for i in range((end_date - start_date).days + 1)]
    
    def run(self):
        # 월간 리포트 생성 로직
        pass

class RetryTask(luigi.Task):
    retry_count = luigi.IntParameter(default=3)
    
    def run_with_retry(self, func, *args, **kwargs):
        for attempt in range(self.retry_count):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                if attempt < self.retry_count - 1:
                    wait_time = 2 ** attempt  # 지수 백오프
                    print(f"시도 {attempt+1}/{self.retry_count} 실패: {e}")
                    print(f"{wait_time}초 후 재시도...")
                    time.sleep(wait_time)
                else:
                    raise  # 모든 재시도 실패 시 예외 발생
    
    def run(self):
        self.run_with_retry(self.process_data)
    
    def process_data(self):
        # 실제 데이터 처리 로직
        pass

[scheduler]
# 작업 기록 보관 설정
record_task_history = True
remove_delay = 86400  # 1일 동안 기록 보관

# 시각화 설정
visualiser_html = /custom/visualiser.html  # 커스텀 시각화 HTML
max_graph_nodes = 100  # 그래프에 표시할 최대 노드 수

[webhdfs]
# HDFS 통합 설정 (HDFS 사용 시)
port = 50070
user = hdfs

import luigi
import pandas as pd
import datetime
import os
import sqlalchemy

# 기본 설정
class Config:
    DATA_DIR = 'data/'
    DB_CONNECTION = 'postgresql://user:pass@localhost:5432/datawarehouse'

# 1. 데이터 추출 작업
class ExtractSalesData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}raw/sales_{self.date}.csv')
    
    def run(self):
        # 실제로는 API나 데이터베이스에서 데이터를 가져옴
        print(f"판매 데이터 추출 중... 📊")
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 예시 데이터 생성
        sales_data = pd.DataFrame({
            'order_id': range(1000, 1100),
            'product_id': [f'P{i}' for i in range(100, 200)],
            'customer_id': [f'C{i}' for i in range(50, 150)],
            'quantity': [i % 5 + 1 for i in range(100)],
            'price': [round(10.0 + i * 0.5, 2) for i in range(100)],
            'timestamp': [f'{self.date} {h:02d}:{m:02d}:00' 
                         for h in range(24) for m in range(0, 60, 15)][:100]
        })
        
        # 파일로 저장
        with self.output().open('w') as f:
            sales_data.to_csv(f, index=False)
        
        print(f"판매 데이터 추출 완료! 🎉")

class ExtractCustomerData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}raw/customers_{self.date}.csv')
    
    def run(self):
        print(f"고객 데이터 추출 중... 👥")
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 예시 데이터 생성
        customer_data = pd.DataFrame({
            'customer_id': [f'C{i}' for i in range(50, 150)],
            'name': [f'Customer {i}' for i in range(50, 150)],
            'email': [f'customer{i}@example.com' for i in range(50, 150)],
            'segment': ['VIP' if i % 10 == 0 else 'Regular' if i % 5 == 0 else 'New' 
                       for i in range(50, 150)]
        })
        
        # 파일로 저장
        with self.output().open('w') as f:
            customer_data.to_csv(f, index=False)
        
        print(f"고객 데이터 추출 완료! 🎉")

class ExtractInventoryData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}raw/inventory_{self.date}.csv')
    
    def run(self):
        print(f"재고 데이터 추출 중... 📦")
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 예시 데이터 생성
        inventory_data = pd.DataFrame({
            'product_id': [f'P{i}' for i in range(100, 200)],
            'product_name': [f'Product {i}' for i in range(100, 200)],
            'category': ['Electronics' if i % 3 == 0 else 'Clothing' if i % 3 == 1 else 'Home' 
                        for i in range(100, 200)],
            'stock_level': [i % 50 + 10 for i in range(100, 200)],
            'reorder_level': [10 for _ in range(100)]
        })
        
        # 파일로 저장
        with self.output().open('w') as f:
            inventory_data.to_csv(f, index=False)
        
        print(f"재고 데이터 추출 완료! 🎉")

# 2. 데이터 변환 작업
class TransformSalesData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def requires(self):
        return {
            'sales': ExtractSalesData(date=self.date),
            'customers': ExtractCustomerData(date=self.date),
            'inventory': ExtractInventoryData(date=self.date)
        }
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}transformed/sales_enriched_{self.date}.csv')
    
    def run(self):
        print(f"판매 데이터 변환 중... ⚙️")
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 데이터 로드
        with self.input()['sales'].open('r') as f:
            sales_df = pd.read_csv(f)
        
        with self.input()['customers'].open('r') as f:
            customers_df = pd.read_csv(f)
        
        with self.input()['inventory'].open('r') as f:
            inventory_df = pd.read_csv(f)
        
        # 데이터 변환 및 통합
        # 1. 판매 데이터에 고객 정보 추가
        enriched_df = sales_df.merge(
            customers_df[['customer_id', 'segment']], 
            on='customer_id', 
            how='left'
        )
        
        # 2. 판매 데이터에 제품 정보 추가
        enriched_df = enriched_df.merge(
            inventory_df[['product_id', 'product_name', 'category']], 
            on='product_id', 
            how='left'
        )
        
        # 3. 총 판매 금액 계산
        enriched_df['total_amount'] = enriched_df['quantity'] * enriched_df['price']
        
        # 4. 타임스탬프를 datetime으로 변환
        enriched_df['timestamp'] = pd.to_datetime(enriched_df['timestamp'])
        enriched_df['hour'] = enriched_df['timestamp'].dt.hour
        
        # 결과 저장
        with self.output().open('w') as f:
            enriched_df.to_csv(f, index=False)
        
        print(f"판매 데이터 변환 완료! 🎉")

class AggregateData(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def requires(self):
        return TransformSalesData(date=self.date)
    
    def output(self):
        return {
            'hourly': luigi.LocalTarget(f'{Config.DATA_DIR}aggregated/hourly_sales_{self.date}.csv'),
            'product': luigi.LocalTarget(f'{Config.DATA_DIR}aggregated/product_sales_{self.date}.csv'),
            'customer': luigi.LocalTarget(f'{Config.DATA_DIR}aggregated/customer_segment_{self.date}.csv')
        }
    
    def run(self):
        print(f"데이터 집계 중... 📊")
        
        # 디렉토리 생성
        for target in self.output().values():
            os.makedirs(os.path.dirname(target.path), exist_ok=True)
        
        # 변환된 데이터 로드
        with self.input().open('r') as f:
            df = pd.read_csv(f)
            df['timestamp'] = pd.to_datetime(df['timestamp'])
        
        # 1. 시간별 판매 집계
        hourly_sales = df.groupby('hour').agg({
            'order_id': 'count',
            'total_amount': 'sum'
        }).reset_index()
        hourly_sales.rename(columns={'order_id': 'order_count'}, inplace=True)
        
        # 2. 제품별 판매 집계
        product_sales = df.groupby(['product_id', 'product_name', 'category']).agg({
            'quantity': 'sum',
            'total_amount': 'sum'
        }).reset_index()
        
        # 3. 고객 세그먼트별 집계
        customer_segment = df.groupby('segment').agg({
            'customer_id': 'nunique',
            'order_id': 'count',
            'total_amount': 'sum'
        }).reset_index()
        customer_segment.rename(columns={
            'customer_id': 'customer_count',
            'order_id': 'order_count'
        }, inplace=True)
        
        # 결과 저장
        with self.output()['hourly'].open('w') as f:
            hourly_sales.to_csv(f, index=False)
        
        with self.output()['product'].open('w') as f:
            product_sales.to_csv(f, index=False)
        
        with self.output()['customer'].open('w') as f:
            customer_segment.to_csv(f, index=False)
        
        print(f"데이터 집계 완료! 🎉")

# 3. 데이터 적재 작업
class LoadDataToWarehouse(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def requires(self):
        return {
            'transformed': TransformSalesData(date=self.date),
            'aggregated': AggregateData(date=self.date)
        }
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}load_complete_{self.date}.txt')
    
    def run(self):
        print(f"데이터 웨어하우스에 데이터 적재 중... 🏢")
        
        # 데이터베이스 연결
        engine = sqlalchemy.create_engine(Config.DB_CONNECTION)
        
        # 변환된 데이터 로드
        with self.input()['transformed'].open('r') as f:
            transformed_df = pd.read_csv(f)
        
        # 집계된 데이터 로드
        with self.input()['aggregated']['hourly'].open('r') as f:
            hourly_df = pd.read_csv(f)
        
        with self.input()['aggregated']['product'].open('r') as f:
            product_df = pd.read_csv(f)
        
        with self.input()['aggregated']['customer'].open('r') as f:
            customer_df = pd.read_csv(f)
        
        # 데이터베이스에 저장
        # 실제로는 아래와 같이 데이터를 DB에 저장
        # transformed_df.to_sql('sales_enriched', engine, if_exists='append', index=False)
        # hourly_df.to_sql('hourly_sales', engine, if_exists='append', index=False)
        # product_df.to_sql('product_sales', engine, if_exists='append', index=False)
        # customer_df.to_sql('customer_segment', engine, if_exists='append', index=False)
        
        # 예시에서는 실제 DB 저장 대신 로그만 출력
        print(f"변환된 데이터: {len(transformed_df)}행 적재 완료")
        print(f"시간별 집계 데이터: {len(hourly_df)}행 적재 완료")
        print(f"제품별 집계 데이터: {len(product_df)}행 적재 완료")
        print(f"고객 세그먼트별 집계 데이터: {len(customer_df)}행 적재 완료")
        
        # 완료 표시 파일 생성
        with self.output().open('w') as f:
            f.write(f"Data load completed at {datetime.datetime.now()}")
        
        print(f"데이터 웨어하우스 적재 완료! 🎉")

# 4. 리포트 생성 작업
class GenerateDailySalesReport(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def requires(self):
        return LoadDataToWarehouse(date=self.date)
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}reports/daily_sales_report_{self.date}.html')
    
    def run(self):
        print(f"일일 판매 보고서 생성 중... 📑")
        
        # 디렉토리 생성
        os.makedirs(os.path.dirname(self.output().path), exist_ok=True)
        
        # 집계 데이터 로드
        aggregated_task = AggregateData(date=self.date)
        
        with aggregated_task.output()['hourly'].open('r') as f:
            hourly_df = pd.read_csv(f)
        
        with aggregated_task.output()['product'].open('r') as f:
            product_df = pd.read_csv(f)
        
        with aggregated_task.output()['customer'].open('r') as f:
            customer_df = pd.read_csv(f)
        
        # 상위 10개 제품 추출
        top_products = product_df.sort_values('total_amount', ascending=False).head(10)
        
        # HTML 리포트 생성
        html_content = f"""
        <html>
        <head>
            <title>일일 판매 보고서 - {self.date}</title>
            <style>
                body {{ font-family: Arial, sans-serif; margin: 20px; }}
                h1, h2 {{ color: #2c3e50; }}
                table {{ border-collapse: collapse; width: 100%; margin-bottom: 20px; }}
                th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
                th {{ background-color: #f2f2f2; }}
                tr:nth-child(even) {{ background-color: #f9f9f9; }}
                .summary {{ background-color: #e8f4f8; padding: 15px; border-radius: 5px; margin-bottom: 20px; }}
            </style>
        </head>
        <body>
            <h1>일일 판매 보고서 - {self.date}</h1>
            
            <div class="summary">
                <h2>요약</h2>
                <p>총 주문 수: {hourly_df['order_count'].sum()}</p>
                <p>총 판매 금액: ${hourly_df['total_amount'].sum():.2f}</p>
                <p>평균 주문 금액: ${hourly_df['total_amount'].sum() / hourly_df['order_count'].sum():.2f}</p>
            </div>
            
            <h2>시간별 판매 현황</h2>
            <table>
                <tr>
                    <th>시간</th>
                    <th>주문 수</th>
                    <th>판매 금액</th>
                </tr>
                {"".join(f"<tr><td>{row['hour']}시</td><td>{row['order_count']}</td><td>${row['total_amount']:.2f}</td></tr>" for _, row in hourly_df.iterrows())}
            </table>
            
            <h2>상위 10개 제품</h2>
            <table>
                <tr>
                    <th>제품 ID</th>
                    <th>제품명</th>
                    <th>카테고리</th>
                    <th>판매 수량</th>
                    <th>판매 금액</th>
                </tr>
                {"".join(f"<tr><td>{row['product_id']}</td><td>{row['product_name']}</td><td>{row['category']}</td><td>{row['quantity']}</td><td>${row['total_amount']:.2f}</td></tr>" for _, row in top_products.iterrows())}
            </table>
            
            <h2>고객 세그먼트별 판매 현황</h2>
            <table>
                <tr>
                    <th>세그먼트</th>
                    <th>고객 수</th>
                    <th>주문 수</th>
                    <th>판매 금액</th>
                    <th>고객당 평균 주문 금액</th>
                </tr>
                {"".join(f"<tr><td>{row['segment']}</td><td>{row['customer_count']}</td><td>{row['order_count']}</td><td>${row['total_amount']:.2f}</td><td>${row['total_amount']/row['customer_count']:.2f}</td></tr>" for _, row in customer_df.iterrows())}
            </table>
            
            <p>이 보고서는 {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}에 생성되었습니다.</p>
        </body>
        </html>
        """
        
        # 리포트 저장
        with self.output().open('w') as f:
            f.write(html_content)
        
        print(f"일일 판매 보고서 생성 완료! 📊")

# 5. 메인 작업 - 전체 파이프라인 실행
class ECommerceETLPipeline(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())
    
    def requires(self):
        return GenerateDailySalesReport(date=self.date)
    
    def output(self):
        return luigi.LocalTarget(f'{Config.DATA_DIR}pipeline_complete_{self.date}.txt')
    
    def run(self):
        print(f"전체 ETL 파이프라인 완료! 🚀")
        
        with self.output().open('w') as f:
            f.write(f"E-commerce ETL pipeline completed successfully at {datetime.datetime.now()}")
        
        print(f"🎉 축하합니다! {self.date} 데이터 처리가 완료되었습니다. 🎉")

python -m luigi --module ecommerce_etl ECommerceETLPipeline --date 2025-03-26 --local-scheduler

class ProcessMultipleFiles(luigi.Task):
    file_pattern = luigi.Parameter()
    
    def requires(self):
        # 파일 패턴에 맞는 모든 파일에 대해 작업 생성
        matching_files = glob.glob(self.file_pattern)
        return [ProcessSingleFile(file_path=f) for f in matching_files]

class RedisTarget(luigi.Target):
    def __init__(self, host, port, db, key):
        self.host = host
        self.port = port
        self.db = db
        self.key = key
        self._client = redis.Redis(host=host, port=port, db=db)
    
    def exists(self):
        return self._client.exists(self.key)
    
    def set(self, value):
        self._client.set(self.key, value)
    
    def get(self):
        return self._client.get(self.key)

def run(self):
    total_items = 1000
    
    for i in range(total_items):
        # 작업 수행
        process_item(i)
        
        # 진행 상황 업데이트 (10% 단위)
        if i % (total_items // 10) == 0:
            progress = i / total_items * 100
            self.set_status_message(f"진행률: {progress:.1f}%")

class HighPriorityTask(luigi.Task):
    priority = 100  # 높은 값일수록 우선순위가 높음
    
    def run(self):
        # 중요한 작업 수행
        pass

class LowPriorityTask(luigi.Task):
    priority = 10  # 낮은 값은 우선순위가 낮음
    
    def run(self):
        # 덜 중요한 작업 수행
        pass

class DataProcessingGroup(luigi.WrapperTask):
    date = luigi.DateParameter()
    
    def requires(self):
        return [
            ExtractTask(date=self.date),
            TransformTask(date=self.date),
            LoadTask(date=self.date)
        ]

@luigi.Task.event_handler(luigi.Event.FAILURE)
def on_failure(self, exception):
    if isinstance(exception, TemporaryError) and self.attempts < self.max_attempts:
        self.attempts += 1
        sleep_time = 2 ** self.attempts  # 지수 백오프
        print(f"일시적 오류 발생, {sleep_time}초 후 재시도 ({self.attempts}/{self.max_attempts})")
        time.sleep(sleep_time)
        self.run()  # 작업 재실행

# luigi.cfg 파일에 이메일 설정 추가
[email]
receiver = your.email@example.com
sender = luigi@example.com
method = smtp
host = smtp.example.com
port = 587
username = username
password = password

# 작업에 이메일 알림 추가
class CriticalTask(luigi.Task):
    email_on_failure = True
    
    def run(self):
        # 중요한 작업 수행
        pass

class RunSparkJob(luigi.contrib.external_program.ExternalProgramTask):
    spark_script = luigi.Parameter()
    
    def program_args(self):
        return ['spark-submit', self.spark_script]
    
    def output(self):
        return luigi.LocalTarget(f'output/spark_job_complete.txt')

# 작업자 수 증가
python -m luigi --module my_tasks MyTask --workers 8

# 대규모 병렬 처리를 위한 작업 분할
class ProcessLargeDataset(luigi.Task):
    def requires(self):
        # 데이터를 10개 청크로 분할
        return [ProcessChunk(chunk_id=i) for i in range(10)]

# 로깅 설정
import logging
logger = logging.getLogger('luigi-interface')

class MyTask(luigi.Task):
    def run(self):
        logger.info("작업 시작")
        try:
            # 작업 수행
            result = process_data()
            logger.info(f"처리 결과: {result}")
        except Exception as e:
            logger.error(f"오류 발생: {e}", exc_info=True)
            raise
        logger.info("작업 완료")