Kafka и SQS: сравнение инструментов потоковой передачи

# Создание топика в Kafka
from kafka.admin import KafkaAdminClient, NewTopic
 
admin_client = KafkaAdminClient(bootstrap_servers='localhost:9092')
topic = NewTopic(name="payment-events", 
                 num_partitions=5,  # Можем распределить данные
                 replication_factor=3)  # И создать избыточность
admin_client.create_topics([topic])
 
# Создание очереди в SQS
import boto3
 
sqs = boto3.client('sqs')
response = sqs.create_queue(
    QueueName='payment-processing',
    Attributes={
        'DelaySeconds': '0',
        'MessageRetentionPeriod': '86400'  # 1 день хранения
    }
)

# Продюсер с указанием ключа партиционирования
from kafka import KafkaProducer
import json
 
producer = KafkaProducer(
    bootstrap_servers=['broker1:9092', 'broker2:9092'],
    value_serializer=lambda m: json.dumps(m).encode('utf-8')
)
 
# Сообщения с одинаковым ключом попадут в одну партицию
producer.send('user-events', 
              key=b'user_123',  # Ключ определяет партицию
              value={'action': 'login', 'timestamp': 1626954730})

# Создание FIFO-очереди в SQS
fifo_queue = sqs.create_queue(
    QueueName='user-events.fifo',  # Обратите внимание на суффикс .fifo
    Attributes={
        'FifoQueue': 'true',
        'ContentBasedDeduplication': 'true'
    }
)
 
# Отправка сообщения с группой ID для сохранения порядка
response = sqs.send_message(
    QueueUrl=fifo_queue['QueueUrl'],
    MessageBody=json.dumps({'action': 'login', 'timestamp': 1626954730}),
    MessageGroupId='user_123'  # Аналог ключа партиции в Kafka
)

# Настройка политики хранения для топика Kafka
kafka_client = KafkaAdminClient(bootstrap_servers='localhost:9092')
config = kafka_client.describe_configs(
config_resources=[ConfigResource(ConfigResourceType.TOPIC, 'payment-events')]
)
 
# Изменяем срок хранения на 7 дней (604800000 мс)
kafka_client.alter_configs([
ConfigResource(ConfigResourceType.TOPIC, 'payment-events', 
              configs={'retention.ms': '604800000'})
])

# Настройка продюсера Kafka для надежной доставки
producer = KafkaProducer(
    bootstrap_servers=['broker1:9092', 'broker2:9092'],
    acks='all',  # Ждем подтверждения от всех реплик
    retries=5,   # Количество повторных попыток
    # Другие параметры для повышения надежности
    max_in_flight_requests_per_connection=1,
    enable_idempotence=True  # Для предотвращения дубликатов
)

# Получение и обработка сообщений из SQS с настройкой visibility timeout
response = sqs.receive_message(
    QueueUrl=queue_url,
    MaxNumberOfMessages=10,
    VisibilityTimeout=60,  # Сообщение не будет видно другим потребителям 60 секунд
    WaitTimeSeconds=20
)
 
# Обработка и удаление сообщений
for message in response.get('Messages', []):
    try:
        # Обработка сообщения
        process_message(message)
        
        # Удаление сообщения из очереди после успешной обработки
        sqs.delete_message(
            QueueUrl=queue_url,
            ReceiptHandle=message['ReceiptHandle']
        )
    except Exception as e:
        # В случае ошибки сообщение автоматически вернется в очередь
        # после истечения visibility timeout
        print(f"Ошибка обработки: {e}")

# Отправка сообщений в FIFO-очередь SQS с сохранением порядка
for i in range(5):
    sqs.send_message(
        QueueUrl=fifo_queue_url,
        MessageBody=f"Заказ перешел в статус: {status[i]}",
        MessageGroupId='order_123',  # Все сообщения с одинаковым MessageGroupId 
                                     # обрабатываются строго последовательно
        MessageDeduplicationId=f'msg_{order_id}_{i}'  # Уникальный ID для дедупликации
    )

# Пример использования транзакций в Kafka для гарантии exactly-once
producer = KafkaProducer(
    bootstrap_servers=['localhost:9092'],
    transactional_id='unique-transaction-id',
    acks='all',
    enable_idempotence=True
)
 
# Начало транзакции
producer.begin_transaction()
try:
    # Отправка нескольких сообщений как атомарная операция
    producer.send('output-topic', key=b'key1', value=b'value1')
    producer.send('output-topic', key=b'key2', value=b'value2')
    
    # Фиксация транзакции
    producer.commit_transaction()
except Exception:
    # Откат транзакции в случае ошибки
    producer.abort_transaction()

# Пример гибридного подхода: получение сообщений из SQS и их загрузка в Kafka
[H2]для дальнейшей высокопроизводительной обработки[/H2]
 
def sqs_to_kafka_bridge():
    # Получение сообщений из SQS
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20
    )
    
    messages = response.get('Messages', [])
    if not messages:
        return
    
    # Отправка сообщений в Kafka
    for message in messages:
        producer.send(
            'local-processing-topic',
            value=message['Body'].encode('utf-8')
        )
        
        # Удаление обработанного сообщения из SQS
        sqs.delete_message(
            QueueUrl=queue_url,
            ReceiptHandle=message['ReceiptHandle']
        )
    
    # Фиксация отправки в Kafka
    producer.flush()

# Настройка продюсера Kafka для пакетной отправки
producer = KafkaProducer(
bootstrap_servers=['localhost:9092'],
batch_size=16384,  # Размер пакета в байтах
linger_ms=100,     # Время ожидания для формирования пакета
compression_type='snappy'  # Сжатие для экономии трафика
)

# Пакетная отправка сообщений в SQS
messages = [
{'Id': f'msg_{i}', 'MessageBody': f'Сообщение {i}'} for i in range(10)
]
 
response = sqs.send_message_batch(
QueueUrl=queue_url,
Entries=messages
)

try:
# Обработка сообщения
process_message(message)
# Фиксация успешной обработки
consumer.commit()
except DatabaseError as e:
# Отправка в dead letter queue для последующего анализа
producer.send('failed-processing', 
         key=message.key, 
         value=message.value,
         headers=[('error', str(e).encode())])
# Всё равно фиксируем, чтобы продолжить обработку
consumer.commit()

# Создание очереди с настройкой DLQ
queue = sqs.create_queue(
QueueName='my-processing-queue',
Attributes={
    'RedrivePolicy': json.dumps({
        'deadLetterTargetArn': dlq_arn,
        'maxReceiveCount': '5'  # После 5 неудачных попыток сообщение пойдет в DLQ
    })
}
)

# Интеграция SQS с Lambda через AWS CDK
from aws_cdk import (
core,
aws_sqs as sqs,
aws_lambda as lambda_,
aws_lambda_event_sources as lambda_events
)
 
class SqsLambdaStack(core.Stack):
def __init__(self, scope: core.Construct, id: str, **kwargs) -> None:
    super().__init__(scope, id, **kwargs)
    
    # Создаем очередь SQS
    queue = sqs.Queue(self, "MyQueue")
    
    # Создаем функцию Lambda
    function = lambda_.Function(self, "ProcessorFunction",
        runtime=lambda_.Runtime.PYTHON_3_9,
        handler="index.handler",
        code=lambda_.Code.from_asset("lambda")
    )
    
    # Подключаем SQS как источник событий для Lambda
    function.add_event_source(lambda_events.SqsEventSource(queue))

# AWS Lambda для обработки сообщений из SQS
def handler(event, context):
for record in event['Records']:
    message = json.loads(record['body'])
    
    # Обработка сообщения
    process_message(message)
    
    # Отправка результатов в другую очередь или сервис
    result = {'processed': True, 'original': message}
    sqs.send_message(
        QueueUrl=output_queue_url,
        MessageBody=json.dumps(result)
    )

# Создание CloudWatch аларма для SQS через boto3
cloudwatch = boto3.client('cloudwatch')
cloudwatch.put_metric_alarm(
  AlarmName='QueueHighDelayAlarm',
  ComparisonOperator='GreaterThanThreshold',
  EvaluationPeriods=1,
  MetricName='ApproximateAgeOfOldestMessage',
  Namespace='AWS/SQS',
  Period=300,
  Statistic='Maximum',
  Threshold=60.0,  # Тревога, если сообщения старше 60 секунд
  AlarmDescription='Alarm when message processing delay is too high',
  Dimensions=[
      {
          'Name': 'QueueName',
          'Value': queue_name
      },
  ]
)

# MirrorMaker 2.0 конфигурация
clusters = source, target
source.bootstrap.servers = source-kafka1:9092,source-kafka2:9092
target.bootstrap.servers = target-kafka1:9092,target-kafka2:9092
 
# Включаем авторепликацию топиков
source->target.enabled = true
source->target.topics = .*  # Реплицировать все топики
 
# Настройки репликации
replication.factor = 3
sync.topic.acls.enabled = false

# Примерный расчет годовой стоимости владения Kafka кластером
def kafka_tco_calculator(servers_count=5, server_cost_monthly=500, 
                        devops_annual_salary=80000,
                        monitoring_tools_monthly=200):
    # Инфраструктурные расходы
    infrastructure_cost = servers_count * server_cost_monthly * 12
    
    # Расходы на персонал (предполагаем, что DevOps тратит 30% времени на Kafka)
    personnel_cost = devops_annual_salary * 0.3
    
    # Инструменты мониторинга и управления
    tools_cost = monitoring_tools_monthly * 12
    
    # Общая стоимость владения
    total_cost = infrastructure_cost + personnel_cost + tools_cost
    
    return total_cost
 
annual_kafka_cost = kafka_tco_calculator()
print(f"Годовая стоимость владения Kafka: ${annual_kafka_cost:,}")
# Вывод: Годовая стоимость владения Kafka: $54,400

# Расчет месячной стоимости SQS
def sqs_cost_calculator(messages_per_second=100, operations_per_message=3,
                      days_in_month=30, cost_per_million=0.40,
                      free_tier_requests=1000000):
    # Общее количество запросов в месяц
    total_requests = messages_per_second * operations_per_message * 86400 * days_in_month
    
    # Вычитаем бесплатный тир
    billable_requests = max(0, total_requests - free_tier_requests)
    
    # Рассчитываем стоимость
    cost = (billable_requests / 1000000) * cost_per_million
    
    return cost, total_requests
 
monthly_cost, requests = sqs_cost_calculator()
print(f"Месячная стоимость SQS: ${monthly_cost:.2f} за {requests:,} запросов")
# Вывод: Месячная стоимость SQS: $311.90 за 777,600,000 запросов

# Сравнение TCO Kafka и SQS при масштабировании
def compare_scaling_tco(messages_per_second_range=[100, 1000, 10000, 100000]):
    results = []
    
    for mps in messages_per_second_range:
        # Kafka - предполагаем, что нам нужен 1 сервер на каждые 5000 сообщений/сек
        kafka_servers = max(3, int(mps / 5000) + 1)  # минимум 3 сервера
        kafka_monthly = kafka_servers * 500 + 2000  # стоимость серверов + фиксированные расходы
        
        # SQS - пропорционально объему
        sqs_cost, _ = sqs_cost_calculator(messages_per_second=mps)
        
        results.append({
            'messages_per_second': mps,
            'kafka_monthly_cost': kafka_monthly,
            'sqs_monthly_cost': sqs_cost
        })
    
    return results
 
scaling_comparison = compare_scaling_tco()
for item in scaling_comparison:
    print(f"При {item['messages_per_second']} сообщений/сек:")
    print(f"  Kafka: ${item['kafka_monthly_cost']:,.2f}/месяц")
    print(f"  SQS: ${item['sqs_monthly_cost']:,.2f}/месяц")

# Чтение исторических данных в Kafka
from kafka import KafkaConsumer
 
# Создаем потребителя, который начнет чтение с начала топика
consumer = KafkaConsumer(
    'user-activity',
    bootstrap_servers=['kafka1:9092', 'kafka2:9092'],
    auto_offset_reset='earliest',  # Важно! Начинаем с самого первого сообщения
    consumer_timeout_ms=10000
)
 
# Читаем все сообщения с начала времен
for message in consumer:
    process_historical_data(message.value)

# Обработка сообщений из SQS с автоматическим масштабированием
import boto3
import time
 
sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'
 
# Этот код может выполняться в Lambda или на EC2 с Auto Scaling
while True:
    # Получаем сообщения - при росте очереди можно увеличить число инстансов
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20
    )
    
    # Обрабатываем полученные сообщения
    if 'Messages' in response:
        for message in response['Messages']:
            process_message(message['Body'])
            
            # Удаляем обработанное сообщение
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )
    else:
        # Если сообщений нет, инстанс можно остановить для экономии
        time.sleep(30)

# Буферизация через SQS перед отправкой в Kafka
def buffer_to_kafka():
    # Получаем сообщения из SQS с высокой скоростью
    messages = receive_from_sqs(max_messages=100)
    
    # Группируем по ключам для правильного партиционирования в Kafka
    grouped_messages = {}
    for msg in messages:
        key = extract_key(msg)
        if key not in grouped_messages:
            grouped_messages[key] = []
        grouped_messages[key].append(msg)
    
    # Отправляем партиями в Kafka с сохранением порядка по ключу
    for key, msgs in grouped_messages.items():
        for msg in msgs:
            kafka_producer.send('high-volume-topic', key=key, value=msg)
    
    # Удаляем обработанные сообщения из SQS
    delete_from_sqs(messages)