अमेज़ॅन किनेसिस और सर्वर रहित सरलता के साथ एविएसेल्स एपीआई एकीकरण

हे हबर!

क्या आपको हवाई जहाज उड़ाना पसंद है? मुझे यह पसंद है, लेकिन आत्म-अलगाव के दौरान मुझे एक प्रसिद्ध संसाधन - एविएसेल्स से हवाई टिकटों पर डेटा का विश्लेषण करने का भी शौक हो गया।

आज हम अमेज़ॅन किनेसिस के काम का विश्लेषण करेंगे, वास्तविक समय विश्लेषण के साथ एक स्ट्रीमिंग सिस्टम का निर्माण करेंगे, अमेज़ॅन डायनेमोडीबी नोएसक्यूएल डेटाबेस को मुख्य डेटा भंडारण के रूप में स्थापित करेंगे, और दिलचस्प टिकटों के लिए एसएमएस सूचनाएं सेट करेंगे।

सभी विवरण कटौती के अधीन हैं! जाना!

परिचय

उदाहरण के लिए, हमें पहुंच की आवश्यकता है एविएसेल्स एपीआई. इस तक पहुंच नि:शुल्क और बिना किसी प्रतिबंध के प्रदान की जाती है; आपको डेटा तक पहुंचने के लिए अपना एपीआई टोकन प्राप्त करने के लिए बस "डेवलपर्स" अनुभाग में पंजीकरण करना होगा।

इस लेख का मुख्य उद्देश्य AWS में सूचना स्ट्रीमिंग के उपयोग की सामान्य समझ देना है; हम इस बात को ध्यान में रखते हैं कि उपयोग की गई एपीआई द्वारा लौटाया गया डेटा सख्ती से अद्यतित नहीं है और कैश से प्रसारित होता है, जो है पिछले 48 घंटों में Aviasales.ru और Jetradar.com साइटों के उपयोगकर्ताओं द्वारा की गई खोजों के आधार पर बनाया गया है।

एपीआई के माध्यम से प्राप्त उत्पादन मशीन पर स्थापित काइनेसिस-एजेंट स्वचालित रूप से काइनेसिस डेटा एनालिटिक्स के माध्यम से डेटा को वांछित स्ट्रीम में पार्स और संचारित करेगा। इस स्ट्रीम का कच्चा संस्करण सीधे स्टोर पर लिखा जाएगा। DynamoDB में तैनात कच्चा डेटा भंडारण AWS क्विक साइट जैसे BI टूल के माध्यम से गहन टिकट विश्लेषण की अनुमति देगा।

हम संपूर्ण बुनियादी ढांचे को तैनात करने के लिए दो विकल्पों पर विचार करेंगे:

• मैनुअल - AWS प्रबंधन कंसोल के माध्यम से;
• टेराफॉर्म कोड का इंफ्रास्ट्रक्चर आलसी ऑटोमेटर्स के लिए है;

विकसित प्रणाली की वास्तुकला

प्रयुक्त घटक:

• एविएसेल्स एपीआई - इस एपीआई द्वारा लौटाए गए डेटा का उपयोग बाद के सभी कार्यों के लिए किया जाएगा;
• EC2 निर्माता उदाहरण - क्लाउड में एक नियमित वर्चुअल मशीन जिस पर इनपुट डेटा स्ट्रीम उत्पन्न होगी:
  • काइनेसिस एजेंट मशीन पर स्थानीय रूप से इंस्टॉल किया गया एक जावा एप्लिकेशन है जो किइनेसिस (काइनेसिस डेटा स्ट्रीम या किनेसिस फायरहोज) को डेटा एकत्र करने और भेजने का एक आसान तरीका प्रदान करता है। एजेंट लगातार निर्दिष्ट निर्देशिकाओं में फ़ाइलों के एक सेट की निगरानी करता है और किनेसिस को नया डेटा भेजता है;
  • एपीआई कॉलर स्क्रिप्ट - एक पायथन स्क्रिप्ट जो एपीआई के लिए अनुरोध करती है और प्रतिक्रिया को एक फ़ोल्डर में डालती है जिसकी निगरानी किनेसिस एजेंट द्वारा की जाती है;
• किनेसिस डेटा स्ट्रीम - व्यापक स्केलिंग क्षमताओं के साथ वास्तविक समय डेटा स्ट्रीमिंग सेवा;
• किनेसिस एनालिटिक्स एक सर्वर रहित सेवा है जो वास्तविक समय में स्ट्रीमिंग डेटा के विश्लेषण को सरल बनाती है। अमेज़ॅन किनेसिस डेटा एनालिटिक्स एप्लिकेशन संसाधनों को कॉन्फ़िगर करता है और आने वाले डेटा की किसी भी मात्रा को संभालने के लिए स्वचालित रूप से स्केल करता है;
• AWS लाम्बा - एक सेवा जो आपको सर्वर का बैकअप या सेटअप किए बिना कोड चलाने की अनुमति देती है। प्रत्येक कॉल के लिए सभी कंप्यूटिंग शक्ति स्वचालित रूप से मापी जाती है;
• अमेज़ॅन डायनेमोडीबी - कुंजी-मूल्य जोड़े और दस्तावेज़ों का एक डेटाबेस जो किसी भी पैमाने पर चलते समय 10 मिलीसेकंड से कम की विलंबता प्रदान करता है। DynamoDB का उपयोग करते समय, किसी सर्वर का प्रावधान, पैच या प्रबंधन करने की कोई आवश्यकता नहीं है। DynamoDB उपलब्ध संसाधनों की मात्रा को समायोजित करने और उच्च प्रदर्शन बनाए रखने के लिए स्वचालित रूप से तालिकाओं को मापता है। किसी सिस्टम प्रशासन की आवश्यकता नहीं है;
• अमेज़ॅन एसएनएस - प्रकाशक-ग्राहक (पब/उप) मॉडल का उपयोग करके संदेश भेजने के लिए एक पूरी तरह से प्रबंधित सेवा, जिसके साथ आप माइक्रोसर्विसेज, वितरित सिस्टम और सर्वर रहित अनुप्रयोगों को अलग कर सकते हैं। एसएनएस का उपयोग मोबाइल पुश नोटिफिकेशन, एसएमएस संदेश और ईमेल के माध्यम से अंतिम उपयोगकर्ताओं को जानकारी भेजने के लिए किया जा सकता है।

प्रारंभिक प्रशिक्षण

डेटा प्रवाह का अनुकरण करने के लिए, मैंने एविएसेल्स एपीआई द्वारा लौटाई गई एयरलाइन टिकट जानकारी का उपयोग करने का निर्णय लिया। में प्रलेखन विभिन्न तरीकों की एक विस्तृत सूची, आइए उनमें से एक को लें - "मासिक मूल्य कैलेंडर", जो महीने के प्रत्येक दिन के लिए कीमतों को स्थानान्तरण की संख्या के आधार पर समूहीकृत करता है। यदि आप अनुरोध में खोज माह निर्दिष्ट नहीं करते हैं, तो वर्तमान माह के बाद वाले माह की जानकारी लौटा दी जाएगी।

तो, आइए पंजीकरण करें और अपना टोकन प्राप्त करें।

एक उदाहरण अनुरोध नीचे है:

http://api.travelpayouts.com/v2/prices/month-matrix?currency=rub&origin=LED&destination=HKT&show_to_affiliates=true&token=TOKEN_API

अनुरोध में टोकन निर्दिष्ट करके एपीआई से डेटा प्राप्त करने की उपरोक्त विधि काम करेगी, लेकिन मैं हेडर के माध्यम से एक्सेस टोकन पास करना पसंद करता हूं, इसलिए हम इस विधि का उपयोग api_caller.py स्क्रिप्ट में करेंगे।

उत्तर उदाहरण:

{{
   "success":true,
   "data":[{
      "show_to_affiliates":true,
      "trip_class":0,
      "origin":"LED",
      "destination":"HKT",
      "depart_date":"2015-10-01",
      "return_date":"",
      "number_of_changes":1,
      "value":29127,
      "found_at":"2015-09-24T00:06:12+04:00",
      "distance":8015,
      "actual":true
   }]
}

उपरोक्त उदाहरण एपीआई प्रतिक्रिया सेंट पीटर्सबर्ग से फुक तक का टिकट दिखाती है... ओह, क्या सपना है...
चूँकि मैं कज़ान से हूँ, और फुकेत अब "केवल एक सपना" है, आइए सेंट पीटर्सबर्ग से कज़ान तक के टिकट देखें।

यह मानता है कि आपके पास पहले से ही एक AWS खाता है। मैं तुरंत इस तथ्य पर विशेष ध्यान आकर्षित करना चाहूंगा कि काइनेसिस और एसएमएस के माध्यम से सूचनाएं भेजना वार्षिक में शामिल नहीं है निःशुल्क टियर (मुफ़्त उपयोग). लेकिन इसके बावजूद, कुछ डॉलर को ध्यान में रखते हुए, प्रस्तावित प्रणाली का निर्माण करना और उसके साथ खेलना काफी संभव है। और, निःसंदेह, जब सभी संसाधनों की आवश्यकता न रह जाए तो उन्हें हटाना न भूलें।

सौभाग्य से, यदि हम अपनी मासिक निःशुल्क सीमा को पूरा करते हैं तो DynamoDb और Lambda फ़ंक्शन हमारे लिए निःशुल्क होंगे। उदाहरण के लिए, DynamoDB के लिए: 25 जीबी स्टोरेज, 25 WCU/RCU और 100 मिलियन क्वेरीज़। और प्रति माह दस लाख लैम्ब्डा फ़ंक्शन कॉल।

मैनुअल सिस्टम परिनियोजन

किनेसिस डेटा स्ट्रीम की स्थापना

आइए किनेसिस डेटा स्ट्रीम सेवा पर जाएं और दो नई स्ट्रीम बनाएं, प्रत्येक के लिए एक शार्ड।

शार्ड क्या है?
शार्ड अमेज़ॅन किनेसिस स्ट्रीम की मूल डेटा ट्रांसफर इकाई है। एक खंड 1 एमबी/सेकेंड की गति से इनपुट डेटा ट्रांसफर और 2 एमबी/सेकेंड की गति से आउटपुट डेटा ट्रांसफर प्रदान करता है। एक खंड प्रति सेकंड 1000 PUT प्रविष्टियों का समर्थन करता है। डेटा स्ट्रीम बनाते समय, आपको सेगमेंट की आवश्यक संख्या निर्दिष्ट करने की आवश्यकता होती है। उदाहरण के लिए, आप दो खंडों के साथ एक डेटा स्ट्रीम बना सकते हैं। यह डेटा स्ट्रीम 2 एमबी/एस पर इनपुट डेटा ट्रांसफर और 4 एमबी/एस पर आउटपुट डेटा ट्रांसफर प्रदान करेगा, प्रति सेकंड 2000 पुट रिकॉर्ड तक का समर्थन करेगा।

आपकी स्ट्रीम में जितने अधिक टुकड़े होंगे, उसका थ्रूपुट उतना ही अधिक होगा। सिद्धांत रूप में, इस तरह से प्रवाह को बढ़ाया जाता है - टुकड़ों को जोड़कर। लेकिन आपके पास जितने अधिक टुकड़े होंगे, कीमत उतनी ही अधिक होगी। प्रत्येक शार्ड की लागत 1,5 सेंट प्रति घंटा और प्रत्येक मिलियन PUT पेलोड इकाइयों के लिए अतिरिक्त 1.4 सेंट है।

आइए नाम के साथ एक नई स्ट्रीम बनाएं एयरलाइन टिकट, 1 टुकड़ा उसके लिए काफी होगा:

अब नाम के साथ एक और थ्रेड बनाते हैं विशेष धारा:

निर्माता सेटअप

किसी कार्य का विश्लेषण करने के लिए, डेटा निर्माता के रूप में नियमित EC2 उदाहरण का उपयोग करना पर्याप्त है। इसके लिए एक शक्तिशाली, महंगी वर्चुअल मशीन होना जरूरी नहीं है; एक स्पॉट t2.micro ठीक काम करेगा।

महत्वपूर्ण नोट: उदाहरण के लिए, आपको छवि का उपयोग करना चाहिए - अमेज़ॅन लिनक्स एएमआई 2018.03.0, इसमें किनेसिस एजेंट को तुरंत लॉन्च करने के लिए कम सेटिंग्स हैं।

EC2 सेवा पर जाएं, एक नई वर्चुअल मशीन बनाएं, t2.micro प्रकार के साथ वांछित AMI का चयन करें, जो फ्री टियर में शामिल है:

नव निर्मित वर्चुअल मशीन को किनेसिस सेवा के साथ बातचीत करने में सक्षम होने के लिए, उसे ऐसा करने का अधिकार दिया जाना चाहिए। ऐसा करने का सबसे अच्छा तरीका एक IAM भूमिका निर्दिष्ट करना है। इसलिए, चरण 3: इंस्टेंस विवरण कॉन्फ़िगर करें स्क्रीन पर, आपको चयन करना चाहिए नई IAM भूमिका बनाएँ:

EC2 के लिए IAM भूमिका बनाना

खुलने वाली विंडो में, चुनें कि हम EC2 के लिए एक नई भूमिका बना रहे हैं और अनुमतियाँ अनुभाग पर जाएँ:

प्रशिक्षण उदाहरण का उपयोग करते हुए, हमें संसाधन अधिकारों के विस्तृत कॉन्फ़िगरेशन की सभी जटिलताओं में जाने की ज़रूरत नहीं है, इसलिए हम अमेज़ॅन द्वारा पूर्व-कॉन्फ़िगर की गई नीतियों का चयन करेंगे: AmazonKineticFullAccess और CloudWatchFullAccess।

आइए इस भूमिका को कुछ सार्थक नाम दें, उदाहरण के लिए: EC2-KinesisStreams-FullAccess। परिणाम वैसा ही होना चाहिए जैसा नीचे चित्र में दिखाया गया है:

इस नई भूमिका को बनाने के बाद, इसे बनाई गई वर्चुअल मशीन इंस्टेंस से जोड़ना न भूलें:

हम इस स्क्रीन पर और कुछ नहीं बदलते हैं और अगली विंडो पर चले जाते हैं।

हार्ड ड्राइव सेटिंग्स को डिफ़ॉल्ट के रूप में छोड़ा जा सकता है, साथ ही टैग भी (हालांकि टैग का उपयोग करना अच्छा अभ्यास है, कम से कम उदाहरण को एक नाम दें और पर्यावरण को इंगित करें)।

अब हम चरण 6 पर हैं: सुरक्षा समूह कॉन्फ़िगर करें टैब, जहां आपको एक नया बनाने या अपने मौजूदा सुरक्षा समूह को निर्दिष्ट करने की आवश्यकता है, जो आपको उदाहरण के लिए एसएसएच (पोर्ट 22) के माध्यम से कनेक्ट करने की अनुमति देता है। वहां स्रोत -> मेरा आईपी चुनें और आप इंस्टेंस लॉन्च कर सकते हैं।

जैसे ही यह चालू स्थिति में आ जाता है, आप ssh के माध्यम से इससे जुड़ने का प्रयास कर सकते हैं।

मशीन से सफलतापूर्वक कनेक्ट होने के बाद, काइनेसिस एजेंट के साथ काम करने में सक्षम होने के लिए, आपको टर्मिनल में निम्नलिखित कमांड दर्ज करना होगा:

sudo yum -y update
sudo yum install -y python36 python36-pip
sudo /usr/bin/pip-3.6 install --upgrade pip
sudo yum install -y aws-kinesis-agent

आइए एपीआई प्रतिक्रियाओं को सहेजने के लिए एक फ़ोल्डर बनाएं:

sudo mkdir /var/log/airline_tickets

एजेंट शुरू करने से पहले, आपको इसकी कॉन्फ़िगरेशन कॉन्फ़िगर करने की आवश्यकता है:

sudo vim /etc/aws-kinesis/agent.json

एजेंट.जेसन फ़ाइल की सामग्री इस तरह दिखनी चाहिए:

{
  "cloudwatch.emitMetrics": true,
  "kinesis.endpoint": "",
  "firehose.endpoint": "",

  "flows": [
    {
      "filePattern": "/var/log/airline_tickets/*log",
      "kinesisStream": "airline_tickets",
      "partitionKeyOption": "RANDOM",
      "dataProcessingOptions": [
         {
            "optionName": "CSVTOJSON",
            "customFieldNames": ["cost","trip_class","show_to_affiliates",
                "return_date","origin","number_of_changes","gate","found_at",
                "duration","distance","destination","depart_date","actual","record_id"]
         }
      ]
    }
  ]
}

जैसा कि कॉन्फ़िगरेशन फ़ाइल से देखा जा सकता है, एजेंट /var/log/airline_tickets/ निर्देशिका में .log एक्सटेंशन वाली फ़ाइलों की निगरानी करेगा, उन्हें पार्स करेगा और उन्हें एयरलाइन_टिकट स्ट्रीम में स्थानांतरित करेगा।

हम सेवा को पुनः आरंभ करते हैं और सुनिश्चित करते हैं कि यह चालू है:

sudo service aws-kinesis-agent restart

अब आइए पायथन स्क्रिप्ट डाउनलोड करें जो एपीआई से डेटा का अनुरोध करेगी:

REPO_PATH=https://raw.githubusercontent.com/igorgorbenko/aviasales_kinesis/master/producer

wget $REPO_PATH/api_caller.py -P /home/ec2-user/
wget $REPO_PATH/requirements.txt -P /home/ec2-user/
sudo chmod a+x /home/ec2-user/api_caller.py
sudo /usr/local/bin/pip3 install -r /home/ec2-user/requirements.txt

Api_caller.py स्क्रिप्ट Aviasales से डेटा का अनुरोध करती है और प्राप्त प्रतिक्रिया को उस निर्देशिका में सहेजती है जिसे Kinetic एजेंट स्कैन करता है। इस स्क्रिप्ट का कार्यान्वयन काफी मानक है, एक टिकटएपी क्लास है, यह आपको एसिंक्रोनस रूप से एपीआई खींचने की अनुमति देता है। हम इस वर्ग के लिए टोकन और अनुरोध पैरामीटर के साथ एक हेडर पास करते हैं:

class TicketsApi:
    """Api caller class."""

    def __init__(self, headers):
        """Init method."""
        self.base_url = BASE_URL
        self.headers = headers

    async def get_data(self, data):
        """Get the data from API query."""
        response_json = {}
        async with ClientSession(headers=self.headers) as session:
            try:
                response = await session.get(self.base_url, data=data)
                response.raise_for_status()
                LOGGER.info('Response status %s: %s',
                            self.base_url, response.status)
                response_json = await response.json()
            except HTTPError as http_err:
                LOGGER.error('Oops! HTTP error occurred: %s', str(http_err))
            except Exception as err:
                LOGGER.error('Oops! An error ocurred: %s', str(err))
            return response_json


def prepare_request(api_token):
    """Return the headers and query fot the API request."""
    headers = {'X-Access-Token': api_token,
               'Accept-Encoding': 'gzip'}

    data = FormData()
    data.add_field('currency', CURRENCY)
    data.add_field('origin', ORIGIN)
    data.add_field('destination', DESTINATION)
    data.add_field('show_to_affiliates', SHOW_TO_AFFILIATES)
    data.add_field('trip_duration', TRIP_DURATION)
    return headers, data


async def main():
    """Get run the code."""
    if len(sys.argv) != 2:
        print('Usage: api_caller.py <your_api_token>')
        sys.exit(1)
        return
    api_token = sys.argv[1]
    headers, data = prepare_request(api_token)

    api = TicketsApi(headers)
    response = await api.get_data(data)
    if response.get('success', None):
        LOGGER.info('API has returned %s items', len(response['data']))
        try:
            count_rows = log_maker(response)
            LOGGER.info('%s rows have been saved into %s',
                        count_rows,
                        TARGET_FILE)
        except Exception as e:
            LOGGER.error('Oops! Request result was not saved to file. %s',
                         str(e))
    else:
        LOGGER.error('Oops! API request was unsuccessful %s!', response)

एजेंट की सही सेटिंग्स और कार्यक्षमता का परीक्षण करने के लिए, आइए api_caller.py स्क्रिप्ट का परीक्षण करें:

sudo ./api_caller.py TOKEN

और हम एजेंट लॉग में और एयरलाइन_टिकट डेटा स्ट्रीम में मॉनिटरिंग टैब पर काम के परिणाम को देखते हैं:

tail -f /var/log/aws-kinesis-agent/aws-kinesis-agent.log

जैसा कि आप देख सकते हैं, सब कुछ काम करता है और काइनेसिस एजेंट सफलतापूर्वक स्ट्रीम में डेटा भेजता है। अब उपभोक्ता को कॉन्फ़िगर करते हैं।

किनेसिस डेटा एनालिटिक्स की स्थापना

आइए पूरे सिस्टम के केंद्रीय घटक की ओर बढ़ते हैं - काइनेसिस डेटा एनालिटिक्स में एक नया एप्लिकेशन बनाएं जिसका नाम किनेसिस_एनालिटिक्स_एयरलाइन्स_एप है:

किनेसिस डेटा एनालिटिक्स आपको SQL भाषा का उपयोग करके किनेसिस स्ट्रीम से वास्तविक समय डेटा विश्लेषण करने की अनुमति देता है। यह पूरी तरह से ऑटोस्केलिंग सेवा है (किनेसिस स्ट्रीम के विपरीत) जो:

1. आपको स्रोत डेटा के अनुरोधों के आधार पर नई स्ट्रीम (आउटपुट स्ट्रीम) बनाने की अनुमति देता है;
2. एप्लिकेशन चलने के दौरान होने वाली त्रुटियों के साथ एक स्ट्रीम प्रदान करता है (त्रुटि स्ट्रीम);
3. स्वचालित रूप से इनपुट डेटा योजना निर्धारित कर सकता है (यदि आवश्यक हो तो इसे मैन्युअल रूप से फिर से परिभाषित किया जा सकता है)।

यह कोई सस्ती सेवा नहीं है - काम के प्रति घंटे 0.11 अमेरिकी डॉलर, इसलिए आपको इसका उपयोग सावधानी से करना चाहिए और काम पूरा होने पर इसे हटा देना चाहिए।

आइए एप्लिकेशन को डेटा स्रोत से कनेक्ट करें:

उस स्ट्रीम का चयन करें जिससे हम कनेक्ट होने जा रहे हैं (एयरलाइन_टिकट):

इसके बाद, आपको एक नई IAM भूमिका संलग्न करने की आवश्यकता है ताकि एप्लिकेशन स्ट्रीम से पढ़ सके और स्ट्रीम पर लिख सके। ऐसा करने के लिए, एक्सेस अनुमति ब्लॉक में कुछ भी न बदलना पर्याप्त है:

अब स्ट्रीम में डेटा स्कीमा की खोज का अनुरोध करें; ऐसा करने के लिए, "डिस्कवर स्कीमा" बटन पर क्लिक करें। परिणामस्वरूप, IAM भूमिका अपडेट की जाएगी (एक नई भूमिका बनाई जाएगी) और स्कीमा डिटेक्शन उस डेटा से लॉन्च किया जाएगा जो पहले ही स्ट्रीम में आ चुका है:

अब आपको SQL एडिटर पर जाना होगा। जब आप इस बटन पर क्लिक करेंगे, तो एक विंडो दिखाई देगी जो आपसे एप्लिकेशन लॉन्च करने के लिए कहेगी - चुनें कि आप क्या लॉन्च करना चाहते हैं:

SQL संपादक विंडो में निम्नलिखित सरल क्वेरी डालें और सहेजें और SQL चलाएँ पर क्लिक करें:

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ("cost" DOUBLE, "gate" VARCHAR(16));

CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM "cost", "gate"
FROM "SOURCE_SQL_STREAM_001"
WHERE "cost" < 5000
    and "gate" = 'Aeroflot';

रिलेशनल डेटाबेस में, आप रिकॉर्ड जोड़ने के लिए INSERT स्टेटमेंट और क्वेरी डेटा के लिए SELECT स्टेटमेंट का उपयोग करके तालिकाओं के साथ काम करते हैं। अमेज़ॅन किनेसिस डेटा एनालिटिक्स में, आप स्ट्रीम (स्ट्रीम) और पंप (पंप) के साथ काम करते हैं - निरंतर सम्मिलित अनुरोध जो एक एप्लिकेशन में एक स्ट्रीम से दूसरे स्ट्रीम में डेटा डालते हैं।

उपरोक्त प्रस्तुत SQL क्वेरी पाँच हजार रूबल से कम कीमत पर एअरोफ़्लोत टिकटों की खोज करती है। इन शर्तों को पूरा करने वाले सभी रिकॉर्ड DESTINATION_SQL_STREAM स्ट्रीम में रखे जाएंगे।

गंतव्य ब्लॉक में, विशेष_स्ट्रीम स्ट्रीम का चयन करें, और इन-एप्लिकेशन स्ट्रीम नाम DESTINATION_SQL_STREAM ड्रॉप-डाउन सूची में:

सभी जोड़तोड़ का परिणाम नीचे दी गई तस्वीर के समान होना चाहिए:

एसएनएस विषय बनाना और उसकी सदस्यता लेना

सिंपल नोटिफिकेशन सर्विस पर जाएं और वहां एयरलाइंस नाम से एक नया विषय बनाएं:

इस विषय की सदस्यता लें और वह मोबाइल फ़ोन नंबर बताएं जिस पर एसएमएस सूचनाएं भेजी जाएंगी:

DynamoDB में एक तालिका बनाएं

उनके एयरलाइन_टिकट स्ट्रीम से कच्चे डेटा को संग्रहीत करने के लिए, आइए DynamoDB में उसी नाम से एक तालिका बनाएं। हम प्राथमिक कुंजी के रूप में रिकॉर्ड_आईडी का उपयोग करेंगे:

लैम्ब्डा फ़ंक्शन कलेक्टर बनाना

आइए कलेक्टर नामक एक लैम्ब्डा फ़ंक्शन बनाएं, जिसका कार्य एयरलाइन_टिकट स्ट्रीम का सर्वेक्षण करना होगा और, यदि वहां नए रिकॉर्ड मिलते हैं, तो इन रिकॉर्ड्स को डायनेमोडीबी तालिका में डालें। जाहिर है, डिफ़ॉल्ट अधिकारों के अलावा, इस लैम्ब्डा के पास किनेसिस डेटा स्ट्रीम तक पढ़ने की पहुंच और डायनेमोडीबी तक लिखने की पहुंच होनी चाहिए।

कलेक्टर लैम्ब्डा फ़ंक्शन के लिए एक IAM भूमिका बनाना
सबसे पहले, आइए लैम्ब्डा के लिए Lambda-TicketsProcessingRole नामक एक नई IAM भूमिका बनाएँ:

परीक्षण उदाहरण के लिए, पूर्व-कॉन्फ़िगर AmazonKinesisReadOnlyAccess और AmazonDynamoDBFullAccess नीतियां काफी उपयुक्त हैं, जैसा कि नीचे चित्र में दिखाया गया है:

जब नई प्रविष्टियाँ एयरलाइन_स्ट्रीम में प्रवेश करती हैं तो इस लैम्ब्डा को किनेसिस के एक ट्रिगर द्वारा लॉन्च किया जाना चाहिए, इसलिए हमें एक नया ट्रिगर जोड़ने की आवश्यकता है:

जो कुछ बचा है वह कोड डालना और लैम्ब्डा को सहेजना है।

"""Parsing the stream and inserting into the DynamoDB table."""
import base64
import json
import boto3
from decimal import Decimal

DYNAMO_DB = boto3.resource('dynamodb')
TABLE_NAME = 'airline_tickets'

class TicketsParser:
    """Parsing info from the Stream."""

    def __init__(self, table_name, records):
        """Init method."""
        self.table = DYNAMO_DB.Table(table_name)
        self.json_data = TicketsParser.get_json_data(records)

    @staticmethod
    def get_json_data(records):
        """Return deserialized data from the stream."""
        decoded_record_data = ([base64.b64decode(record['kinesis']['data'])
                                for record in records])
        json_data = ([json.loads(decoded_record)
                      for decoded_record in decoded_record_data])
        return json_data

    @staticmethod
    def get_item_from_json(json_item):
        """Pre-process the json data."""
        new_item = {
            'record_id': json_item.get('record_id'),
            'cost': Decimal(json_item.get('cost')),
            'trip_class': json_item.get('trip_class'),
            'show_to_affiliates': json_item.get('show_to_affiliates'),
            'origin': json_item.get('origin'),
            'number_of_changes': int(json_item.get('number_of_changes')),
            'gate': json_item.get('gate'),
            'found_at': json_item.get('found_at'),
            'duration': int(json_item.get('duration')),
            'distance': int(json_item.get('distance')),
            'destination': json_item.get('destination'),
            'depart_date': json_item.get('depart_date'),
            'actual': json_item.get('actual')
        }
        return new_item

    def run(self):
        """Batch insert into the table."""
        with self.table.batch_writer() as batch_writer:
            for item in self.json_data:
                dynamodb_item = TicketsParser.get_item_from_json(item)
                batch_writer.put_item(dynamodb_item)

        print('Has been added ', len(self.json_data), 'items')

def lambda_handler(event, context):
    """Parse the stream and insert into the DynamoDB table."""
    print('Got event:', event)
    parser = TicketsParser(TABLE_NAME, event['Records'])
    parser.run()

लैम्ब्डा फ़ंक्शन नोटिफ़ायर बनाना

दूसरा लैम्ब्डा फ़ंक्शन, जो दूसरी स्ट्रीम (विशेष_स्ट्रीम) की निगरानी करेगा और एसएनएस को एक अधिसूचना भेजेगा, उसी तरह बनाया गया है। इसलिए, इस लैम्ब्डा के पास किनेसिस से पढ़ने और किसी दिए गए एसएनएस विषय पर संदेश भेजने की पहुंच होनी चाहिए, जिसे बाद में एसएनएस सेवा द्वारा इस विषय के सभी ग्राहकों (ईमेल, एसएमएस, आदि) को भेजा जाएगा।

IAM भूमिका बनाना
सबसे पहले, हम इस लैम्ब्डा के लिए IAM भूमिका लैम्ब्डा-किनेसिसअलार्म बनाते हैं, और फिर इस भूमिका को बनाए जा रहे अलार्म_नोटिफ़ायर लैम्ब्डा को सौंपते हैं:

इस लैम्ब्डा को विशेष_स्ट्रीम में प्रवेश करने के लिए नए रिकॉर्ड के लिए एक ट्रिगर पर काम करना चाहिए, इसलिए आपको ट्रिगर को उसी तरह कॉन्फ़िगर करने की आवश्यकता है जैसे हमने कलेक्टर लैम्ब्डा के लिए किया था।

इस लैम्ब्डा को कॉन्फ़िगर करना आसान बनाने के लिए, आइए एक नया पर्यावरण चर पेश करें - TOPIC_ARN, जहां हम एयरलाइंस विषय के ANR (अमेज़ॅन रीकोर्स नाम) रखते हैं:

और लैम्ब्डा कोड डालें, यह बिल्कुल भी जटिल नहीं है:

import boto3
import base64
import os

SNS_CLIENT = boto3.client('sns')
TOPIC_ARN = os.environ['TOPIC_ARN']


def lambda_handler(event, context):
    try:
        SNS_CLIENT.publish(TopicArn=TOPIC_ARN,
                           Message='Hi! I have found an interesting stuff!',
                           Subject='Airline tickets alarm')
        print('Alarm message has been successfully delivered')
    except Exception as err:
        print('Delivery failure', str(err))

ऐसा लगता है कि यहीं पर मैन्युअल सिस्टम कॉन्फ़िगरेशन पूरा हो गया है। जो कुछ बचा है वह परीक्षण करना और यह सुनिश्चित करना है कि हमने सब कुछ सही ढंग से कॉन्फ़िगर किया है।

टेराफ़ॉर्म कोड से परिनियोजन करें

आवश्यक तैयारी

terraform कोड से बुनियादी ढांचे को तैनात करने के लिए एक बहुत ही सुविधाजनक ओपन-सोर्स टूल है। इसका अपना सिंटैक्स है जिसे सीखना आसान है और कैसे और क्या तैनात करना है इसके कई उदाहरण हैं। एटम एडिटर या विज़ुअल स्टूडियो कोड में कई उपयोगी प्लगइन्स हैं जो टेराफॉर्म के साथ काम करना आसान बनाते हैं।

आप वितरण डाउनलोड कर सकते हैं अत:. सभी टेराफॉर्म क्षमताओं का विस्तृत विश्लेषण इस लेख के दायरे से परे है, इसलिए हम खुद को मुख्य बिंदुओं तक ही सीमित रखेंगे।

कैसे चलाना है?

प्रोजेक्ट का पूरा कोड है मेरे भंडार में. हम रिपॉजिटरी को अपने पास क्लोन करते हैं। शुरू करने से पहले, आपको यह सुनिश्चित करना होगा कि आपने AWS CLI स्थापित और कॉन्फ़िगर किया है, क्योंकि... टेराफ़ॉर्म ~/.aws/credentials फ़ाइल में क्रेडेंशियल खोजेगा।

टेराफॉर्म वर्तमान में क्लाउड में हमारे लिए क्या बना रहा है यह देखने के लिए संपूर्ण बुनियादी ढांचे को तैनात करने से पहले योजना कमांड चलाना एक अच्छा अभ्यास है:

terraform.exe plan

आपको सूचनाएं भेजने के लिए एक फ़ोन नंबर दर्ज करने के लिए कहा जाएगा। इस स्तर पर इसे दर्ज करना आवश्यक नहीं है।

कार्यक्रम की संचालन योजना का विश्लेषण करने के बाद, हम संसाधन बनाना शुरू कर सकते हैं:

terraform.exe apply

इस आदेश को भेजने के बाद, आपको फिर से एक फ़ोन नंबर दर्ज करने के लिए कहा जाएगा; जब वास्तव में कार्रवाई करने के बारे में कोई प्रश्न दिखाया जाए तो "हां" डायल करें। यह आपको संपूर्ण बुनियादी ढाँचा स्थापित करने, EC2 के सभी आवश्यक कॉन्फ़िगरेशन करने, लैम्ब्डा फ़ंक्शंस को तैनात करने आदि की अनुमति देगा।

टेराफ़ॉर्म कोड के माध्यम से सभी संसाधन सफलतापूर्वक बनाए जाने के बाद, आपको किनेसिस एनालिटिक्स एप्लिकेशन के विवरण में जाने की आवश्यकता है (दुर्भाग्य से, मुझे यह नहीं मिला कि इसे सीधे कोड से कैसे किया जाए)।

एप्लिकेशन लॉन्च करें:

इसके बाद, आपको ड्रॉप-डाउन सूची से चयन करके इन-एप्लिकेशन स्ट्रीम नाम स्पष्ट रूप से सेट करना होगा:

अब सब कुछ जाने के लिए तैयार है.

एप्लिकेशन का परीक्षण

भले ही आपने सिस्टम को मैन्युअल रूप से या टेराफॉर्म कोड के माध्यम से कैसे भी तैनात किया हो, यह उसी तरह काम करेगा।

हम SSH के माध्यम से EC2 वर्चुअल मशीन में लॉग इन करते हैं जहां काइनेसिस एजेंट स्थापित है और api_caller.py स्क्रिप्ट चलाते हैं

sudo ./api_caller.py TOKEN

आपको बस अपने नंबर पर एक एसएमएस की प्रतीक्षा करनी है:

एसएमएस - लगभग 1 मिनट में फ़ोन पर एक संदेश आता है:

यह देखना बाकी है कि क्या रिकॉर्ड बाद के अधिक विस्तृत विश्लेषण के लिए डायनेमोडीबी डेटाबेस में सहेजे गए थे। एयरलाइन_टिकट तालिका में लगभग निम्नलिखित डेटा शामिल है:

निष्कर्ष

किए गए कार्य के दौरान, अमेज़ॅन किनेसिस पर आधारित एक ऑनलाइन डेटा प्रोसेसिंग सिस्टम बनाया गया था। किनेसिस डेटा स्ट्रीम और वास्तविक समय विश्लेषण के साथ किनेसिस एजेंट का उपयोग करने के विकल्पों पर एसक्यूएल कमांड का उपयोग करने वाले किनेसिस एनालिटिक्स के साथ-साथ अन्य एडब्ल्यूएस सेवाओं के साथ अमेज़ॅन किनेसिस की बातचीत पर विचार किया गया।

हमने उपरोक्त प्रणाली को दो तरीकों से तैनात किया है: एक काफी लंबा मैनुअल और टेराफॉर्म कोड से एक त्वरित।

सभी प्रोजेक्ट स्रोत कोड उपलब्ध है मेरे GitHub रिपॉजिटरी में, मेरा सुझाव है कि आप स्वयं को इससे परिचित कर लें।

मुझे लेख पर चर्चा करने में खुशी हो रही है, मैं आपकी टिप्पणियों की प्रतीक्षा कर रहा हूं। मैं रचनात्मक आलोचना की आशा करता हूँ।

मैं आपको सफलता की कामना करता हूं!

स्रोत: www.habr.com