Маппинги источников событий и потоковая обработка

Бессерверные приложения с AWS Lambda

Claudio Canales

Senior DevOps Engineer

Место маппингов источников событий

Очереди и потоки часто опрашиваются пакетами.
Маппинги источников событий управляют опросом и пакетной обработкой.

Позиция маппинга источника событий

Что такое маппинг источника событий?

Ресурс, связывающий источник с вашей функцией.
Lambda опрашивает источник на наличие записей.
Lambda вызывает обработчик с пакетом.

Схема потока маппинга источника событий

Push vs poll: почему это важно

Модель push

Push-источники отправляют события немедленно.
Пример: загрузка файла в Amazon S3.

Модель poll

Poll-источники проверяются Lambda.
Lambda формирует пакет и вызывает обработчик.
Это влияет на задержку и поведение при повторных попытках.

Модели push и poll

Очередь vs поток

Очередь (Amazon SQS)

Управляемая очередь сообщений, как папка «Входящие».

Поток (DynamoDB Streams)

Журнал изменений таблицы DynamoDB.
Оба доставляются как Records, но их смысл различается.

Сравнение очереди и потока

Структура пакетного события (упрощённо)

{
  "Records": [{
    "messageId": "abc-123",
    "body": "{\"order_id\": \"A-42\"}"
  }]
}

Большинство событий маппинга начинаются с Records.
Каждая запись содержит messageId и строку body.
Разберите body в собственный объект данных.

Разбор: обработка каждой записи

def lambda_handler(event, context):
    records = event.get("Records", [])
    for record in records:
        body = record.get("body", "")
        print("BODY:", body)
    return {"statusCode": 200}

Читайте Records с пустым списком по умолчанию.
Перебирайте записи и безопасно извлекайте body.
Логируйте нужные данные и возвращайте результат.

Разбор: парсинг тела JSON

import json

def lambda_handler(event, context):
    record = event.get("Records", [])[0]
    payload = json.loads(record.get("body", "{}"))
    order_id = payload.get("order_id")
    print("ORDER_ID:", order_id)
    return {"statusCode": 200}

Импортируйте json и читайте body с безопасным значением по умолчанию.
Разберите строку с помощью json.loads, чтобы получить словарь.
Извлекайте нужные поля, предварительно проверяя их наличие.

Пакетная обработка: размер и окно пакета

Размер пакета: количество записей за одно выполнение.
Окно пакета: время ожидания Lambda для заполнения пакета.
Крупные пакеты увеличивают пропускную способность, но добавляют работы за запуск.
Для SQS отслеживайте ApproximateAgeOfOldestMessage для контроля накопления.

Диаграмма компромисса размера пакета

Компромиссы размера пакета

Крупные пакеты эффективны, но могут увеличивать задержку.
Малые пакеты сокращают время до первой обработки, но увеличивают число вызовов.
Выбирайте параметры исходя из характера нагрузки.

График: размер пакета vs задержка

Масштабирование с помощью параллелизма

Несколько пакетов обрабатываются одновременно.
Высокий уровень параллелизма повышает пропускную способность.
При этом возрастает нагрузка на зависимые системы.

Параллельная пакетная обработка

Защита зависимых систем

Избыточный параллелизм может перегрузить базу данных или API.
Используйте ограничения для контроля нагрузки.
Предпочитайте небольшие, безопасные единицы работы.

Защита зависимых систем от перегрузки

Частичные сбои пакета

Пакет может содержать как корректные, так и некорректные записи.
Без обработки частичных сбоев одна плохая запись вызывает повтор всего пакета.
Обработка частичных сбоев позволяет повторять только упавшие элементы.

Схема частичного сбоя пакета

Разбор: отчёт о частичных сбоях (SQS)

def lambda_handler(event, context):
    failures = []
    for record in event.get("Records", []):
        try:
            process(record)
        except Exception:
            failures.append({"itemIdentifier": record["messageId"]})
    return {"batchItemFailures": failures}

Собирайте messageId упавших записей в batchItemFailures.
Lambda повторит только эти элементы.
Успешно обработанные записи повторно не отправляются.

Основные выводы

Маппинги источников событий опрашивают очереди и потоки.
Они доставляют Records пакетами в обработчик.
Размер и окно пакета определяют баланс между задержкой и пропускной способностью.
Частичные сбои позволяют не повторять успешно обработанные записи.

Итоговая схема маппинга

Давайте потренируемся!

Бессерверные приложения с AWS Lambda