इवेंट सोर्स मैपिंग और स्ट्रीम प्रोसेसिंग

AWS Lambda के साथ Serverless Applications

Claudio Canales

Senior DevOps Engineer

इवेंट सोर्स मैपिंग कहाँ फिट होती हैं

क्व्यू और स्ट्रीम अक्सर बैचों में पोल किए जाते हैं।
इवेंट सोर्स मैपिंग पोलिंग और बैचिंग को नियंत्रित करती हैं।

इवेंट सोर्स मैपिंग की स्थिति

इवेंट सोर्स मैपिंग क्या है?

एक संसाधन जो source को आपकी फंक्शन से जोड़ता है।
Lambda रिकॉर्ड्स के लिए source को polls करती है।
Lambda आपके हैंडलर को batch के साथ कॉल करती है।

इवेंट सोर्स मैपिंग फ्लो

Push बनाम Poll (क्यों महत्वपूर्ण)

Push मॉडल

Push सोर्सेज तुरंत इवेंट भेजते हैं।
उदाहरण: Amazon S3 पर अपलोड।

Poll मॉडल

Poll सोर्सेज को Lambda जाँचती है।
Lambda बैच बनाती है और आपका हैंडलर कॉल करती है।
इससे लेटेंसी और रिट्राई व्यवहार बदलता है।

Push बनाम Poll मॉडल

क्व्यू बनाम स्ट्रीम

Queue (Amazon SQS)

एक managed मैसेज क्व्यू, जैसे इनबॉक्स।

Stream (DynamoDB Streams)

DynamoDB टेबल का change log।
दोनों Records के रूप में आते हैं, पर अर्थ अलग होता है।

क्व्यू बनाम स्ट्रीम तुलना

बैच इवेंट का रूप (सरल)

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

ज्यादातर मैपिंग इवेंट Records से शुरू होते हैं।
हर रिकॉर्ड में messageId और body string होती है।
body पार्स करके अपना payload बनाइए।

वॉकथ्रू: हर रिकॉर्ड प्रोसेस करें

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 से पार्स करके dict पाएँ।
ज़रूरी फील्ड निकालें; हमेशा पहले वैलिडेट करें।

बैचिंग: बैच साइज और बैच विंडो

Batch size: प्रति invocation कितने रिकॉर्ड।
Batch window: बैच भरने के लिए Lambda कितनी देर इंतज़ार करती है।
बड़े बैच थ्रूपुट बढ़ाते हैं पर प्रति-रन काम भी बढ़ता है।
SQS में बैकलॉग के लिए ApproximateAgeOfOldestMessage मॉनिटर करें।

बैच साइज ट्रेडऑफ डायग्राम

बैच साइज के ट्रेड-ऑफ्स

बड़े बैच कुशल होते हैं पर लेटेंसी बढ़ा सकते हैं।
छोटे बैच time-to-first-processing घटाते हैं पर invocations बढ़ाते हैं।
अपनी वर्कलोड के अनुसार चुनें।

बैच साइज बनाम लेटेंसी चार्ट

Concurrency से स्केलिंग

कई बैच एक साथ प्रोसेस होते हैं।
ज़्यादा concurrency थ्रूपुट बढ़ाती है।
लेकिन डाउनस्ट्रीम लोड भी बढ़ता है।

समानांतर बैच प्रोसेसिंग

डाउनस्ट्रीम सिस्टम की सुरक्षा

बहुत अधिक concurrency डेटाबेस या API को ओवरव्हेल्म कर सकती है।
लोड नियंत्रित करने के लिए लिमिट्स लगाएँ।
छोटे, सुरक्षित units of work को प्राथमिकता दें।

डाउनस्ट्रीम लोड सुरक्षा

आंशिक बैच फेल्योर

एक बैच में अच्छे और खराब दोनों रिकॉर्ड हो सकते हैं।
आंशिक फेल्योर न हो तो एक खराब रिकॉर्ड पूरा बैच रिट्राई कराता है।
आंशिक फेल्योर हैंडलिंग से केवल फेल आइटम्स रिट्राई होते हैं।

आंशिक बैच फेल्योर फ्लो

वॉकथ्रू: आंशिक फेल्योर रिपोर्टिंग (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 के साथ Serverless Applications