airflow sensor 를 활용하여 s3 변경사항 확인 하는 방법 조사

[Kdreamtomaster]

• 버전 변경을 확인하려면 boto3를 활용하여 python operator 써야함
• 어떤 경로에 어떤 확장자의 파일이 있는지 없는지 체크 할 수 있음, 다만 어떤 경로가 새로생겼는지는 확인 불가
• XCOM 을 활용하거나 별도의 파일 명부를 관리해야 이 파일이 이전에 있었는지 없었는지 확인가능

[Kdreamtomaster]

우선 코치님이 EC2에 부여해주신 IAM 역할 또는 권한으로 컨테이너 airflow에서 dag으로 S3 파일 업로드가 가능한지 테스트 해봤습니다

관련 테스트 코드 입니다.

관련하여 airflow dag 테스트도 해보았습니다.
메모리 사용량을 txt파일로 만들어 s3 로 전송하는 샘플코드입니다.

코드 보기/닫기

import os
import subprocess
import boto3
from datetime import datetime,timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.utils.dates import days_ago

# S3 버킷 이름
S3_BUCKET_NAME = 'BUCKET_NAME 수정해서 쓰시오'

# S3에 파일 업로드하는 함수
def upload_file_to_s3(file_path, bucket_name, object_name):
    s3 = boto3.client('s3')
    s3.upload_file(file_path, bucket_name, object_name)
    print(f"File {file_path} uploaded to S3 bucket {bucket_name} as {object_name}")

# free -h 명령어 실행하고 파일에 저장한 후 S3에 업로드하는 함수
def generate_and_upload_file():
    # 현재 날짜 및 시간 가져오기
    now = datetime.now()
    file_name = now.strftime("%Y%m%d_%H%M%S") + '.txt'
    file_path = f'/tmp/{file_name}'

    # free -h 명령어 실행
    command = ['free', '-h']
    result = subprocess.run(command, capture_output=True, text=True)

    # 명령어 결과를 파일에 저장
    with open(file_path, 'w') as f:
        f.write(result.stdout)

    # S3에 파일 업로드
    upload_file_to_s3(file_path, S3_BUCKET_NAME, file_name)

    # 업로드 후 파일 삭제
    os.remove(file_path)

# DAG 정의
default_args = {
    'owner': 'airflow',
    'start_date': datetime(2024, 9, 9, 0, 0),  # 특정 시작 시간 설정
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 생성
dag = DAG(
    'memory_usage_to_s3',
    default_args=default_args,
    description='DAG to save free -h output and upload to S3',
    schedule_interval='@hourly',  # 매시간 실행
)

# PythonOperator를 사용하여 주기적으로 실행
task = PythonOperator(
    task_id='generate_and_upload_file_task',
    python_callable=generate_and_upload_file,
    dag=dag,
)

• 테스트로 인입된 파일들

• 파일 내용 예시

[Kdreamtomaster]

코드 보기 숨기기

import boto3
from confluent_kafka import Producer
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
import json
import os

# S3 및 Kafka 설정
BUCKET_NAME = 'BUCKET_NAME 수정해서 쓰시오'
PARENT_FOLDER_PATH = 'my-folder/'  # S3의 상위 폴더 경로
KAFKA_BROKER = '172.31.3.183:9092,172.31.2.37:9092,172.31.0.17:9092'
KAFKA_TOPIC = 'test'

# boto3 S3 클라이언트 생성
s3_client = boto3.client('s3')

# Confluent Kafka Producer 설정
conf = {
    'bootstrap.servers': KAFKA_BROKER,
    'client.id': 'airflow-producer'
}
producer = Producer(conf)

# 이전에 감지된 폴더를 기록하기 위한 파일 경로
FOLDER_TRACK_FILE = '/opt/airflow/dags/last_folders.json'

# S3에서 폴더를 감지하는 함수
def check_for_new_folders(**kwargs):
    # S3의 상위 경로에서 하위 폴더를 검사
    response = s3_client.list_objects_v2(Bucket=BUCKET_NAME, Prefix=PARENT_FOLDER_PATH, Delimiter='/')
    
    # CommonPrefixes에 하위 폴더들이 포함됨
    current_folders = [prefix['Prefix'] for prefix in response.get('CommonPrefixes', [])]

    # 이전에 감지된 폴더 목록 불러오기
    try:
        with open(FOLDER_TRACK_FILE, 'r') as f:
            last_folders = json.load(f)
    except FileNotFoundError:
        last_folders = []

    # 새로 생성된 폴더 감지 (현재 폴더 목록에서 이전 폴더 목록을 제외)
    new_folders = list(set(current_folders) - set(last_folders))

    # 이전 폴더 목록을 업데이트
    with open(FOLDER_TRACK_FILE, 'w') as f:
        json.dump(current_folders, f)

    # 새로운 폴더가 감지되었으면 반환 (XCom에 전달)
    if new_folders:
        return new_folders
    else:
        return None

# Kafka로 새 폴더 경로 전송 (예외 처리 추가)
def send_to_kafka(new_folders, **kwargs):
    if len(new_folders) == 0:
        raise ValueError("No new folders to process. Expected non-empty list.")

    try:
        for folder in new_folders:
            # Kafka로 메시지 전송
            message = {
                'bucket_name': BUCKET_NAME,
                'folder_path': folder
            }
            producer.produce(KAFKA_TOPIC, key=None, value=json.dumps(message))
        
        # 모든 메시지 전송 후 한 번만 flush
        producer.flush()
        print(f"Sent {len(new_folders)} messages to Kafka.")

    except Exception as e:
        # 예외 처리: 로그 출력 및 Airflow에 오류 반환
        print(f"Failed to send message to Kafka: {e}")
        raise ValueError(f"Kafka 전송 실패: {str(e)}")  # 예외를 Airflow로 전달하여 DAG 실행 실패로 처리

# DAG 정의
with DAG(
    dag_id='s3_folder_to_kafka_dag_2',
    start_date=datetime(2024, 9, 9),  # 2024년 9월 9일 시작
    schedule_interval=timedelta(minutes=30),  # 10분마다 실행
    catchup=False,
    default_args={
        'retries': 3,  # 최대 3회 재시도
        'retry_delay': timedelta(minutes=10)  # 재시도 간 2분 대기
    }
) as dag:

    # 새로운 폴더 감지 작업
    check_s3_folders = PythonOperator(
        task_id='check_s3_folders',
        python_callable=check_for_new_folders,
        provide_context=True
    )

    # Kafka로 폴더 경로 전송 작업
    send_folders_to_kafka = PythonOperator(
        task_id='send_folders_to_kafka',
        python_callable=send_to_kafka,
        op_args=["{{ task_instance.xcom_pull(task_ids='check_s3_folders') }}"],
        provide_context=True,
        retries=1,  # 최대 1번만 재시도
        retry_delay=timedelta(minutes=2),  # 재시도 시 2분 대기
        execution_timeout=timedelta(minutes=5)  # 5분 내에 완료되지 않으면 실패 처리
    )

    # 작업 순서 정의: 폴더 검사 후 Kafka로 경로 전송
    check_s3_folders >> send_folders_to_kafka

일단 이렇게 작성은 하였으나 send_folders_to_kafka 가 작동을 안하는 무한 루프에 빠져있음.

지금 까지 해본것

• docker-compose.yml 에 x-airflow-common: 하위 environment: 에 AIRFLOW__WEBSERVER__SECRET_KEY: 항목추가 -> 로그에 관련에러가 있었고 넣으니 없어짐
• 카프카 연결에 문제가 있나 싶어 호스트 네임으로 해둔걸 프라이빗 아이피로 변경하였으나 변동없음
• 그밖에 도커 컴포즈를 수정해보았으나 소용없었음
• 결국 DAG 코드의 문제인데 아직 찾지 못함

[Kdreamtomaster]

S3 변경사항을 확인하고 디렉토리 구조도 바꾼뒤에 다른 S3에 업로드 하고 해당 경로를 DB 에 보낸다.

코드 접기/펼치기

• G 선생님 초안( 수정중 )

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import os
import shutil

# 기본 설정
default_args = {
    'owner': 'user',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    's3_to_s3_and_log_to_postgres',
    default_args=default_args,
    description='Process data from S3-1 to S3-2 and log to Postgres',
    schedule_interval='*/15 * * * *',
    start_date=datetime(2024, 9, 10),
    catchup=False,
)

# Task 1: S3-1에서 새로운 폴더 선택
def select_latest_folder():
    s3 = S3Hook(aws_conn_id='aws_default')
    bucket = 'your-s3-1-bucket'
    prefix = 'split_'  # 폴더 prefix로 split_{n} 형식
    
    # 모든 split 폴더 목록 가져오기
    all_folders = s3.list_prefixes(bucket_name=bucket, prefix=prefix)
    
    # S3-1에 폴더가 없는 경우 처리
    if not all_folders:
        raise ValueError("S3-1 버킷에 폴더가 없습니다.")
    
    # Postgres에서 이전에 처리한 폴더 목록 가져오기
    postgres = PostgresHook(postgres_conn_id='postgres_default')
    query = "SELECT path FROM FOLDER_LOG"
    processed_folders = [row[0] for row in postgres.get_records(query)]

    # 새로운 폴더만 선택
    new_folders = [folder for folder in all_folders if folder not in processed_folders]
    
    # 새로운 폴더가 없을 경우 예외 처리
    if not new_folders:
        raise ValueError("새로운 폴더가 없습니다.")
    
    # 가장 최신의 폴더 반환
    latest_folder = max(new_folders)
    return latest_folder

select_latest_folder_task = PythonOperator(
    task_id='select_latest_folder',
    python_callable=select_latest_folder,
    dag=dag,
)

# Task 2: S3-1에서 이미지 데이터를 가져와 S3-2로 디렉토리 재구성 후 업로드
def restructure_and_upload_to_s3(ti):
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    
    if not latest_folder:
        raise ValueError("처리할 폴더가 없습니다.")

    s3_bucket_1 = 'your-s3-1-bucket'
    s3_bucket_2 = 'your-s3-2-bucket'
    
    s3 = S3Hook(aws_conn_id='aws_default')
    
    local_dir = '/tmp/s3_restructured'
    if not os.path.exists(local_dir):
        os.makedirs(local_dir)

    # Step 1: 최신 split_{n} 데이터 다운로드
    sub_folders = ['test/good/', 'test/defective/', 'train/good/', 'train/defective/']
    for sub_folder in sub_folders:
        folder_path = os.path.join(latest_folder, sub_folder)
        files = s3.list_keys(bucket_name=s3_bucket_1, prefix=folder_path)
        
        for file_key in files:
            file_name = os.path.basename(file_key)
            local_file_path = os.path.join(local_dir, sub_folder, file_name)
            os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
            s3.download_file(bucket_name=s3_bucket_1, key=file_key, local_path=local_file_path)

    # Step 2: S3-2에 ver_{n} 디렉토리 구조 생성
    ver_n = 1  # 초기 ver_1. 이미 존재하는 경우에 맞춰 업데이트 가능
    base_ver_path = f'ver_{ver_n}'
    for sub_folder in sub_folders:
        s3.upload_directory(os.path.join(local_dir, sub_folder), s3_bucket_2, os.path.join(base_ver_path, sub_folder))

    # Step 3: ver_{n+1} 폴더 복사 및 split_{n+1} 데이터 추가
    ver_n_plus_1 = f'ver_{ver_n + 1}'
    for sub_folder in sub_folders:
        # ver_{n} 데이터를 ver_{n+1}로 복사
        s3.copy_directory(s3_bucket_2, os.path.join(base_ver_path, sub_folder), os.path.join(ver_n_plus_1, sub_folder))
    
        # 새로운 split_{n+1} 데이터 추가
        new_split_folder = f'split_{ver_n+1}/{sub_folder}'
        new_files = s3.list_keys(bucket_name=s3_bucket_1, prefix=new_split_folder)
        
        for file_key in new_files:
            file_name = os.path.basename(file_key)
            local_file_path = os.path.join(local_dir, sub_folder, file_name)
            os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
            s3.download_file(bucket_name=s3_bucket_1, key=file_key, local_path=local_file_path)
            
            new_s3_key = os.path.join(ver_n_plus_1, sub_folder, file_name)
            s3.upload_file(local_path=local_file_path, bucket_name=s3_bucket_2, key=new_s3_key)

    shutil.rmtree(local_dir)

restructure_and_upload_task = PythonOperator(
    task_id='restructure_and_upload_to_s3',
    python_callable=restructure_and_upload_to_s3,
    provide_context=True,
    dag=dag,
)

# Task 3: Postgres DB에 로그 기록
def log_new_folder_in_db(ti):
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    if not latest_folder:
        raise ValueError("폴더 경로가 없습니다.")
    
    postgres = PostgresHook(postgres_conn_id='postgres_default')
    
    # SQL 쿼리 작성
    sql = """
    INSERT INTO FOLDER_LOG (path)
    VALUES (%s)
    """
    
    # 최신 폴더 경로를 DB에 기록
    postgres.run(sql, parameters=[latest_folder])

log_new_folder_task = PythonOperator(
    task_id='log_new_folder_in_db',
    python_callable=log_new_folder_in_db,
    provide_context=True,
    dag=dag,
)

# Task 4: Postgres DB에 데이터 적재
def insert_data_into_db(ti):
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    if not latest_folder:
        raise ValueError("폴더 경로가 없습니다.")
    
    postgres = PostgresHook(postgres_conn_id='postgres_default')
    
    # SQL 쿼리 작성
    sql = """
    INSERT INTO FOLDER_LOG (path)
    VALUES (%s)
    """
    
    # 최신 폴더 값을 삽입
    postgres.run(sql, parameters=[latest_folder])

insert_data_task = PythonOperator(
    task_id='insert_data_into_db',
    python_callable=insert_data_into_db,
    provide_context=True,
    dag=dag,
)

# DAG의 태스크들 간의 종속성 설정
select_latest_folder_task >> restructure_and_upload_task >> log_new_folder_task >> insert_data_task

[Kdreamtomaster]

원본 path 를 가져와서 해야함

폴더 만드는 예시

import boto3
s3 = boto3.client('s3')
bucket_name = "YOUR-BUCKET-NAME"
directory_name = "DIRECTORY/THAT/YOU/WANT/TO/CREATE" #it's name of your folders
s3.put_object(Bucket=bucket_name, Key=(directory_name+'/'))

[Kdreamtomaster]

수정 중인 코드 접기/펼치기

from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.providers.amazon.aws.hooks.s3 import S3Hook
from airflow.providers.postgres.hooks.postgres import PostgresHook
from datetime import datetime, timedelta
import os
import logging

# 기본 설정
default_args = {
    'owner': 'user',
    'depends_on_past': False,
    'email_on_failure': False,
    'email_on_retry': False,
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    's3_to_s3_copy_log_to_postgres',
    default_args=default_args,
    description='Process data from S3-1 to S3-2 and log to Postgres',
    schedule_interval='*/15 * * * *',
    start_date=datetime(2024, 9, 10),
    catchup=False,
)

# Task 1: S3-1에서 새로운 폴더 선택
def select_latest_folder():
    s3 = S3Hook(aws_conn_id='aws_default')
    bucket = 'deprojteam07-labeledrawdata'
    prefix = 'ver_'  # 폴더 prefix로 ver_{n} 형식
    
    # 모든 ver_{n} 폴더 목록 가져오기
    all_folders = s3.list_prefixes(bucket_name=bucket, prefix=prefix)
    
    # S3-1에 폴더가 없는 경우 처리
    if not all_folders:
        raise ValueError("rawdata 버킷에 폴더가 없습니다.")
    
    # Postgres에서 이전에 처리한 폴더 목록 가져오기
    postgres = PostgresHook(postgres_conn_id='PostgresDataLoading')
    query = "SELECT path FROM FOLDER_LOG"
    processed_folders = [row[0] for row in postgres.get_records(query)]

    # 새로운 폴더만 선택
    new_folders = [folder for folder in all_folders if folder not in processed_folders]
    
    # 새로운 폴더가 없을 경우 예외 처리
    if not new_folders:
        raise ValueError("새로운 폴더가 없습니다.")
    
    # 가장 최신의 폴더 반환
    latest_folder = max(new_folders)
    return latest_folder

select_latest_folder_task = PythonOperator(
    task_id='select_latest_folder',
    python_callable=select_latest_folder,
    dag=dag,
)

# Task 2: S3-1에서 이미지 데이터를 가져와 S3-2로 디렉토리 재구성 후 업로드 (S3 to S3 copy)
def restructure_and_copy_s3_to_s3(**kwargs):
    ti = kwargs['ti']
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    
    if not latest_folder:
        raise AirflowSkipException("No folder to process.")

    s3_hook = S3Hook(aws_conn_id='aws_default')
    
    sub_folders = ['train/good/', 'train/defective/', 'test/good/', 'test/defective/']
    
    try:
        for sub_folder in sub_folders:
            folder_path = os.path.join(latest_folder, sub_folder)
            logging.info(f"Checking folder: {folder_path}")
            keys = s3_hook.list_keys(bucket_name=S3_BUCKET_SOURCE, prefix=folder_path)
            
            if not keys:
                logging.warning(f"No files found in {folder_path}")
                continue
            
            logging.info(f"Found {len(keys)} files in {folder_path}")
            for key in keys:
                if key.endswith('/'):  # Skip directories
                    continue
                try:
                    s3_hook.copy_object(
                        source_bucket_key=key,
                        dest_bucket_key=key,
                        source_bucket_name=S3_BUCKET_SOURCE,
                        dest_bucket_name=S3_BUCKET_DEST
                    )
                    logging.info(f"Successfully copied: {key}")
                except Exception as copy_error:
                    logging.error(f"Failed to copy {key}: {str(copy_error)}")
        
        logging.info(f"Successfully processed all files from {latest_folder}")
    except Exception as e:
        logging.error(f"Error in restructure_and_copy_s3_to_s3: {str(e)}")
        raise AirflowException(f"Failed to copy files: {str(e)}")

restructure_and_copy_task = PythonOperator(
    task_id='restructure_and_copy_s3_to_s3',
    python_callable=restructure_and_copy_s3_to_s3,
    provide_context=True,
    dag=dag,
)

# Task 3: Postgres DB에 로그 기록
def log_new_folder_in_db(ti):
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    if not latest_folder:
        raise ValueError("폴더 경로가 없습니다.")
    
    postgres = PostgresHook(postgres_conn_id='PostgresDataLoading')
    
    # SQL 쿼리 작성
    sql = """
    INSERT INTO FOLDER_LOG (path)
    VALUES (%s)
    """
    
    # 최신 폴더 경로를 DB에 기록
    postgres.run(sql, parameters=[latest_folder])

log_new_folder_task = PythonOperator(
    task_id='log_new_folder_in_db',
    python_callable=log_new_folder_in_db,
    provide_context=True,
    dag=dag,
)

# Task 4: Postgres DB에 데이터 적재
def insert_data_into_db(ti):
    latest_folder = ti.xcom_pull(task_ids='select_latest_folder')
    if not latest_folder:
        raise ValueError("폴더 경로가 없습니다.")
    
    postgres = PostgresHook(postgres_conn_id='PostgresDataLoading')
    
    # SQL 쿼리 작성
    sql = """
    INSERT INTO FOLDER_LOG (path)
    VALUES (%s)
    """
    
    # 최신 폴더 값을 삽입
    postgres.run(sql, parameters=[latest_folder])

insert_data_task = PythonOperator(
    task_id='insert_data_into_db',
    python_callable=insert_data_into_db,
    provide_context=True,
    dag=dag,
)

# DAG의 태스크들 간의 종속성 설정
select_latest_folder_task >> restructure_and_copy_task >> log_new_folder_task >> insert_data_task

• db에 인입하는 것까지 테스트 중입니다. 아직 실패중

[Kdreamtomaster]

dag으로 만들기 이전에 python 코드로 먼저 테스트 진행중

• 컨테이너 환경 , 이미지는 python3.12-slim

0911 13시 기준테스트 완료한것

• 수정시각 기준으로 가장 최근 폴더를 찾아내서 이전에 기록된 최신 폴더와 다를 경우 복사 진행

가장 최근 S3 폴더 체크해서 복사하는 파이썬 코드

import boto3
import os

# S3 클라이언트 생성
s3 = boto3.client('s3')

# 히스토리 파일 경로 설정
HISTORY_FILE = '/app/history/latest_folder_history.txt'

def get_latest_versioned_folder(bucket_name):
    response = s3.list_objects_v2(Bucket=bucket_name, Delimiter='/')
    
    # 최상위 폴더들만 가져옴 (Prefix가 "testver_"로 시작하는 것들 필터링)
    top_folders = [prefix['Prefix'] for prefix in response.get('CommonPrefixes', []) if prefix['Prefix'].startswith('testver_')]
    
    if not top_folders:
        return None
    
    # 최상위 폴더들을 가장 최근 생성된 순으로 정렬
    top_folders_with_dates = []
    for folder in top_folders:
        response = s3.list_objects_v2(Bucket=bucket_name, Prefix=folder, MaxKeys=1)
        if 'Contents' in response:
            top_folders_with_dates.append((folder, response['Contents'][0]['LastModified']))
    
    if not top_folders_with_dates:
        return None
    
    # 가장 최근 생성된 폴더를 찾음
    latest_folder = max(top_folders_with_dates, key=lambda x: x[1])
    
    return latest_folder[0]

def save_history(latest_folder):
    with open(HISTORY_FILE, 'w') as f:
        f.write(latest_folder)

def load_history():
    # 파일이 존재하지 않거나 비어 있으면 None 반환
    if not os.path.exists(HISTORY_FILE):
        return None
    
    with open(HISTORY_FILE, 'r') as f:
        content = f.read().strip()
        if not content:
            return None
        return content

# S3 객체 복사 함수
def copy_directory(source_bucket, source_prefix, destination_bucket, destination_prefix):
    # source_prefix: 원본 버킷에서 복사하고자 하는 디렉토리 경로
    # destination_prefix: 복사할 대상 디렉토리 경로
    
    # 해당 디렉토리와 하위 객체 목록 가져오기
    paginator = s3.get_paginator('list_objects_v2')
    pages = paginator.paginate(Bucket=source_bucket, Prefix=source_prefix)

    for page in pages:
        if 'Contents' in page:
            for obj in page['Contents']:
                # 객체의 키 (경로)
                source_key = obj['Key']
                
                # 대상 키 설정 (디렉토리 구조 유지)
                destination_key = source_key.replace(source_prefix, destination_prefix, 1)
                
                # 객체 복사
                copy_source = {
                    'Bucket': source_bucket,
                    'Key': source_key
                }
                
                s3.copy_object(
                    CopySource=copy_source,
                    Bucket=destination_bucket,
                    Key=destination_key
                )
                
                print(f"Copied {source_key} to {destination_key}")

# 새로운 폴더를 찾으면 복사하는 로직
def check_and_copy_new_folder(bucket_name, destination_bucket):
    latest_folder = get_latest_versioned_folder(bucket_name)

    # 이전에 기록된 폴더를 불러옴
    previous_folder = load_history()

    if previous_folder is None:
        # 히스토리가 없으면 현재 최신 폴더를 기록
        print(f"히스토리가 없습니다. 최초 실행으로 최신 폴더를 기록합니다: {latest_folder}")
        save_history(latest_folder)
    elif previous_folder == latest_folder:
        print(f"이 폴더({latest_folder})는 이전에도 존재했습니다.")
    else:
        print(f"새로운 폴더가 발견되었습니다: {latest_folder}")
        
        # 새 폴더 복사
        destination_prefix = latest_folder
        copy_directory(bucket_name, latest_folder, destination_bucket, destination_prefix)
        
        # 복사 후 히스토리 업데이트
        save_history(latest_folder)

# 실행 부분
bucket_name = 'BUCKET_NAME 수정해서 쓰시오'
destination_bucket = 'BUCKET_NAME 수정해서 쓰시오'

check_and_copy_new_folder(bucket_name, destination_bucket)

Dockerfile

FROM python:3.12-slim

# 작업 디렉토리 설정
WORKDIR /app

# 필요한 Python 패키지 설치
RUN pip install boto3

# 애플리케이션 파일 복사
COPY app.py .

# history 디렉토리 생성
RUN mkdir -p /app/history

# Python 애플리케이션 실행
CMD ["python", "app.py"]

docker-compose.yml

version: '3'
services:
  s3-app:
    build: .
    volumes:
      - /home/ubuntu/test/test_folder_check/test_folder_volume:/app/history

[Kdreamtomaster]

#22

0911 18시 기준 수정코드 (airflow dag 이름: s3_sensor_to_db_dag_ver2)

• 폴더 변경사항 확인하기 위한 임시 텍스트 파일이 아닌 DB 조회통해서 하기

DAG 코드 보기 s3_sensor_to_db_dag_ver2.py

import os
import boto3
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.hooks.postgres_hook import PostgresHook
from datetime import datetime, timedelta
import pytz

# S3 클라이언트 생성
s3 = boto3.client('s3')

# 한국 시간대 설정 (UTC+9)
KST = pytz.timezone('Asia/Seoul')

default_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime(2024, 9, 11),
    'retries': 1,
    'retry_delay': timedelta(minutes=5),
}

# DAG 정의
dag = DAG(
    's3_sensor_to_db_dag_ver2',
    default_args=default_args,
    description='S3에서 데이터를 복사하고 DB에 삽입하는 DAG',
    schedule_interval=timedelta(days=1),
    catchup=False,
)

# 최신 폴더 찾기 함수 (S3)
def get_latest_versioned_folder(bucket_name):
    response = s3.list_objects_v2(Bucket=bucket_name, Delimiter='/')
    
    top_folders = [prefix['Prefix'] for prefix in response.get('CommonPrefixes', []) if prefix['Prefix'].startswith('testver_')]
    
    if not top_folders:
        return None
    
    top_folders_with_dates = []
    for folder in top_folders:
        response = s3.list_objects_v2(Bucket=bucket_name, Prefix=folder, MaxKeys=1)
        if 'Contents' in response:
            top_folders_with_dates.append((folder, response['Contents'][0]['LastModified']))
    
    if not top_folders_with_dates:
        return None
    
    latest_folder = max(top_folders_with_dates, key=lambda x: x[1])
    
    return latest_folder[0]

# PostgreSQL에서 가장 최근 폴더 정보 가져오기
def get_latest_folder_from_db():
    try:
        # PostgresHook 사용
        pg_hook = PostgresHook(postgres_conn_id='postgres_default')
        connection = pg_hook.get_conn()
        cursor = connection.cursor()
        
        # 가장 최근 폴더 경로를 가져오는 SQL 쿼리
        cursor.execute("SELECT path FROM folder_log ORDER BY id DESC LIMIT 1")
        result = cursor.fetchone()
        cursor.close()

        if result:
            return result[0]  # 가장 최근 폴더의 path 값 반환
        return None
    except Exception as error:
        print(f"Error fetching latest folder from PostgreSQL: {error}")
        return None

# 새로운 폴더가 있는지 확인하는 Task
def task_check_for_new_folder(**kwargs):
    bucket_name = kwargs['bucket_name']
    
    # S3에서 최신 폴더 확인
    latest_folder_s3 = get_latest_versioned_folder(bucket_name)
    
    # PostgreSQL에서 최신 폴더 확인
    latest_folder_db = get_latest_folder_from_db()
    
    if latest_folder_s3 and latest_folder_s3 != latest_folder_db:
        kwargs['ti'].xcom_push(key='new_folder', value=latest_folder_s3)
        print(f"새로운 폴더가 발견되었습니다: {latest_folder_s3}")
    else:
        print("새로운 폴더가 없습니다.")

# S3 객체 복사 Task
def task_copy_directory(**kwargs):
    source_bucket = kwargs['source_bucket']
    destination_bucket = kwargs['destination_bucket']
    new_folder = kwargs['ti'].xcom_pull(task_ids='check_for_new_folder', key='new_folder')
    
    if new_folder:
        paginator = s3.get_paginator('list_objects_v2')
        pages = paginator.paginate(Bucket=source_bucket, Prefix=new_folder)

        for page in pages:
            if 'Contents' in page:
                for obj in page['Contents']:
                    source_key = obj['Key']
                    destination_key = source_key.replace(new_folder, new_folder, 1)
                    copy_source = {'Bucket': source_bucket, 'Key': source_key}
                    
                    s3.copy_object(
                        CopySource=copy_source,
                        Bucket=destination_bucket,
                        Key=destination_key
                    )
                    print(f"Copied {source_key} to {destination_key}")

# PostgreSQL에 데이터 삽입 Task (새로운 폴더 정보 기록)
def task_insert_db(**kwargs):
    new_folder = kwargs['ti'].xcom_pull(task_ids='check_for_new_folder', key='new_folder')
    
    if new_folder:
        # PostgresHook 사용
        pg_hook = PostgresHook(postgres_conn_id='postgres_default')
        
        try:
            current_time_kst = datetime.now(KST)
            insert_query = """
                INSERT INTO folder_log (path, created_at)
                VALUES (%s, %s);
            """
            pg_hook.run(insert_query, parameters=(new_folder, current_time_kst))
            print(f"Inserted {new_folder} into folder_log")
        except Exception as error:
            print(f"Error inserting into PostgreSQL: {error}")

# Task 정의
t1 = PythonOperator(
    task_id='check_for_new_folder',
    python_callable=task_check_for_new_folder,
    op_kwargs={'bucket_name': 'BUCKET_NAME 수정해서 쓰시오'},
    provide_context=True,
    dag=dag,
)

t2 = PythonOperator(
    task_id='copy_directory',
    python_callable=task_copy_directory,
    op_kwargs={'source_bucket': 'BUCKET_NAME 수정해서 쓰시오', 'destination_bucket': 'BUCKET_NAME 수정해서 쓰시오'},
    provide_context=True,
    dag=dag,
)

t3 = PythonOperator(
    task_id='insert_into_db',
    python_callable=task_insert_db,
    provide_context=True,
    dag=dag,
)

# Task 순서 설정
t1 >> t2 >> t3