Amazon Kinesis सह Aviasales API एकत्रीकरण आणि सर्व्हरलेस साधेपणा

अहो हाब्र!

तुम्हाला उडणारी विमाने आवडतात का? मला ते आवडते, परंतु सेल्फ-आयसोलेशन दरम्यान मी एका सुप्रसिद्ध स्त्रोत - एव्हियासेल्सच्या हवाई तिकिटावरील डेटाचे विश्लेषण करण्याच्या प्रेमात पडलो.

आज आम्ही Amazon Kinesis च्या कार्याचे विश्लेषण करू, रीअल-टाइम विश्लेषणासह स्ट्रीमिंग सिस्टम तयार करू, Amazon DynamoDB NoSQL डेटाबेस मुख्य डेटा स्टोरेज म्हणून स्थापित करू आणि मनोरंजक तिकिटांसाठी SMS सूचना सेट करू.

सर्व तपशील कट अंतर्गत आहेत! जा!

परिचय

उदाहरणार्थ, आम्हाला प्रवेश आवश्यक आहे Aviasales API. त्यात प्रवेश विनामूल्य आणि निर्बंधांशिवाय प्रदान केला जातो; डेटा ऍक्सेस करण्यासाठी तुमचे API टोकन प्राप्त करण्यासाठी तुम्हाला फक्त “डेव्हलपर” विभागात नोंदणी करणे आवश्यक आहे.

या लेखाचा मुख्य उद्देश AWS मधील माहिती प्रवाहाच्या वापराबद्दल सामान्य समज देणे आहे; आम्ही हे लक्षात घेतो की वापरलेल्या API द्वारे परत केलेला डेटा काटेकोरपणे अद्ययावत नाही आणि कॅशेमधून प्रसारित केला जातो, जे गेल्या ४८ तासांपासून Aviasales.ru आणि Jetradar.com साइट्सच्या वापरकर्त्यांनी केलेल्या शोधांवर आधारित.

उत्पादन करणार्‍या मशीनवर स्थापित केलेला Kinesis-एजंट, API द्वारे प्राप्त झालेला, Kinesis डेटा विश्लेषणाद्वारे इच्छित प्रवाहात डेटा स्वयंचलितपणे पार्स करेल आणि प्रसारित करेल. या प्रवाहाची कच्ची आवृत्ती थेट स्टोअरवर लिहिली जाईल. DynamoDB मध्ये तैनात केलेला कच्चा डेटा स्टोरेज BI टूल्स, जसे की AWS Quick Sight द्वारे सखोल तिकीट विश्लेषणास अनुमती देईल.

संपूर्ण पायाभूत सुविधा तैनात करण्यासाठी आम्ही दोन पर्यायांचा विचार करू:

• मॅन्युअल - AWS व्यवस्थापन कन्सोल द्वारे;
• टेराफॉर्म कोडमधील पायाभूत सुविधा आळशी ऑटोमेटर्ससाठी आहे;

विकसित प्रणालीचे आर्किटेक्चर

वापरलेले घटक:

• Aviasales API — या API द्वारे परत केलेला डेटा पुढील सर्व कामांसाठी वापरला जाईल;
• EC2 निर्माता उदाहरण — क्लाउडमधील एक नियमित व्हर्च्युअल मशीन ज्यावर इनपुट डेटा प्रवाह तयार केला जाईल:
  • किनेसिस एजंट मशीनवर स्थानिकरित्या स्थापित केलेला Java अनुप्रयोग आहे जो Kinesis (Kinesis Data Streams किंवा Kinesis Firehose) ला डेटा गोळा करण्याचा आणि पाठवण्याचा एक सोपा मार्ग प्रदान करतो. एजंट निर्दिष्ट निर्देशिकेतील फायलींच्या संचाचे सतत निरीक्षण करतो आणि किनेसिसला नवीन डेटा पाठवतो;
  • API कॉलर स्क्रिप्ट — एक पायथन स्क्रिप्ट जी API ला विनंती करते आणि प्रतिसाद एका फोल्डरमध्ये ठेवते ज्याचे निरीक्षण किनेसिस एजंटद्वारे केले जाते;
• Kinesis डेटा प्रवाह — विस्तृत स्केलिंग क्षमतांसह रिअल-टाइम डेटा स्ट्रीमिंग सेवा;
• Kinesis विश्लेषण ही एक सर्व्हरलेस सेवा आहे जी रिअल टाइममध्ये स्ट्रीमिंग डेटाचे विश्लेषण सुलभ करते. Amazon Kinesis Data Analytics अनुप्रयोग संसाधने कॉन्फिगर करते आणि येणार्‍या डेटाचा कोणताही खंड हाताळण्यासाठी स्वयंचलितपणे स्केल करते;
• एडब्ल्यूएस लंबडा — एक सेवा जी तुम्हाला सर्व्हरचा बॅकअप न घेता किंवा सेट अप न करता कोड चालवण्याची परवानगी देते. प्रत्येक कॉलसाठी सर्व संगणकीय शक्ती स्वयंचलितपणे मोजली जाते;
• अ‍ॅमेझॉन डायनामोडीबी - की-व्हॅल्यू जोड्या आणि दस्तऐवजांचा डेटाबेस जो कोणत्याही स्केलवर चालत असताना 10 मिलिसेकंदांपेक्षा कमी विलंब प्रदान करतो. DynamoDB वापरताना, तुम्हाला कोणत्याही सर्व्हरची तरतूद, पॅच किंवा व्यवस्थापित करण्याची आवश्यकता नाही. DynamoDB उपलब्ध संसाधनांचे प्रमाण समायोजित करण्यासाठी आणि उच्च कार्यप्रदर्शन राखण्यासाठी आपोआप सारण्यांचे स्केल करते. सिस्टम प्रशासनाची आवश्यकता नाही;
• ऍमेझॉन SNS - प्रकाशक-सदस्यता (पब/सब) मॉडेलचा वापर करून संदेश पाठवण्‍यासाठी एक पूर्णतः व्यवस्थापित सेवा, जिच्‍या मदतीने तुम्ही मायक्रोसर्व्हिसेस, वितरित सिस्‍टम आणि सर्व्हरलेस अॅप्लिकेशन वेगळे करू शकता. मोबाइल पुश सूचना, एसएमएस संदेश आणि ईमेलद्वारे अंतिम वापरकर्त्यांना माहिती पाठवण्यासाठी SNS चा वापर केला जाऊ शकतो.

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

डेटा प्रवाहाचे अनुकरण करण्यासाठी, मी Aviasales API द्वारे परत केलेली एअरलाइन तिकीट माहिती वापरण्याचे ठरवले. IN दस्तऐवजीकरण वेगवेगळ्या पद्धतींची विस्तृत यादी, त्यापैकी एक घेऊ - “मासिक किंमत कॅलेंडर”, जे महिन्याच्या प्रत्येक दिवसासाठी किंमती परत करते, हस्तांतरणाच्या संख्येनुसार गटबद्ध केले जाते. तुम्ही विनंतीमध्ये शोध महिना निर्दिष्ट न केल्यास, वर्तमान महिन्याच्या पुढील महिन्यासाठी माहिती परत केली जाईल.

तर, चला नोंदणी करून आमचे टोकन मिळवूया.

खाली एक उदाहरण विनंती आहे:

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

विनंतीमध्ये टोकन निर्दिष्ट करून 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
   }]
}

वरील उदाहरण API प्रतिसाद सेंट पीटर्सबर्ग ते फुकचे तिकीट दाखवते... अरे काय स्वप्न आहे...
मी कझानचा असल्यामुळे आणि फुकेत आता "फक्त एक स्वप्न" आहे, चला सेंट पीटर्सबर्ग ते कझान तिकीट शोधूया.

हे गृहीत धरते की तुमच्याकडे आधीपासूनच AWS खाते आहे. मी ताबडतोब या वस्तुस्थितीकडे विशेष लक्ष वेधू इच्छितो की किनेसिस आणि एसएमएसद्वारे सूचना पाठवणे वार्षिक मध्ये समाविष्ट नाही फ्री टियर (विनामूल्य वापर). परंतु असे असूनही, काही डॉलर्स लक्षात घेऊन, प्रस्तावित प्रणाली तयार करणे आणि त्यासह खेळणे शक्य आहे. आणि, अर्थातच, सर्व संसाधने यापुढे आवश्यक नसल्यानंतर ते हटविण्यास विसरू नका.

सुदैवाने, आम्ही आमच्या मासिक मोफत मर्यादा पूर्ण केल्यास DynamoDb आणि lambda फंक्शन्स आमच्यासाठी विनामूल्य असतील. उदाहरणार्थ, DynamoDB साठी: 25 GB स्टोरेज, 25 WCU/RCU आणि 100 दशलक्ष क्वेरी. आणि दरमहा एक दशलक्ष लॅम्बडा फंक्शन कॉल.

मॅन्युअल सिस्टम उपयोजन

Kinesis डेटा प्रवाह सेट करणे

चला Kinesis डेटा स्ट्रीम सेवेवर जाऊ आणि दोन नवीन प्रवाह तयार करू, प्रत्येकासाठी एक शार्ड.

शार्ड म्हणजे काय?
शार्ड हे Amazon Kinesis प्रवाहाचे मूलभूत डेटा ट्रान्सफर युनिट आहे. एक विभाग 1 MB/s वेगाने इनपुट डेटा ट्रान्सफर आणि 2 MB/s वेगाने डेटा ट्रान्सफर प्रदान करतो. एक विभाग प्रति सेकंद 1000 PUT नोंदींना समर्थन देतो. डेटा प्रवाह तयार करताना, आपल्याला विभागांची आवश्यक संख्या निर्दिष्ट करणे आवश्यक आहे. उदाहरणार्थ, तुम्ही दोन विभागांसह डेटा प्रवाह तयार करू शकता. हा डेटा स्ट्रीम 2 MB/s वर इनपुट डेटा ट्रान्सफर आणि 4 MB/s वर आउटपुट डेटा ट्रान्सफर प्रदान करेल, प्रति सेकंद 2000 PUT रेकॉर्डला समर्थन देईल.

तुमच्या प्रवाहात जितके जास्त शार्ड्स, तितके त्याचे थ्रूपुट जास्त. तत्वतः, अशा प्रकारे प्रवाह मोजले जातात - शार्ड्स जोडून. परंतु आपल्याकडे जितके अधिक शार्ड्स असतील तितकी किंमत जास्त. प्रत्येक शार्डची किंमत 1,5 सेंट प्रति तास आणि प्रत्येक दशलक्ष PUT पेलोड युनिटसाठी अतिरिक्त 1.4 सेंट.

नावासह एक नवीन प्रवाह तयार करूया विमान_तिकीटे, 1 शार्ड त्याच्यासाठी पुरेसे असेल:

आता नावाचा दुसरा धागा बनवू विशेष_प्रवाह:

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

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

महत्त्वाची टीप: उदाहरणार्थ, तुम्ही इमेज वापरावी - Amazon Linux AMI 2018.03.0, त्यात Kinesis Agent लाँच करण्यासाठी कमी सेटिंग्ज आहेत.

EC2 सेवेवर जा, नवीन व्हर्च्युअल मशीन तयार करा, t2.micro प्रकारासह इच्छित AMI निवडा, जे फ्री टियरमध्ये समाविष्ट आहे:

नवीन निर्मित व्हर्च्युअल मशीन किनेसिस सेवेशी संवाद साधण्यास सक्षम होण्यासाठी, त्यास तसे करण्याचे अधिकार दिले पाहिजेत. हे करण्याचा सर्वोत्तम मार्ग म्हणजे IAM भूमिका नियुक्त करणे. म्हणून, चरण 3: इन्स्टन्स तपशील कॉन्फिगर करा स्क्रीनवर, तुम्ही निवडले पाहिजे नवीन IAM भूमिका तयार करा:

EC2 साठी IAM भूमिका तयार करणे

उघडणाऱ्या विंडोमध्ये, आम्ही EC2 साठी नवीन भूमिका तयार करत आहोत हे निवडा आणि परवानग्या विभागात जा:

प्रशिक्षणाचे उदाहरण वापरून, आम्हाला संसाधन अधिकारांच्या ग्रॅन्युलर कॉन्फिगरेशनच्या सर्व गुंतागुंतींमध्ये जाण्याची गरज नाही, म्हणून आम्ही Amazon द्वारे पूर्व-कॉन्फिगर केलेली धोरणे निवडू: AmazonKinesisFullAccess आणि CloudWatchFullAccess.

चला या भूमिकेसाठी काही अर्थपूर्ण नाव देऊ, उदाहरणार्थ: EC2-KinesisStreams-FullAccess. परिणाम खालील चित्रात दर्शविल्याप्रमाणेच असावा:

ही नवीन भूमिका तयार केल्यानंतर, तयार केलेल्या व्हर्च्युअल मशीनच्या उदाहरणाशी संलग्न करण्यास विसरू नका:

आम्ही या स्क्रीनवर दुसरे काहीही बदलत नाही आणि पुढील विंडोवर जाऊ.

हार्ड ड्राइव्ह सेटिंग्ज डीफॉल्ट, तसेच टॅग म्हणून सोडल्या जाऊ शकतात (जरी टॅग वापरणे चांगले आहे, किमान उदाहरणाला नाव द्या आणि वातावरण सूचित करा).

आता आम्ही स्टेप 6: कॉन्फिगर सिक्युरिटी ग्रुप टॅबवर आहोत, जिथे तुम्हाला एक नवीन तयार करायचा आहे किंवा तुमचा सध्याचा सिक्युरिटी ग्रुप निर्दिष्ट करायचा आहे, जो तुम्हाला ssh (पोर्ट 22) द्वारे उदाहरणाशी कनेक्ट करण्याची परवानगी देतो. तेथे स्त्रोत -> माझा आयपी निवडा आणि तुम्ही उदाहरण लाँच करू शकता.

ते चालू स्थितीवर स्विच करताच, तुम्ही ssh द्वारे त्याच्याशी कनेक्ट करण्याचा प्रयत्न करू शकता.

Kinesis एजंटसह कार्य करण्यास सक्षम होण्यासाठी, मशीनशी यशस्वीरित्या कनेक्ट केल्यानंतर, आपण टर्मिनलमध्ये खालील आदेश प्रविष्ट करणे आवश्यक आहे:

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

API प्रतिसाद जतन करण्यासाठी फोल्डर तयार करूया:

sudo mkdir /var/log/airline_tickets

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

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

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 विस्तारासह फाइल्सचे निरीक्षण करेल, त्यांचे विश्लेषण करेल आणि त्यांना airline_tickets प्रवाहात स्थानांतरित करेल.

आम्ही सेवा रीस्टार्ट करतो आणि ती चालू असल्याची खात्री करतो:

sudo service aws-kinesis-agent restart

आता पायथन स्क्रिप्ट डाउनलोड करूया जी API कडून डेटाची विनंती करेल:

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 कडून डेटाची विनंती करते आणि Kinesis एजंट स्कॅन करत असलेल्या निर्देशिकेत प्राप्त प्रतिसाद जतन करते. या स्क्रिप्टची अंमलबजावणी अगदी मानक आहे, एक TicketsApi वर्ग आहे, तो तुम्हाला असिंक्रोनसपणे API खेचण्याची परवानगी देतो. आम्ही टोकनसह हेडर पास करतो आणि या वर्गाला पॅरामीटर्सची विनंती करतो:

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

आणि आम्ही एजंट लॉगमधील कामाचा परिणाम आणि airline_tickets डेटा प्रवाहातील मॉनिटरिंग टॅबवर पाहतो:

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

जसे आपण पाहू शकता, सर्वकाही कार्य करते आणि Kinesis एजंट यशस्वीरित्या प्रवाहात डेटा पाठवते. आता ग्राहक कॉन्फिगर करूया.

Kinesis डेटा विश्लेषणे सेट करत आहे

चला संपूर्ण सिस्टमच्या मध्यवर्ती घटकाकडे वळूया - Kinesis_analytics_airlines_app नावाचे Kinesis डेटा अॅनालिटिक्स मध्ये एक नवीन अनुप्रयोग तयार करूया:

किनेसिस डेटा अॅनालिटिक्स तुम्हाला एसक्यूएल भाषा वापरून किनेसिस स्ट्रीममधून रिअल-टाइम डेटा विश्लेषणे करण्याची परवानगी देते. ही एक पूर्णपणे ऑटोस्केलिंग सेवा आहे (कायनेसिस स्ट्रीम्सच्या विपरीत) जी:

1. स्रोत डेटाच्या विनंत्यांवर आधारित तुम्हाला नवीन प्रवाह (आउटपुट प्रवाह) तयार करण्याची परवानगी देते;
2. ऍप्लिकेशन्स चालू असताना आलेल्या त्रुटींसह प्रवाह प्रदान करते (एरर स्ट्रीम);
3. इनपुट डेटा योजना स्वयंचलितपणे निर्धारित करू शकते (आवश्यक असल्यास ते व्यक्तिचलितपणे पुन्हा परिभाषित केले जाऊ शकते).

ही स्वस्त सेवा नाही - 0.11 USD प्रति तास काम, त्यामुळे तुम्ही ती काळजीपूर्वक वापरावी आणि पूर्ण झाल्यावर ती हटवावी.

चला ऍप्लिकेशनला डेटा स्त्रोताशी कनेक्ट करूया:

आम्ही कनेक्ट करणार आहोत तो प्रवाह निवडा (airline_tickets):

पुढे, तुम्हाला नवीन IAM भूमिका संलग्न करणे आवश्यक आहे जेणेकरून अनुप्रयोग प्रवाहातून वाचू शकेल आणि प्रवाहावर लिहू शकेल. हे करण्यासाठी, प्रवेश परवानगी ब्लॉकमध्ये काहीही न बदलणे पुरेसे आहे:

आता प्रवाहात डेटा स्कीमा शोधण्याची विनंती करूया; हे करण्यासाठी, “डिस्कव्हर स्कीमा” बटणावर क्लिक करा. परिणामी, IAM भूमिका अद्यतनित केली जाईल (एक नवीन तयार केली जाईल) आणि स्ट्रीममध्ये आधीपासून आलेल्या डेटावरून स्कीमा शोध सुरू केला जाईल:

आता तुम्हाला 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 स्टेटमेंट वापरता. Amazon Kinesis डेटा अ‍ॅनालिटिक्समध्ये, तुम्ही स्ट्रीम (स्ट्रीम) आणि पंप (PUMPs) सह कार्य करता—सतत इन्सर्ट विनंत्या ज्या एका अॅप्लिकेशनमधील एका स्ट्रीममधून दुसऱ्या स्ट्रीममध्ये डेटा घालतात.

वर सादर केलेली SQL क्वेरी एरोफ्लॉट तिकिटांसाठी पाच हजार रूबलपेक्षा कमी किंमतीत शोधते. या अटी पूर्ण करणाऱ्या सर्व नोंदी DESTINATION_SQL_STREAM प्रवाहात ठेवल्या जातील.

गंतव्य ब्लॉकमध्ये, स्पेशल_स्ट्रीम स्ट्रीम निवडा आणि अॅप्लिकेशन स्ट्रीम नाव DESTINATION_SQL_STREAM ड्रॉप-डाउन सूचीमध्ये:

सर्व हाताळणीचा परिणाम खालील चित्रासारखाच असावा:

SNS विषय तयार करणे आणि सदस्यता घेणे

सिंपल नोटिफिकेशन सेवेवर जा आणि तेथे एअरलाइन्स नावाने एक नवीन विषय तयार करा:

या विषयाची सदस्यता घ्या आणि मोबाइल फोन नंबर सूचित करा ज्यावर एसएमएस सूचना पाठवल्या जातील:

DynamoDB मध्ये एक टेबल तयार करा

त्यांच्या airline_tickets स्ट्रीममधून रॉ डेटा संग्रहित करण्यासाठी, DynamoDB मध्ये त्याच नावाने एक टेबल तयार करूया. आम्ही प्राथमिक की म्हणून रेकॉर्ड_आयडी वापरू:

लॅम्बडा फंक्शन कलेक्टर तयार करणे

चला कलेक्टर नावाचे एक लॅम्बडा फंक्शन तयार करूया, ज्याचे कार्य एअरलाइन_टिकेट स्ट्रीमचे मतदान करणे असेल आणि तेथे नवीन रेकॉर्ड आढळल्यास, हे रेकॉर्ड डायनामोडीबी टेबलमध्ये समाविष्ट करा. साहजिकच, डीफॉल्ट अधिकारांव्यतिरिक्त, या लॅम्बडाने Kinesis डेटा स्ट्रीमचा वाचन आणि DynamoDB वर लेखन प्रवेश असणे आवश्यक आहे.

कलेक्टर लॅम्बडा फंक्शनसाठी IAM भूमिका तयार करणे
प्रथम, Lambda-TicketsProcessingRole नावाच्या लॅम्बडासाठी एक नवीन IAM भूमिका तयार करूया:

चाचणी उदाहरणासाठी, खालील चित्रात दाखवल्याप्रमाणे, पूर्व-कॉन्फिगर केलेली AmazonKinesisReadOnlyAccess आणि AmazonDynamoDBFullAccess धोरणे अगदी योग्य आहेत:

जेव्हा नवीन नोंदी airline_stream मध्ये प्रवेश करतात तेव्हा हा lambda Kinesis च्या ट्रिगरद्वारे लॉन्च केला जावा, म्हणून आम्हाला एक नवीन ट्रिगर जोडण्याची आवश्यकता आहे:

कोड घालणे आणि लॅम्बडा जतन करणे बाकी आहे.

"""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()

लॅम्बडा फंक्शन नोटिफायर तयार करणे

दुसरे lambda फंक्शन, जे दुसऱ्या प्रवाहाचे (विशेष_स्ट्रीम) निरीक्षण करेल आणि SNS ला सूचना पाठवेल, त्याच प्रकारे तयार केले आहे. म्हणून, या lambda ला Kinesis वरून वाचण्यासाठी आणि दिलेल्या SNS विषयावर संदेश पाठवण्याचा प्रवेश असणे आवश्यक आहे, जे नंतर SNS सेवेद्वारे या विषयाच्या सर्व सदस्यांना (ईमेल, एसएमएस, इ.) पाठवले जाईल.

IAM भूमिका तयार करणे
प्रथम, आम्ही या लॅम्बडासाठी IAM भूमिका Lambda-KinesisAlarm तयार करतो आणि नंतर तयार होत असलेल्या अलार्म_नोटिफायरला ही भूमिका नियुक्त करतो:

स्पेशल_स्ट्रीममध्ये प्रवेश करण्यासाठी या लॅम्बडाने नवीन रेकॉर्डसाठी ट्रिगरवर कार्य केले पाहिजे, म्हणून आम्ही कलेक्टर लॅम्बडासाठी जसे केले तसे तुम्हाला ट्रिगर कॉन्फिगर करणे आवश्यक आहे.

हा लॅम्बडा कॉन्फिगर करणे सोपे करण्यासाठी, एक नवीन पर्यावरण व्हेरिएबल - TOPIC_ARN सादर करूया, जिथे आम्ही एअरलाइन्स विषयाचा ANR (Amazon Recourse Names) ठेवतो:

आणि लॅम्बडा कोड घाला, हे अजिबात क्लिष्ट नाही:

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))

असे दिसते की येथेच मॅन्युअल सिस्टम कॉन्फिगरेशन पूर्ण झाले आहे. आम्ही सर्व काही योग्यरितीने कॉन्फिगर केले आहे याची चाचणी घेणे आणि खात्री करणे बाकी आहे.

टेराफॉर्म कोडमधून उपयोजित करा

आवश्यक तयारी

टेराफॉर्म कोडमधून पायाभूत सुविधा तैनात करण्यासाठी एक अतिशय सोयीस्कर मुक्त-स्रोत साधन आहे. त्याची स्वतःची वाक्यरचना आहे जी शिकणे सोपे आहे आणि कसे आणि काय उपयोजित करावे याची अनेक उदाहरणे आहेत. अॅटम एडिटर किंवा व्हिज्युअल स्टुडिओ कोडमध्ये अनेक सुलभ प्लगइन आहेत जे टेराफॉर्मसह काम करणे सोपे करतात.

आपण वितरण डाउनलोड करू शकता येथून. सर्व टेराफॉर्म क्षमतांचे तपशीलवार विश्लेषण या लेखाच्या व्याप्तीच्या पलीकडे आहे, म्हणून आम्ही स्वतःला मुख्य मुद्द्यांपुरते मर्यादित करू.

कसे सुरू करावे

प्रकल्पाचा संपूर्ण कोड आहे माझ्या भांडारात. आम्ही स्वतःसाठी भांडार क्लोन करतो. सुरू करण्यापूर्वी, तुम्ही AWS CLI स्थापित आणि कॉन्फिगर केले असल्याची खात्री करणे आवश्यक आहे, कारण... टेराफॉर्म ~/.aws/credentials फाइलमध्ये क्रेडेंशियल शोधेल.

टेराफॉर्म सध्या क्लाउडमध्ये आमच्यासाठी काय तयार करत आहे हे पाहण्यासाठी संपूर्ण पायाभूत सुविधा तैनात करण्यापूर्वी प्लॅन कमांड चालवणे हा एक चांगला सराव आहे:

terraform.exe plan

तुम्हाला सूचना पाठवण्यासाठी फोन नंबर एंटर करण्यास सांगितले जाईल. या टप्प्यावर ते प्रविष्ट करणे आवश्यक नाही.

प्रोग्रामच्या ऑपरेशन योजनेचे विश्लेषण केल्यावर, आम्ही संसाधने तयार करणे सुरू करू शकतो:

terraform.exe apply

ही आज्ञा पाठवल्यानंतर, तुम्हाला पुन्हा फोन नंबर प्रविष्ट करण्यास सांगितले जाईल; जेव्हा क्रिया प्रत्यक्षात केल्याबद्दल प्रश्न दर्शविला जाईल तेव्हा "होय" डायल करा. हे तुम्हाला संपूर्ण इन्फ्रास्ट्रक्चर सेट करण्यास, EC2 चे सर्व आवश्यक कॉन्फिगरेशन पूर्ण करण्यास, लॅम्बडा फंक्शन्स तैनात करण्यास अनुमती देईल.

टेराफॉर्म कोडद्वारे सर्व संसाधने यशस्वीरित्या तयार केल्यावर, तुम्हाला किनेसिस अॅनालिटिक्स ऍप्लिकेशनच्या तपशीलांमध्ये जाण्याची आवश्यकता आहे (दुर्दैवाने, कोडमधून हे कसे करायचे ते मला सापडले नाही).

अनुप्रयोग लाँच करा:

यानंतर, तुम्ही ड्रॉप-डाउन सूचीमधून निवडून ऍप्लिकेशनमधील प्रवाहाचे नाव स्पष्टपणे सेट केले पाहिजे:

आता सर्वकाही जाण्यासाठी तयार आहे.

अर्जाची चाचणी घेत आहे

तुम्‍ही सिस्‍टम कसे उपयोजित केले, मॅन्युअली किंवा टेराफॉर्म कोडद्वारे, ते तसेच कार्य करेल.

आम्ही SSH द्वारे EC2 व्हर्च्युअल मशीनवर लॉग इन करतो जिथे Kinesis Agent स्थापित आहे आणि api_caller.py स्क्रिप्ट चालवतो

sudo ./api_caller.py TOKEN

तुम्हाला फक्त तुमच्या नंबरवर एसएमएसची प्रतीक्षा करायची आहे:

एसएमएस - जवळजवळ 1 मिनिटात फोनवर संदेश येतो:

त्यानंतरच्या, अधिक तपशीलवार विश्लेषणासाठी DynamoDB डेटाबेसमध्ये रेकॉर्ड जतन केले होते की नाही हे पाहणे बाकी आहे. airline_tickets टेबलमध्ये अंदाजे खालील डेटा आहे:

निष्कर्ष

केलेल्या कामाच्या दरम्यान, Amazon Kinesis वर आधारित ऑनलाइन डेटा प्रोसेसिंग सिस्टम तयार करण्यात आली. किनेसिस डेटा स्ट्रीम्स आणि रीअल-टाइम अॅनालिटिक्स किनेसिस अॅनालिटिक्सच्या संयोगाने एसक्यूएल कमांड्स वापरून किनेसिस एजंट वापरण्याचे पर्याय, तसेच अॅमेझॉन किनेसिसचा इतर AWS सेवांसह परस्परसंवादाचा विचार करण्यात आला.

आम्ही वरील प्रणाली दोन प्रकारे तैनात केली: एक लांब मॅन्युअल आणि टेराफॉर्म कोडमधून एक द्रुत.

सर्व प्रकल्प स्त्रोत कोड उपलब्ध आहे माझ्या GitHub भांडारात, मी सुचवितो की तुम्ही त्याच्याशी परिचित व्हा.

लेखावर चर्चा करण्यात मला आनंद झाला, मी तुमच्या टिप्पण्यांसाठी उत्सुक आहे. मला विधायक टीकेची अपेक्षा आहे.

मी तुम्हाला यश इच्छितो!

स्त्रोत: www.habr.com