Реалізація Serverless File Processing (Lambda + S3 trigger)

Реалізація Serverless File Processing (Lambda + S3 trigger)

Lambda + S3 trigger — класичний serverless паттерн для обробки файлів. Файл завантажується в S3, подія автоматично запускає Lambda, яка обробляє файл і зберігає результат. Немає постійно запущеного сервера, автоматичне масштабування.

Типові сценарії

• Генерація thumbnail при завантаженні зображення
• Конвертація відео в різні формати та розрізнення
• Обробка CSV/Excel файлів, імпорт даних в БД
• PDF-генерація з шаблонів
• Антивірусне сканування завантажених файлів
• OCR та видобування тексту з документів
• Трансформація даних (XML → JSON, нормалізація)

Базова архітектура

[Користувач] → S3 upload → [S3 Event Notification]
                                    ↓
                            [Lambda Function]
                                    ↓
                    [Обробленвихід → S3 Output]
                    [Метадані → DynamoDB]
                    [Нотифікація → SQS/SNS]

# S3 bucket для вхідних файлів
resource "aws_s3_bucket" "uploads" {
  bucket = "myapp-uploads"
}

# S3 bucket для обробленихфайлів
resource "aws_s3_bucket" "processed" {
  bucket = "myapp-processed"
}

# Lambda сповіщення від S3
resource "aws_s3_bucket_notification" "upload_trigger" {
  bucket = aws_s3_bucket.uploads.id

  lambda_function {
    lambda_function_arn = aws_lambda_function.processor.arn
    events              = ["s3:ObjectCreated:*"]
    filter_prefix       = "images/"  # Тільки файли в цій папці
    filter_suffix       = ".jpg"     # Тільки JPG файлі
  }
}

Lambda обробник

import boto3
import json
import os
from urllib.parse import unquote_plus
from PIL import Image
import io

s3 = boto3.client('s3')

def handler(event, context):
    results = []

    for record in event['Records']:
        bucket = record['s3']['bucket']['name']
        key = unquote_plus(record['s3']['object']['key'])

        try:
            result = process_image(bucket, key)
            results.append({'key': key, 'status': 'success', **result})
        except Exception as e:
            print(f"Error processing {key}: {e}")
            results.append({'key': key, 'status': 'error', 'error': str(e)})

    return results

def process_image(bucket: str, key: str) -> dict:
    # Скачати оригінал
    obj = s3.get_object(Bucket=bucket, Key=key)
    image_data = obj['Body'].read()

    image = Image.open(io.BytesIO(image_data))

    thumbnails = {}
    for size_name, (width, height) in [('sm', (150, 150)), ('md', (400, 400)), ('lg', (800, 800))]:
        thumb = image.copy()
        thumb.thumbnail((width, height), Image.LANCZOS)

        buffer = io.BytesIO()
        thumb.save(buffer, format=image.format or 'JPEG', quality=85)
        buffer.seek(0)

        output_key = key.replace('images/', f'thumbnails/{size_name}/')
        s3.put_object(
            Bucket=os.environ['OUTPUT_BUCKET'],
            Key=output_key,
            Body=buffer,
            ContentType=f'image/{(image.format or "JPEG").lower()}'
        )
        thumbnails[size_name] = output_key

    return {'thumbnails': thumbnails, 'original_size': image.size}

Обробка великих файлів

Lambda має обмеження: /tmp до 10GB, timeout до 15 хвилин, пам'ять до 10GB. Для файлів >100MB — потокова обробка:

import boto3
import csv
import io

def process_large_csv(bucket: str, key: str):
    s3 = boto3.client('s3')

    # StreamingBody — читати по частинах без завантаження в пам'ять
    obj = s3.get_object(Bucket=bucket, Key=key)

    batch = []
    batch_size = 1000

    for line in obj['Body'].iter_lines():
        row = line.decode('utf-8')
        batch.append(parse_csv_row(row))

        if len(batch) >= batch_size:
            save_batch_to_db(batch)
            batch = []

    if batch:
        save_batch_to_db(batch)

Для трансодування відео — використовувати AWS MediaConvert або Elastic Transcoder замість Lambda (не обмежені часом).

Обробка помилок та DLQ

S3 event notifications не підтримують DLQ безпосередньо. Надійна схема:

S3 → SNS Topic → SQS Queue → Lambda
                     ↓ (після maxReceiveCount)
                   SQS DLQ

resource "aws_s3_bucket_notification" "upload_trigger" {
  bucket = aws_s3_bucket.uploads.id

  topic {
    topic_arn = aws_sns_topic.file_events.arn
    events    = ["s3:ObjectCreated:*"]
  }
}

resource "aws_sns_topic_subscription" "to_sqs" {
  topic_arn = aws_sns_topic.file_events.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.file_processing.arn
}

Lambda конфігурація для обробки файлів

resource "aws_lambda_function" "processor" {
  filename         = "processor.zip"
  function_name    = "file-processor"
  role             = aws_iam_role.processor.arn
  handler          = "handler.handler"
  runtime          = "python3.12"
  timeout          = 300    # 5 хвилин
  memory_size      = 1024   # 1GB для обробки зображень

  ephemeral_storage {
    size = 2048  # 2GB /tmp для тимчасових файлів
  }

  environment {
    variables = {
      OUTPUT_BUCKET = aws_s3_bucket.processed.bucket
    }
  }
}

Моніторинг та метрики

• Число обробленихфайлів на годину
• Середній час обробки за типом файлу
• Error rate + вміст DLQ
• Lambda duration distribution (outliers = проблемні файлі)

CloudWatch Dashboard з цими метриками + алерт при зростанні DLQ.

Графік впровадження

• Базовий S3 trigger + Lambda обробник — 1-2 дні
• Надійна схема (SNS + SQS + DLQ) — 1-2 дні
• Обробка конкретного типу файлів — 2-5 днів
• Моніторинг + тестування — 1-2 дні