Spark를 가지고 S3-2에 train, test에 있는 이미지들에 대한 train.json, test.json 만들기

[wwkler]

Spark를 가지고 S3-2에 train, test에 있는 이미지들에 대한 train.json, test.json 만들기

• 더 해야할 작업
  • Kafka로부터 최신 폴더 path를 가져와서 코드를 돌려보는 과정이 필요하다.
    • 지금은 Kafka로부터 최신 폴더 path를 받았다고 가정하고 진행을 했다.
    • deprojteam07-airflowdagtest 라는 가상의 성격을 지닌 S3-2에서 kimyoungwoo-test/ 라는 최신 폴더 path를 Kafka로부터 받았다고 가정하고 진행했다.
  • 예외 처리와 같은 코드 고도화 작업이 필요하다.
  • Airflow SparkSubmitOperator를 이용해서 .py 파일을 돌려지는지에 대한 작업을 수행해야 한다.
    • 지금은 Spark Master 컨테이너 내부에 들어가서 spark-submit을 수행했다.

• 구현 내용 (위에서 아래로 갈수록 코드가 진행됩니다...)

  • spark Session 설정
  
  

  '''
  지금 상황은 Kafka로부터 최근 폴더 path를 받았다고 가정하고 진행했습니다... 
  
  Step 1. Kafka로부터 S3-2에 있는 최근 폴더 path를 받았다. 
  Step 2. 최근 폴더 path를 이용해서 하위 경로를 탐색해서 이미지에 대한 메타 데이터를 추출한다. 
  Step 3. traian.json, test.json파일을 만들고 S3-2에 적재한다. 
  '''
  
  from pyspark.sql import SparkSession 
  
  # SparkSession 설정  
  spark = SparkSession \
          .builder \
          .appName("kafka_consumer_test") \
          .config("spark.driver.host", "172.31.3.183") \
          .config("spark.driver.bindAddress", "0.0.0.0") \
          .config("spark.driver.port", "7077") \
          .config("spark.executor.instances", "2") \
          .config("spark.jars", "/opt/bitnami/spark/jars/spark-sql-kafka-0-10_2.12-3.5.0.jar,"
                                            "/opt/bitnami/spark/jars/kafka-clients-3.5.0.jar,"
                                            "/opt/bitnami/spark/jars/commons-pool2-2.11.0.jar,"
                                            "/opt/bitnami/spark/jars/metrics-core-3.2.6.jar,"
                                            "/opt/bitnami/spark/jars/spark-token-provider-kafka-0-10_2.12-3.5.1.jar") \
          .master("spark://172.31.3.183:7077") \
          .getOrCreate()
         
  
  

  
  
  • S3 설정과 train.json과 test.json을 만들기 위한 라이브러리 import
  
  

   # S3 설정과 train.json과 test.json을 만들기 위한 라이브러리 import 
   from io import BytesIO
   import json
   import math 
   from zoneinfo import ZoneInfo  
   from PIL import Image
   import boto3 
   
   s3_client = boto3.client("s3")
   bucket_name = '{{버킷 명}}'  # s3-2의 의미를 가지는 bucket 
   received_path = 'kimyoungwoo-test/'          # Kafka로부터 최근 folder_path를 받았다고 가정 
   
   train_good_folder = f'{received_path}train/good/'           # train용 good에 담긴 이미지들 파일 경로 
   train_defective_folder = f'{received_path}train/defective/' # train용 defective에 담긴 이미지들 파일 경로 
   
   test_good_folder = f'{received_path}test/good/'             # test용 good에 담긴 이미지들 파일 경로 
   test_defective_folder = f'{received_path}test/defective/'   # test용 defective에 담긴 이미지들 파일 경로 
  
  

  
  
  • Kafka로부터 받은 최신 폴더 path를 가지고 하위 경로로 탐색해서 .jpg 파일을 가져온다.
  
  

   # Kafka로부터 받은 최신 폴더 path를 가지고 하위 경로로 탐색해서 .jpg 파일을 가져오는 함수 
   def get_list_jpg_files(bucket_name, path):
       jpg_files = []
   
       # Paginator 사용하여 S3 객체 목록 가져오기
       paginator = s3_client.get_paginator('list_objects_v2')
       pages = paginator.paginate(Bucket=bucket_name, Prefix=path)
       
       for page in pages:
           if 'Contents' in page:
               # 'Contents'에서 .jpg 파일만 필터링해서 Key 값을 리스트에 추가
               for content in page['Contents']:
                   if content['Key'].endswith('.jpg'):
                       jpg_files.append(content['Key'])
       
       return jpg_files
   
   # {{Kafka로부터 받은 최신 폴더 path}}/train/good/에 있는 .jpg 파일을 가져온다. 
   train_good_images = get_list_jpg_files(bucket_name=bucket_name, path=train_good_folder)
  
   # {{Kafka로부터 받은 최신 폴더 path}}/train/defective/에 있는 .jpg 파일을 가져온다. 
   train_defective_images = get_list_jpg_files(bucket_name=bucket_name, path=train_defective_folder)
  
   # {{Kafka로부터 받은 최신 폴더 path}}/train/defective/에 있는 .jpg 파일을 가져온다.  
   test_good_images = get_list_jpg_files(bucket_name=bucket_name, path=test_good_folder)
  
   # {{Kafka로부터 받은 최신 폴더 path}}/test/defective/에 있는 .jpg 파일을 가져온다. 
   test_defective_images = get_list_jpg_files(bucket_name=bucket_name, path=test_defective_folder)
  
  
  

  
  
  • train에 있는 good, defective 그리고 test에 있는 good, defective에 있는 이미지들에 대한 메타 데이터를 만든다.
  
  

   # 이미지 파일 크기(용량)를 적절한 단위 (KB, MB, GB)로 변환하는 함수
   def calculate_image_size(size_bytes):
       if size_bytes == 0:
           return "0B"
       size_name = ("B", "KB", "MB", "GB", "TB", "PB")
       i = int(math.floor(math.log(size_bytes, 1024)))
       p = math.pow(1024, i)
       s = round(size_bytes / p, 2)
       return f"{s} {size_name[i]}"
   
   # 이미지 파일의 LastModified 시간을 한국 시간대(KST)로 변환하는 함수
   def format_last_modified_to_kst(last_modified):
       # UTC 시간대를 한국 시간대(KST)로 변환
       kst_timezone = ZoneInfo("Asia/Seoul")
   
       last_modified_kst = last_modified.astimezone(kst_timezone)
       
       # 'YYYY-MM-DD-HH:MM' 형식으로 변환
       return last_modified_kst.strftime('%Y-%m-%d-%H:%M')
   
   # 이미지의 가로/세로 크기를 추출하는 함수
   def get_image_dimensions(image_body):
       # 이미지를 메모리로 로드
       image_data = image_body.read()
       image = Image.open(BytesIO(image_data))
       
       # 이미지의 가로/세로 크기 추출
       width, height = image.size
       
       return width, height
   
   # 각 이미지에 대한 메타 데이터를 추출하는 함수 
   def make_image_metadata(bucket_name, file_key, label_value):
       
       # S3 버킷 내의 특정 이미지에 대한 메타 데이터를 가져온다. 
       response = s3_client.head_object(Bucket=bucket_name, Key=file_key)
       
       # 이미지 파일에 대한 크기(용량)을 계산한다. 
       size_in_proper_unit = calculate_image_size(response['ContentLength'])
       
       # 이미지 파일에 대한 마지막 수정 시간을 변환한다.
       last_modified_formatted = format_last_modified_to_kst(response['LastModified'])
       
       # S3에서 이미지 객체 다운로드
       response = s3_client.get_object(Bucket=bucket_name, Key=file_key)
       
       # 이미지에 대한 가로, 세로 크기를 구한다. 
       width, height = get_image_dimensions(response["Body"])
       
       metadata = {
           'FilePath': file_key,                                                      # 파일 경로
           'Size': size_in_proper_unit,                                   # 크기 (MB, GB 등의 단위로 변환)
           'LastModified': last_modified_formatted,               # 형식 변환된 마지막 수정 시간
           'Width * Height' : f"{width} * {height}",                   # 이미지 크기 
           'label_value' : label_value,                                    # 라벨값 
       }
       
   
       return metadata
   
   # Train Good 이미지들에 대한 메타 데이터를 가져온다. 
   train_good_metadata = [make_image_metadata(bucket_name=bucket_name, file_key=img,  label_value=1) for img in train_good_images]
   
   # Train Defective 이미지들에 대한 메타 데이터를 가져온다. 
   train_defective_metadata = [make_image_metadata(bucket_name=bucket_name, file_key=img, label_value=0)  for img in train_defective_images]
                                                                
                                                                                                                                                              
   
   # Test Good 이미지들에 대한 메타 데이터를 가져온다. 
   test_good_metadata = [make_image_metadata(bucket_name=bucket_name, file_key=img, label_value=1) for img in test_good_images]
                                                                                               
   
   # Test Defective 이미지들에 대한 메타 데이터를 가져온다. 
   test_defective_metadata = [make_image_metadata(bucket_name=bucket_name, file_key=img, label_value=0)   for img in test_defective_images]
                                                                                           
                                                                                              
  

  
  
  • train.json과 test.json을 만든다.
  
  

   # train에 속한 good_meta_data와 defective_meta_data를 합쳐서 train.json를 만드는 함수 
   # test에 속한 test_good_meta_data와 test_defective_meta_data를 합쳐서 test.json으로 만드는 함수 
   def create_annotation_json(ver_info, 
                              type,
                              good_metadata,
                              defective_metadata):
       # good_meta_data와 defective_meta_data를 합쳐서 data 속성에 넣는다. 
       data = good_metadata + defective_metadata
       
       # print(f"data 속성 : {data}")
       # print()
       # print()
       
       # 최종 JSON 구조를 정의한다. 
       annotation = {
           "ver_Info" : ver_info,      # 버전 정보 
           "type" : type,              # 데이터 타입(train or test) 
           "annotation" : {
               "data" : data           # 메타 데이터 리스트 
           }
       }
       
       # JSON 데이터를 문자열로 변환
       annotation_json = json.dumps(annotation, indent=4, ensure_ascii=False)
       
       # print(f"최종 json 구조 : {annotation_json}")
       
       return annotation_json
       
       
       
   # train_good_meta_data와 train_defective_meta_data를 합쳐서 train.json를 받는 함수 
   train_annotation_json = create_annotation_json(ver_info=f"{received_path}", 
                                                  type="train", 
                                                  good_metadata=train_good_metadata, 
                                                  defective_metadata=train_defective_metadata)
   
   
   # test_good_meta_data와 test_defective_meta_data를 합쳐서 test.json를 받는 함수 
   test_annotation_json = create_annotation_json(ver_info=f"{received_path}", 
                                                 type="test", 
                                                 good_metadata=test_good_metadata, 
                                                 defective_metadata=test_defective_metadata)
  
  

  
  
  
  • S3-2에 약속된 경로에 train.json과 test.json을 업로드한다.
  
  

    # 최종적으로 나온 train.json과 test.json을 S3-2에 업로드 하는 함수 
   def upload_json_to_s3(bucket_name, 
                         dest_s3_path, 
                         json_data):
       
       # S3에 업로드하기 위해 메모리에서 BytesIO로 처리
       s3_client.put_object(Bucket=bucket_name, 
                            Key=dest_s3_path, 
                            Body=BytesIO(json_data.encode('utf-8')),
                            ContentType='application/json')
   
       print(f"JSON 데이터가 S3의 '{dest_s3_path}' 경로에 업로드되었습니다.")
   
   
   # S3-2 어떤 경로에 train.json을 저장할 것인지 정의한다. 
   dest_s3_path1 = f"{received_path}train/train.json"
   
   # S3-2 어떤 경로에 test.json을 저장할 것인지 정의한다. 
   dest_s3_path2 = f"{received_path}test/test.json"
   
   # S3-2에 train.json을 업로드한다. 
   upload_json_to_s3(bucket_name=bucket_name,
                     dest_s3_path=dest_s3_path1,
                     json_data=train_annotation_json)
   
   # S3-2에 test.json을 업로드한다. 
   upload_json_to_s3(bucket_name=bucket_name, 
                     dest_s3_path=dest_s3_path2,
                     json_data=test_annotation_json)
      
  

  
  

• train.json 결과

  {
    "ver_Info": "kimyoungwoo-test/",
    "type": "train",
    "annotation": {
        "data": [
            {
                "FilePath": "kimyoungwoo-test/train/good/good_00000004.jpg",
                "Size": "424.38 KB",
                "LastModified": "2024-09-12-09:22",
                "Width * Height": "1557 * 1557",
                "label_value": 1
            },
            {
                "FilePath": "kimyoungwoo-test/train/good/good_00000009.jpg",
                "Size": "469.97 KB",
                "LastModified": "2024-09-12-09:22",
                "Width * Height": "1590 * 1590",
                "label_value": 1
            },
            {
                "FilePath": "kimyoungwoo-test/train/defective/defective_00000001.jpg",
                "Size": "434.98 KB",
                "LastModified": "2024-09-12-09:23",
                "Width * Height": "748 * 748",
                "label_value": 0
            },
            {
                "FilePath": "kimyoungwoo-test/train/defective/defective_00000008.jpg",
                "Size": "125.81 KB",
                "LastModified": "2024-09-12-09:23",
                "Width * Height": "379 * 379",
                "label_value": 0
            }
        ]
    }
}

• test.json 결과

  {
    "ver_Info": "kimyoungwoo-test/",
    "type": "test",
    "annotation": {
        "data": [
            {
                "FilePath": "kimyoungwoo-test/test/good/good_00000025.jpg",
                "Size": "797.92 KB",
                "LastModified": "2024-09-12-11:20",
                "Width * Height": "2258 * 2258",
                "label_value": 1
            },
            {
                "FilePath": "kimyoungwoo-test/test/good/good_00000034.jpg",
                "Size": "1.11 MB",
                "LastModified": "2024-09-12-11:20",
                "Width * Height": "2394 * 2394",
                "label_value": 1
            },
            {
                "FilePath": "kimyoungwoo-test/test/defective/defective_00001021.jpg",
                "Size": "41.54 KB",
                "LastModified": "2024-09-12-11:21",
                "Width * Height": "242 * 242",
                "label_value": 0
            },
            {
                "FilePath": "kimyoungwoo-test/test/defective/defective_00001027.jpg",
                "Size": "68.4 KB",
                "LastModified": "2024-09-12-11:21",
                "Width * Height": "305 * 305",
                "label_value": 0
            }
        ]
    }
}

[Kdreamtomaster]

test 용 kafka-producer Dag 입니다

apache kafka producer operator 를 쓰려고 했으나 오류가 종잡을 수 없어서 python operator로 하니까 되었습니다.

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

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

dag = DAG(
    'kafka_producer_python_dag',
    default_args=default_args,
    description='A simple Kafka producer DAG',
    schedule_interval=timedelta(days=1),
)

def send_message_to_kafka(**kwargs):
    producer = Producer({
        'bootstrap.servers': '172.31.3.183:9092,172.31.2.37:9092,172.31.0.17:9092',
    })
    producer.produce('test', 'kimyoungwoo-test'.encode('utf-8'))
    producer.flush()

send_message_task = PythonOperator(
    task_id='send_message_to_kafka',
    python_callable=send_message_to_kafka,
    dag=dag,
)