AWS Kinesis के साथ इवेंट-ड्राइवन माइक्रोसर्विसेज बनाना

import boto3
import json
from datetime import datetime

kinesis = boto3.client('kinesis', region_name='us-east-1')

def publish_event(stream_name, event_type, payload):
    """Publish an event to Kinesis stream"""
    event = {
        'eventId': generate_unique_id(),
        'eventType': event_type,
        'timestamp': datetime.utcnow().isoformat(),
        'payload': payload
    }

    response = kinesis.put_record(
        StreamName=stream_name,
        Data=json.dumps(event),
        PartitionKey=payload.get('userId', 'default')
    )

    return response['SequenceNumber']

# Example: User registration event
publish_event(
    stream_name='user-events',
    event_type='USER_REGISTERED',
    payload={
        'userId': '12345',
        'email': 'user@example.com',
        'registrationDate': '2025-10-30'
    }
)

package main

import (
    "encoding/json"
    "github.com/aws/aws-sdk-go/aws"
    "github.com/aws/aws-sdk-go/service/kinesis"
)

type OrderCommand struct {
    CommandType string      `json:"commandType"`
    OrderID     string      `json:"orderId"`
    UserID      string      `json:"userId"`
    Items       []OrderItem `json:"items"`
}

func ProcessCommand(kinesisClient *kinesis.Kinesis, command OrderCommand) error {
    data, err := json.Marshal(command)
    if err != nil {
        return err
    }

    _, err = kinesisClient.PutRecord(&kinesis.PutRecordInput{
        StreamName:   aws.String("order-commands"),
        Data:         data,
        PartitionKey: aws.String(command.OrderID),
    })

    return err
}

// Lambda consumer for email notifications
exports.handler = async (event) => {
    for (const record of event.Records) {
        const payload = JSON.parse(
            Buffer.from(record.kinesis.data, 'base64').toString('utf-8')
        );

        if (payload.eventType === 'ORDER_PLACED') {
            await sendOrderConfirmationEmail(payload);
        }
    };
};

// Another Lambda for inventory updates
exports.inventoryHandler = async (event) => {
    for (const record of event.Records) {
        const payload = JSON.parse(
            Buffer.from(record.kinesis.data, 'base64').toString('utf-8')
        );

        if (payload.eventType === 'ORDER_PLACED') {
            await updateInventory(payload.items);
        }
    };
};

def calculate_shards(records_per_second, avg_record_size_kb):
    """Calculate required number of shards"""
    # Ingress capacity
    ingress_shards = max(
        records_per_second / 1000,
        (records_per_second * avg_record_size_kb) / 1024
    )

    return int(ingress_shards) + 1  # Add buffer

from botocore.exceptions import ClientError
import time

def put_record_with_retry(kinesis_client, stream_name, data, partition_key,
                          max_retries=3):
    """Put record with exponential backoff retry"""
    for attempt in range(max_retries):
        try:
            response = kinesis_client.put_record(
                StreamName=stream_name,
                Data=data,
                PartitionKey=partition_key
            )
            return response
        except ClientError as e:
            if e.response['Error']['Code'] == 'ProvisionedThroughputExceededException':
                if attempt < max_retries - 1:
                    time.sleep(2 ** attempt)  # Exponential backoff
                    continue
            raise

# Register a consumer with enhanced fan-out
aws kinesis register-stream-consumer \
    --stream-arn arn:aws:kinesis:us-east-1:123456789:stream/my-stream \
    --consumer-name my-consumer-app

import hashlib

def get_partition_key(user_id, shard_count=10):
    """Generate partition key with even distribution"""
    # Use consistent hashing for even distribution
    hash_value = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
    shard_id = hash_value % shard_count
    return f"{user_id}#{shard_id}"

import boto3
import json
from decimal import Decimal
from typing import Dict, List

class OrderProcessingService:
    def __init__(self, stream_name: str):
        self.kinesis = boto3.client('kinesis')
        self.stream_name = stream_name

    def create_order(self, user_id: str, items: List[Dict]) -> str:
        """Create order and publish events"""
        order_id = self.generate_order_id()

        # Publish order created event
        self.publish_event('ORDER_CREATED', {
            'orderId': order_id,
            'userId': user_id,
            'items': items,
            'status': 'PENDING',
            'total': self.calculate_total(items)
        }, partition_key=user_id)

        return order_id

    def publish_event(self, event_type: str, payload: Dict,
                     partition_key: str):
        """Publish event to Kinesis stream"""
        event = {
            'eventType': event_type,
            'timestamp': datetime.utcnow().isoformat(),
            'payload': payload
        }

        self.kinesis.put_record(
            StreamName=self.stream_name,
            Data=json.dumps(event, default=str),
            PartitionKey=partition_key
        )

class InventoryService:
    """Consumes order events and updates inventory"""

    def process_records(self, records):
        for record in records:
            data = json.loads(record['kinesis']['data'])

            if data['eventType'] == 'ORDER_CREATED':
                self.reserve_inventory(data['payload'])
            elif data['eventType'] == 'ORDER_CANCELLED':
                self.release_inventory(data['payload'])

    def reserve_inventory(self, order_data):
        # Update inventory database
        for item in order_data['items']:
            # Implementation here
            pass

def dual_write_pattern(legacy_db, kinesis_stream, data):
    """नए सिस्टम और घटना स्ट्रीम दोनों में लिखें"""
    try:
        # पहले नए सिस्टम में लिखें
        publish_to_kinesis(kinesis_stream, data)

        # फिर पुराने सिस्टम को अपडेट करें
        legacy_db.update(data)
    except Exception as e:
        # क्षतिपूर्ति तर्क लागू करें
        rollback_kinesis_event(kinesis_stream, data)
        raise

# ऑन-डिमांड मोड के साथ स्ट्रीम बनाएं
aws kinesis create-stream \
    --stream-name my-stream \
    --stream-mode-details StreamMode=ON_DEMAND

from aws_kinesis_agg.aggregator import RecordAggregator

def batch_put_records(kinesis_client, stream_name, records):
    """लागत कम करने के लिए रेकॉर्ड्स को एग्रीगेट करें"""
    aggregator = RecordAggregator()

    for record in records:
        aggregator.add_user_record(
            partition_key=record['partition_key'],
            data=record['data']
        )

    # एग्रीगेटेड रेकॉर्ड भेजें
    kinesis_client.put_record(
        StreamName=stream_name,
        Data=aggregator.serialize(),
        PartitionKey=records[0]['partition_key']
    )

# रिटेंशन को 7 दिनों तक सेट करें
aws kinesis increase-stream-retention-period \
    --stream-name my-stream \
    --retention-period-hours 168

# एन्क्रिप्टेड स्ट्रीम बनाएं
kinesis.create_stream(
    StreamName='secure-stream',
    ShardCount=1,
    StreamModeDetails={'StreamMode': 'PROVISIONED'}
)

# एन्क्रिप्शन सक्षम करें
kinesis.start_stream_encryption(
    StreamName='secure-stream',
    EncryptionType='KMS',
    KeyId='alias/aws/kinesis'
)

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "kinesis:PutRecord",
        "kinesis:PutRecords"
      ],
      "Resource": "arn:aws:kinesis:us-east-1:123456789:stream/orders"
    }
  ]
}

aws ec2 create-vpc-endpoint \
    --vpc-id vpc-12345678 \
    --service-name com.amazonaws.us-east-1.kinesis-streams \
    --route-table-ids rtb-12345678

from aws_xray_sdk.core import xray_recorder
from aws_xray_sdk.core import patch_all

patch_all()

@xray_recorder.capture('publish_event')
def publish_event_with_tracing(stream_name, event_data):
    xray_recorder.put_annotation('eventType', event_data['type'])
    xray_recorder.put_metadata('payload', event_data['payload'])

    kinesis.put_record(
        StreamName=stream_name,
        Data=json.dumps(event_data),
        PartitionKey=event_data['id']
    )

-- धीमी प्रोसेसिंग टाइम्स ढूंढें
fields @timestamp, eventType, processingTime
| filter processingTime > 1000
| sort @timestamp desc
| limit 20

-- एरर रेट ट्रैक करें
fields @timestamp, eventType
| filter error = true
| stats count() by eventType

class OrderSaga:
    def __init__(self, kinesis_client, stream_name):
        self.kinesis = kinesis_client
        self.stream_name = stream_name
        self.saga_id = str(uuid.uuid4())

    def execute(self, order_data):
        """क्षतिपूर्ति तर्क के साथ सागा निष्पादित करें"""
        try:
            # चरण 1: इन्वेंटरी रिजर्व करें
            self.publish_command('RESERVE_INVENTORY', order_data)

            # चरण 2: भुगतान प्रोसेस करें
            self.publish_command('PROCESS_PAYMENT', order_data)

            # चरण 3: ऑर्डर शिप करें
            self.publish_command('SHIP_ORDER', order_data)

        except SagaException as e:
            # विपरीत क्रम में क्षतिपूर्ति करें
            self.compensate(e.failed_step)

    def compensate(self, failed_step):
        """क्षतिपूर्ति ट्रांजैक्शन्स निष्पादित करें"""
        compensation_steps = {
            'PROCESS_PAYMENT': ['RELEASE_INVENTORY'],
            'SHIP_ORDER': ['REFUND_PAYMENT', 'RELEASE_INVENTORY']
        }

        for step in compensation_steps.get(failed_step, []):
            self.publish_command(step, {'sagaId': self.saga_id})

# Kinesis के साथ LocalStack शुरू करें
docker run -d \
  -p 4566:4566 \
  -e SERVICES=kinesis \
  localstack/localstack

# परीक्षण स्ट्रीम बनाएं
aws --endpoint-url=http://localhost:4566 kinesis create-stream \
    --stream-name test-stream \
    --shard-count 1

import pytest
from moto import mock_kinesis

@mock_kinesis
def test_event_publishing():
    """मॉक्ड Kinesis के साथ घटना प्रकाशन परीक्षण करें"""
    kinesis = boto3.client('kinesis', region_name='us-east-1')
    kinesis.create_stream(StreamName='test-stream', ShardCount=1)

    service = OrderProcessingService('test-stream')
    order_id = service.create_order('user123', [
        {'productId': 'prod1', 'quantity': 2}
    ])

    assert order_id is not None

def optimize_batch_processing(records, batch_size=500):
    """अनुकूलित बैच में रेकॉर्ड्स प्रोसेस करें"""
    for i in range(0, len(records), batch_size):
        batch = records[i:i + batch_size]
        process_batch(batch)

from botocore.config import Config

config = Config(
    max_pool_connections=50,
    retries={'max_attempts': 3, 'mode': 'adaptive'}
)

kinesis = boto3.client('kinesis', config=config)