Amazon Kinesis र सर्भररहित सादगीको साथ Aviasales API एकीकरण

हे हाबर!

के तपाई उड्ने हवाइजहाज मन पराउनुहुन्छ? मलाई यो मनपर्छ, तर आत्म-पृथकताको समयमा म पनि एक प्रसिद्ध स्रोत - Aviasales बाट एयर टिकटहरूमा डेटा विश्लेषण गर्न प्रेममा परें।

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

सबै विवरणहरू कट अन्तर्गत छन्! जाउ!

परिचय

उदाहरणका लागि, हामीलाई पहुँच चाहिन्छ Aviasales API। यसमा पहुँच निःशुल्क र प्रतिबन्ध बिना प्रदान गरिएको छ; डाटा पहुँच गर्नको लागि तपाइँको एपीआई टोकन प्राप्त गर्न "विकासकर्ता" खण्डमा दर्ता गर्न आवश्यक छ।

यस लेखको मुख्य उद्देश्य AWS मा जानकारी स्ट्रिमिङको प्रयोगको सामान्य बुझाइ दिनु हो; हामीले प्रयोग गरेको API द्वारा फिर्ता गरिएको डाटा कडा रूपमा अप-टु-डेट छैन र क्यासबाट प्रसारित हुन्छ भन्ने कुरालाई ध्यानमा राख्छौं। पछिल्लो ४८ घण्टामा Aviasales.ru र Jetradar.com साइटका प्रयोगकर्ताहरूले खोजेको आधारमा बनाइएको हो।

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

हामी सम्पूर्ण पूर्वाधार परिचालनको लागि दुई विकल्पहरू विचार गर्नेछौं:

• म्यानुअल - AWS व्यवस्थापन कन्सोल मार्फत;
• टेराफर्म कोडबाट पूर्वाधार अल्छी स्वचालनकर्ताहरूको लागि हो;

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

प्रयोग गरिएका अवयवहरू:

• Aviasales API — यस API द्वारा फिर्ता गरिएको डाटा सबै पछिल्ला कामका लागि प्रयोग गरिनेछ;
• EC2 निर्माता उदाहरण - क्लाउडमा एक नियमित भर्चुअल मेसिन जसमा इनपुट डेटा स्ट्रिम उत्पन्न हुनेछ:
  • Kinesis एजेन्ट मेसिनमा स्थानीय रूपमा स्थापित एउटा Java अनुप्रयोग हो जसले Kinesis (Kinesis Data Streams वा Kinesis Firehose) लाई डाटा सङ्कलन र पठाउन सजिलो तरिका प्रदान गर्दछ। एजेन्टले निर्दिष्ट डाइरेक्टरीहरूमा फाइलहरूको सेटलाई निरन्तर निगरानी गर्दछ र Kinesis मा नयाँ डाटा पठाउँछ;
  • API कलर स्क्रिप्ट - एक पाइथन स्क्रिप्ट जसले API मा अनुरोध गर्दछ र प्रतिक्रियालाई फोल्डरमा राख्छ जुन किनेसिस एजेन्टद्वारा निगरानी गरिन्छ;
• Kinesis डाटा स्ट्रिमहरू - व्यापक स्केलिंग क्षमताहरूको साथ वास्तविक-समय डाटा स्ट्रिमिङ सेवा;
• Kinesis एनालिटिक्स सर्भररहित सेवा हो जसले वास्तविक समयमा स्ट्रिमिङ डेटाको विश्लेषणलाई सरल बनाउँछ। Amazon Kinesis डेटा एनालिटिक्सले अनुप्रयोग स्रोतहरू कन्फिगर गर्दछ र आगमन डेटाको कुनै पनि मात्रा ह्यान्डल गर्न स्वचालित रूपमा मापन गर्दछ;
• AWS Lambda - एक सेवा जसले तपाईंलाई जगेडा वा सर्भर सेटअप बिना कोड चलाउन अनुमति दिन्छ। सबै कम्प्युटिङ पावर प्रत्येक कलको लागि स्वचालित रूपमा मापन गरिएको छ;
• अमेजन डायनामोडिबी - कुनै पनि स्केलमा चल्दा १० मिलिसेकेन्ड भन्दा कमको विलम्बता प्रदान गर्ने कुञ्जी-मान जोडी र कागजातहरूको डाटाबेस। DynamoDB प्रयोग गर्दा, तपाईंले कुनै पनि सर्भरहरू प्रावधान, प्याच वा व्यवस्थापन गर्न आवश्यक पर्दैन। DynamoDB स्वचालित रूपमा उपलब्ध स्रोतहरूको मात्रा समायोजन गर्न र उच्च प्रदर्शन कायम राख्न तालिकाहरू मापन गर्दछ। कुनै प्रणाली प्रशासन आवश्यक छैन;
• अमेजन एसएनएस - प्रकाशक-सदस्यता (पब/सब) मोडेल प्रयोग गरेर सन्देशहरू पठाउनको लागि पूर्ण रूपमा व्यवस्थित सेवा, जसको साथ तपाईं माइक्रोसेवाहरू, वितरण प्रणालीहरू र सर्भररहित अनुप्रयोगहरू अलग गर्न सक्नुहुन्छ। मोबाइल पुश सूचनाहरू, एसएमएस सन्देशहरू र इमेलहरू मार्फत अन्त प्रयोगकर्ताहरूलाई जानकारी पठाउन 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 मिलियन प्रश्नहरु। र प्रति महिना एक लाख Lambda समारोह कल।

म्यानुअल प्रणाली तैनाती

Kinesis डाटा स्ट्रिमहरू सेटअप गर्दै

Kinesis डेटा स्ट्रिम सेवामा जाऔं र दुईवटा नयाँ स्ट्रिमहरू सिर्जना गरौं, प्रत्येकको लागि एक शार्ड।

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

तपाईंको स्ट्रिममा जति धेरै शार्डहरू, यसको थ्रुपुट त्यति नै ठूलो हुन्छ। सिद्धान्तमा, यसरी प्रवाहहरू मापन गरिन्छ - शार्डहरू थपेर। तर तपाईंसँग जति धेरै शार्डहरू छन्, उति उच्च मूल्य। प्रत्येक शार्डको लागत १.५ सेन्ट प्रति घण्टा र थप १.४ सेन्ट प्रत्येक मिलियन PUT पेलोड एकाइहरूको लागि।

नामको साथ नयाँ स्ट्रिम सिर्जना गरौं airline_tickets, 1 शर्ड उनको लागि पर्याप्त हुनेछ:

अब नामको साथ अर्को थ्रेड सिर्जना गरौं विशेष_स्ट्रिम:

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

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

महत्त्वपूर्ण नोट: उदाहरणका लागि, तपाईंले छवि प्रयोग गर्नुपर्छ - Amazon Linux AMI 2018.03.0, यसमा छिटो Kinesis एजेन्ट सुरू गर्न कम सेटिङ्हरू छन्।

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 डाटा एनालिटिक्समा एउटा नयाँ अनुप्रयोग सिर्जना गर्नुहोस्:

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

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

यो सस्तो सेवा होइन - ०.११ USD प्रति घण्टा काम, त्यसैले तपाईंले यसलाई सावधानीपूर्वक प्रयोग गर्नुपर्छ र समाप्त भएपछि यसलाई मेटाउनुपर्छ।

अनुप्रयोगलाई डाटा स्रोतमा जडान गरौं:

हामीले जडान गर्न लागेको स्ट्रिम चयन गर्नुहोस् (airline_tickets):

अर्को, तपाईंले नयाँ 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 स्टेटमेन्ट प्रयोग गरेर तालिकाहरूसँग काम गर्नुहुन्छ। Amazon Kinesis डेटा एनालिटिक्समा, तपाईं स्ट्रिमहरू (स्ट्रिमहरू) र पम्पहरू (PUMPs) सँग काम गर्नुहुन्छ — एक स्ट्रिमबाट अर्को स्ट्रिममा एप्लिकेसनमा डेटा घुसाउने निरन्तर अनुरोधहरू।

माथि प्रस्तुत गरिएको SQL क्वेरीले पाँच हजार रूबलभन्दा कम लागतमा एरोफ्लोट टिकटहरूको खोजी गर्दछ। यी सर्तहरू पूरा गर्ने सबै रेकर्डहरू DESTINATION_SQL_STREAM स्ट्रिममा राखिनेछ।

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

सबै हेरफेर को नतिजा तल चित्र जस्तै केहि हुनुपर्छ:

SNS विषय सिर्जना र सदस्यता लिँदै

साधारण सूचना सेवामा जानुहोस् र त्यहाँ एयरलाइन्स नामको साथ नयाँ विषय सिर्जना गर्नुहोस्:

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

DynamoDB मा तालिका सिर्जना गर्नुहोस्

तिनीहरूको airline_tickets स्ट्रिमबाट कच्चा डाटा भण्डारण गर्न, DynamoDB मा समान नामको तालिका सिर्जना गरौं। हामी प्राथमिक कुञ्जीको रूपमा record_id प्रयोग गर्नेछौं:

Lambda प्रकार्य कलेक्टर सिर्जना गर्दै

कलेक्टर नामक lambda प्रकार्य सिर्जना गरौं, जसको कार्य airline_tickets स्ट्रिममा पोल गर्ने हो र, यदि त्यहाँ नयाँ रेकर्डहरू भेटिएमा, यी रेकर्डहरूलाई DynamoDB तालिकामा घुसाउनुहोस्। जाहिर छ, पूर्वनिर्धारित अधिकारको अतिरिक्त, यो ल्याम्ब्डाले किनेसिस डाटा स्ट्रिममा पढ्ने पहुँच र DynamoDB मा लेख्न पहुँच गरेको हुनुपर्छ।

कलेक्टर lambda प्रकार्यको लागि IAM भूमिका सिर्जना गर्दै
पहिले, Lambda-TicketsProcessingRole नामक lambda को लागी नयाँ IAM भूमिका सिर्जना गरौं:

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

यो ल्याम्ब्डा किनेसिसबाट ट्रिगरद्वारा सुरु गरिनु पर्छ जब नयाँ प्रविष्टिहरू airline_stream मा प्रवेश गर्छन्, त्यसैले हामीले नयाँ ट्रिगर थप्न आवश्यक छ:

कोड घुसाउन र lambda बचत गर्न बाँकी छ।

"""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 लाई सूचना पठाउनेछ, त्यस्तै तरिकाले सिर्जना गरिएको छ। तसर्थ, यो लाम्ब्डासँग किनेसिसबाट पढ्न र दिइएको SNS शीर्षकमा सन्देशहरू पठाउन पहुँच हुनुपर्दछ, जुन SNS सेवाद्वारा यस विषयका सबै सदस्यहरूलाई पठाइनेछ (इमेल, एसएमएस, आदि)।

IAM भूमिका सिर्जना गर्दै
पहिले, हामी यो lambda को लागि IAM भूमिका Lambda-KinesisAlarm सिर्जना गर्छौं, र त्यसपछि यो भूमिका alarm_notifier lambda सिर्जना गरिँदै छ:

यो lambda ले special_stream मा प्रवेश गर्नको लागि नयाँ रेकर्डहरूको लागि ट्रिगरमा काम गर्नुपर्छ, त्यसैले हामीले कलेक्टर lambda को लागि गरे जस्तै तपाईंले ट्रिगर कन्फिगर गर्न आवश्यक छ।

यो lambda कन्फिगर गर्न सजिलो बनाउनको लागि, एउटा नयाँ वातावरणीय चर परिचय गरौं - TOPIC_ARN, जहाँ हामीले एयरलाइन्स विषयको ANR (Amazon Recourse Names) राख्छौं:

र lambda कोड घुसाउनुहोस्, यो कुनै पनि जटिल छैन:

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 कोडबाट डिप्लोइ गर्नुहोस्

आवश्यक तयारी

ट्राफर्म कोड बाट पूर्वाधार तैनाती को लागी एक धेरै सुविधाजनक खुला स्रोत उपकरण हो। यसको आफ्नै सिन्ट्याक्स छ जुन सिक्न सजिलो छ र कसरी र के प्रयोग गर्ने भन्ने धेरै उदाहरणहरू छन्। एटम सम्पादक वा भिजुअल स्टुडियो कोडसँग धेरै उपयोगी प्लगइनहरू छन् जसले Terraform सँग काम गर्न सजिलो बनाउँदछ।

तपाईं वितरण डाउनलोड गर्न सक्नुहुन्छ यहाँ देखि। सबै Terraform क्षमताहरूको विस्तृत विश्लेषण यस लेखको दायराभन्दा बाहिर छ, त्यसैले हामी मुख्य बुँदाहरूमा सीमित गर्नेछौं।

कसरी सुरू गर्ने

परियोजनाको पूर्ण कोड हो मेरो भण्डारमा। हामी आफैंमा भण्डार क्लोन गर्छौं। सुरु गर्नु अघि, तपाईंले AWS CLI स्थापित र कन्फिगर गरिएको छ भनी सुनिश्चित गर्न आवश्यक छ, किनभने... Terraform ले ~/.aws/credentials फाइलमा प्रमाणहरू खोज्नेछ।

टेराफर्मले हाल क्लाउडमा हाम्रो लागि के सिर्जना गरिरहेको छ भनेर हेर्नको लागि सम्पूर्ण पूर्वाधारहरू प्रयोग गर्नु अघि योजना आदेश चलाउनु राम्रो अभ्यास हो:

terraform.exe plan

तपाईंलाई सूचनाहरू पठाउनको लागि फोन नम्बर प्रविष्ट गर्न प्रेरित गरिनेछ। यो चरणमा प्रवेश गर्न आवश्यक छैन।

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

terraform.exe apply

यो आदेश पठाएपछि, तपाईंलाई फेरि फोन नम्बर प्रविष्ट गर्न सोधिनेछ; वास्तवमा कार्यहरू प्रदर्शन गर्ने बारे प्रश्न देखाउँदा "हो" डायल गर्नुहोस्। यसले तपाईंलाई सम्पूर्ण पूर्वाधार सेटअप गर्न, EC2 को सबै आवश्यक कन्फिगरेसनहरू पूरा गर्न, lambda प्रकार्यहरू प्रयोग गर्न, आदि गर्न अनुमति दिन्छ।

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

अनुप्रयोग सुरु गर्नुहोस्:

यसपछि, तपाईंले ड्रप-डाउन सूचीबाट चयन गरेर इन-एप्लिकेशन स्ट्रिम नाम स्पष्ट रूपमा सेट गर्नुपर्छ:

अब सबै जान तयार छ।

आवेदन परीक्षण गर्दै

तपाईले प्रणालीलाई म्यानुअल रूपमा वा टेराफर्म कोड मार्फत कसरी डिप्लोय गर्नुभयो, यसले उस्तै काम गर्नेछ।

हामी SSH मार्फत EC2 भर्चुअल मेसिनमा लग इन गर्छौं जहाँ Kinesis Agent स्थापना गरिएको छ र api_caller.py स्क्रिप्ट चलाउनुहोस्।

sudo ./api_caller.py TOKEN

तपाईले गर्नुपर्ने भनेको तपाईको नम्बरमा एसएमएसको लागि पर्खनु हो:

एसएमएस - लगभग 1 मिनेटमा फोनमा सन्देश आउँछ:

DynamoDB डाटाबेसमा पछिल्ला, थप विस्तृत विश्लेषणका लागि रेकर्डहरू सुरक्षित गरियो कि भएन भनी हेर्न बाँकी छ। airline_tickets तालिकाले लगभग निम्न डेटा समावेश गर्दछ:

निष्कर्षमा

कामको क्रममा, एक अनलाइन डाटा प्रोसेसिंग प्रणाली Amazon Kinesis मा आधारित बनाइएको थियो। किनेसिस एजेन्ट प्रयोग गर्नका लागि विकल्पहरू Kinesis डाटा स्ट्रिमहरू र वास्तविक-समय एनालिटिक्स Kinesis एनालिटिक्स SQL ​​आदेशहरू प्रयोग गरेर, साथै अन्य AWS सेवाहरूसँग Amazon Kinesis को अन्तरक्रियालाई विचार गरियो।

हामीले माथिको प्रणालीलाई दुई तरिकामा डिप्लोय गर्यौं: एउटा लामो म्यानुअल र टेराफर्म कोडबाट द्रुत।

सबै परियोजना स्रोत कोड उपलब्ध छ मेरो GitHub भण्डारमा, म सुझाव दिन्छु कि तपाई आफैलाई यससँग परिचित हुनुहोस्।

म लेखमा छलफल गर्न पाउँदा खुसी छु, म तपाईंको टिप्पणीहरूको लागि तत्पर छु। रचनात्मक आलोचनाको अपेक्षा गर्दछु।

म तपाईंलाई सफलताको कामना गर्दछु!

स्रोत: www.habr.com