Amazon Kinesis અને સર્વરલેસ સરળતા સાથે Aviasales API એકીકરણ

હે હબર!

શું તમને ઉડતા એરોપ્લેન ગમે છે? મને તે ગમે છે, પરંતુ સ્વ-અલગતા દરમિયાન મને એક જાણીતા સંસાધન - Aviasales માંથી એર ટિકિટ પરના ડેટાનું વિશ્લેષણ કરવામાં પણ પ્રેમ થયો.

આજે અમે Amazon Kinesis ના કાર્યનું વિશ્લેષણ કરીશું, રીઅલ-ટાઇમ એનાલિટિક્સ સાથે સ્ટ્રીમિંગ સિસ્ટમ બનાવીશું, Amazon DynamoDB NoSQL ડેટાબેસને મુખ્ય ડેટા સ્ટોરેજ તરીકે ઇન્સ્ટોલ કરીશું અને રસપ્રદ ટિકિટો માટે SMS સૂચનાઓ સેટ કરીશું.

બધી વિગતો કટ હેઠળ છે! જાઓ!

પરિચય

ઉદાહરણ તરીકે, અમને ઍક્સેસની જરૂર છે Aviasales API. તેની ઍક્સેસ વિના મૂલ્યે અને પ્રતિબંધો વિના પ્રદાન કરવામાં આવે છે; તમારે ફક્ત ડેટાને ઍક્સેસ કરવા માટે તમારું API ટોકન પ્રાપ્ત કરવા માટે "વિકાસકર્તા" વિભાગમાં નોંધણી કરવાની જરૂર છે.

આ લેખનો મુખ્ય હેતુ AWS માં માહિતી સ્ટ્રીમિંગના ઉપયોગની સામાન્ય સમજ આપવાનો છે; અમે ધ્યાનમાં લઈએ છીએ કે ઉપયોગમાં લેવાયેલ API દ્વારા પરત કરવામાં આવેલ ડેટા સખત રીતે અપ-ટુ-ડેટ નથી અને તે કેશમાંથી પ્રસારિત થાય છે, જે છેલ્લા 48 કલાકથી Aviasales.ru અને Jetradar.com સાઇટના વપરાશકર્તાઓ દ્વારા શોધના આધારે રચના કરવામાં આવી છે.

કાઇનેસિસ-એજન્ટ, પ્રોડ્યુસિંગ મશીન પર ઇન્સ્ટોલ કરેલું, API દ્વારા પ્રાપ્ત થાય છે, તે આપમેળે કાઇનેસિસ ડેટા એનાલિટિક્સ દ્વારા ઇચ્છિત સ્ટ્રીમમાં ડેટાને પાર્સ કરશે અને ટ્રાન્સમિટ કરશે. આ સ્ટ્રીમનું રો વર્ઝન સ્ટોર પર સીધું લખવામાં આવશે. DynamoDB માં તૈનાત કરાયેલ કાચો ડેટા સ્ટોરેજ BI ટૂલ્સ, જેમ કે AWS ક્વિક સાઇટ દ્વારા ટિકિટ વિશ્લેષણ માટે પરવાનગી આપશે.

અમે સમગ્ર ઈન્ફ્રાસ્ટ્રક્ચરને જમાવવા માટે બે વિકલ્પો પર વિચાર કરીશું:

• મેન્યુઅલ - AWS મેનેજમેન્ટ કન્સોલ દ્વારા;
• ટેરાફોર્મ કોડમાંથી ઇન્ફ્રાસ્ટ્રક્ચર આળસુ ઓટોમેટર્સ માટે છે;

વિકસિત સિસ્ટમનું આર્કિટેક્ચર

વપરાયેલ ઘટકો:

• Aviasales API — આ API દ્વારા પરત કરવામાં આવેલ ડેટાનો ઉપયોગ પછીના તમામ કાર્ય માટે કરવામાં આવશે;
• EC2 નિર્માતા ઉદાહરણ — ક્લાઉડમાં નિયમિત વર્ચ્યુઅલ મશીન કે જેના પર ઇનપુટ ડેટા સ્ટ્રીમ જનરેટ થશે:
  • કિનેસિસ એજન્ટ મશીન પર સ્થાનિક રીતે ઇન્સ્ટોલ કરેલી જાવા એપ્લિકેશન છે જે કાઇનેસિસ (કાઇનેસિસ ડેટા સ્ટ્રીમ્સ અથવા કિનેસિસ ફાયરહોઝ) ને ડેટા એકત્રિત કરવા અને મોકલવાની સરળ રીત પ્રદાન કરે છે. એજન્ટ નિર્દિષ્ટ ડિરેક્ટરીઓમાં ફાઇલોના સેટનું સતત નિરીક્ષણ કરે છે અને કિનેસિસને નવો ડેટા મોકલે છે;
  • API કોલર સ્ક્રિપ્ટ — એક પાયથોન સ્ક્રિપ્ટ કે જે API ને વિનંતી કરે છે અને પ્રતિભાવને ફોલ્ડરમાં મૂકે છે જેનું કાઈનેસિસ એજન્ટ દ્વારા નિરીક્ષણ કરવામાં આવે છે;
• કાઇનેસિસ ડેટા સ્ટ્રીમ્સ - વિશાળ સ્કેલિંગ ક્ષમતાઓ સાથે રીઅલ-ટાઇમ ડેટા સ્ટ્રીમિંગ સેવા;
• કિનેસિસ એનાલિટિક્સ એક સર્વરલેસ સેવા છે જે વાસ્તવિક સમયમાં સ્ટ્રીમિંગ ડેટાના વિશ્લેષણને સરળ બનાવે છે. Amazon Kinesis Data Analytics એપ્લિકેશન સંસાધનોને ગોઠવે છે અને આવનારા ડેટાના કોઈપણ વોલ્યુમને હેન્ડલ કરવા માટે આપમેળે સ્કેલ કરે છે;
• AWS લેમ્બડા — એવી સેવા કે જે તમને સર્વરનો બેકઅપ લીધા વિના અથવા સેટઅપ કર્યા વિના કોડ ચલાવવાની મંજૂરી આપે છે. દરેક કૉલ માટે તમામ કમ્પ્યુટિંગ પાવર આપમેળે માપવામાં આવે છે;
• એમેઝોન ડાયનામોડીબી - કી-વેલ્યુ જોડી અને દસ્તાવેજોનો ડેટાબેઝ જે કોઈપણ સ્કેલ પર ચાલતી વખતે 10 મિલીસેકંડથી ઓછી લેટન્સી પ્રદાન કરે છે. DynamoDB નો ઉપયોગ કરતી વખતે, તમારે કોઈપણ સર્વરને જોગવાઈ, પેચ અથવા મેનેજ કરવાની જરૂર નથી. DynamoDB ઉપલબ્ધ સંસાધનોની માત્રાને સમાયોજિત કરવા અને ઉચ્ચ પ્રદર્શન જાળવવા માટે આપમેળે કોષ્ટકોને સ્કેલ કરે છે. કોઈ સિસ્ટમ વહીવટ જરૂરી નથી;
• એમેઝોન SNS - પ્રકાશક-સબ્સ્ક્રાઇબર (પબ/સબ) મૉડલનો ઉપયોગ કરીને સંદેશા મોકલવા માટેની સંપૂર્ણ વ્યવસ્થાપિત સેવા, જેની મદદથી તમે માઇક્રોસર્વિસિસ, ડિસ્ટ્રિબ્યુટેડ સિસ્ટમ્સ અને સર્વરલેસ એપ્લિકેશન્સને અલગ કરી શકો છો. SNS નો ઉપયોગ મોબાઇલ પુશ સૂચનાઓ, SMS સંદેશાઓ અને ઇમેઇલ્સ દ્વારા અંતિમ વપરાશકર્તાઓને માહિતી મોકલવા માટે કરી શકાય છે.

પ્રારંભિક તાલીમ

ડેટા ફ્લોનું અનુકરણ કરવા માટે, મેં 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 મિલિયન ક્વેરીઝ. અને દર મહિને એક મિલિયન લેમ્બડા ફંક્શન કૉલ્સ.

મેન્યુઅલ સિસ્ટમ જમાવટ

કાઇનેસિસ ડેટા સ્ટ્રીમ્સ સેટ કરી રહ્યા છીએ

ચાલો કાઈનેસિસ ડેટા સ્ટ્રીમ્સ સેવા પર જઈએ અને બે નવી સ્ટ્રીમ બનાવીએ, દરેક માટે એક શાર્ડ.

શાર્ડ શું છે?
શાર્ડ એ એમેઝોન કિનેસિસ સ્ટ્રીમનું મૂળભૂત ડેટા ટ્રાન્સફર યુનિટ છે. એક સેગમેન્ટ 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, તેમાં કાઈનેસિસ એજન્ટને ઝડપથી લોન્ચ કરવા માટે ઓછા સેટિંગ્સ છે.

EC2 સેવા પર જાઓ, નવું વર્ચ્યુઅલ મશીન બનાવો, t2.micro પ્રકાર સાથે ઇચ્છિત AMI પસંદ કરો, જે ફ્રી ટિયરમાં શામેલ છે:

નવી બનાવેલ વર્ચ્યુઅલ મશીન કાઇનેસિસ સેવા સાથે ક્રિયાપ્રતિક્રિયા કરવા સક્ષમ બને તે માટે, તેને આમ કરવાના અધિકારો આપવા આવશ્યક છે. આ કરવાની શ્રેષ્ઠ રીત એ છે કે IAM ભૂમિકા સોંપવી. તેથી, સ્ટેપ 3: ઇન્સ્ટન્સ ડિટેલ્સ રૂપરેખાંકિત કરો સ્ક્રીન પર, તમારે પસંદ કરવું જોઈએ નવી IAM ભૂમિકા બનાવો:

EC2 માટે IAM ભૂમિકા બનાવવી

ખુલતી વિંડોમાં, પસંદ કરો કે અમે EC2 માટે નવી ભૂમિકા બનાવી રહ્યા છીએ અને પરવાનગી વિભાગ પર જાઓ:

તાલીમના ઉદાહરણનો ઉપયોગ કરીને, અમારે સંસાધન અધિકારોના દાણાદાર રૂપરેખાંકનની બધી જટિલતાઓમાં જવાની જરૂર નથી, તેથી અમે Amazon દ્વારા પૂર્વ-રૂપરેખાંકિત નીતિઓ પસંદ કરીશું: AmazonKinesisFullAccess અને CloudWatchFullAccess.

ચાલો આ ભૂમિકા માટે કેટલાક અર્થપૂર્ણ નામ આપીએ, ઉદાહરણ તરીકે: EC2-KinesisStreams-FullAccess. પરિણામ નીચેના ચિત્રમાં બતાવ્યા પ્રમાણે જ હોવું જોઈએ:

આ નવી ભૂમિકા બનાવ્યા પછી, તેને બનાવેલ વર્ચ્યુઅલ મશીન ઉદાહરણ સાથે જોડવાનું ભૂલશો નહીં:

અમે આ સ્ક્રીન પર બીજું કંઈપણ બદલતા નથી અને આગળની વિંડોઝ પર આગળ વધીએ છીએ.

હાર્ડ ડ્રાઈવ સેટિંગ્સને ડિફૉલ્ટ તરીકે છોડી શકાય છે, સાથે સાથે ટૅગ્સ (જોકે ટૅગ્સનો ઉપયોગ કરવો એ સારી પ્રથા છે, ઓછામાં ઓછું ઉદાહરણને નામ આપો અને પર્યાવરણ સૂચવો).

હવે અમે સ્ટેપ 6 પર છીએ: સિક્યોરિટી ગ્રૂપ ટેબને ગોઠવો, જ્યાં તમારે એક નવું બનાવવાની જરૂર છે અથવા તમારું હાલનું સિક્યુરિટી ગ્રૂપ સ્પષ્ટ કરવું પડશે, જે તમને ssh (પોર્ટ 22) દ્વારા દાખલા સાથે કનેક્ટ કરવાની મંજૂરી આપે છે. ત્યાં સ્ત્રોત -> My IP પસંદ કરો અને તમે દાખલો લોંચ કરી શકો છો.

જલદી તે ચાલુ સ્થિતિ પર સ્વિચ કરે છે, તમે તેને 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

ચાલો 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 એક્સ્ટેંશન સાથે ફાઇલોનું નિરીક્ષણ કરશે, તેમને પાર્સ કરશે અને એરલાઇન_ટિકેટ સ્ટ્રીમમાં સ્થાનાંતરિત કરશે.

અમે સેવા પુનઃપ્રારંભ કરીએ છીએ અને ખાતરી કરીએ છીએ કે તે ચાલુ છે અને ચાલુ છે:

sudo service aws-kinesis-agent restart

હવે ચાલો Python સ્ક્રિપ્ટ ડાઉનલોડ કરીએ જે 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

અને અમે એરલાઇન_ટિકિટ ડેટા સ્ટ્રીમમાં એજન્ટ લોગમાં અને મોનિટરિંગ ટેબ પર કામના પરિણામને જોઈએ છીએ:

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

જેમ તમે જોઈ શકો છો, બધું કામ કરે છે અને કિનેસિસ એજન્ટ સફળતાપૂર્વક સ્ટ્રીમમાં ડેટા મોકલે છે. હવે ચાલો ગ્રાહકને ગોઠવીએ.

કાઇનેસિસ ડેટા એનાલિટિક્સ સેટ કરી રહ્યું છે

ચાલો સમગ્ર સિસ્ટમના કેન્દ્રિય ઘટક તરફ આગળ વધીએ - Kinesis_analytics_airlines_app નામની Kinesis Data Analytics માં એક નવી એપ્લિકેશન બનાવો:

કિનેસિસ ડેટા એનાલિટિક્સ તમને એસક્યુએલ ભાષાનો ઉપયોગ કરીને કિનેસિસ સ્ટ્રીમ્સમાંથી રીઅલ-ટાઇમ ડેટા એનાલિટિક્સ કરવા દે છે. તે સંપૂર્ણ ઓટોસ્કેલિંગ સેવા છે (કાઈનેસિસ સ્ટ્રીમ્સથી વિપરીત) જે:

1. તમને સ્રોત ડેટાની વિનંતીઓના આધારે નવી સ્ટ્રીમ્સ (આઉટપુટ સ્ટ્રીમ) બનાવવાની મંજૂરી આપે છે;
2. એપ્લીકેશન ચાલી રહી હતી ત્યારે થયેલી ભૂલો સાથેનો સ્ટ્રીમ પૂરો પાડે છે (ભૂલ સ્ટ્રીમ);
3. ઇનપુટ ડેટા સ્કીમ આપમેળે નક્કી કરી શકે છે (જો જરૂરી હોય તો તેને મેન્યુઅલી ફરીથી વ્યાખ્યાયિત કરી શકાય છે).

આ સસ્તી સેવા નથી - કામના કલાક દીઠ 0.11 USD, તેથી તમારે તેનો કાળજીપૂર્વક ઉપયોગ કરવો જોઈએ અને જ્યારે તમે પૂર્ણ કરી લો ત્યારે તેને કાઢી નાખો.

ચાલો એપ્લિકેશનને ડેટા સ્ત્રોત સાથે કનેક્ટ કરીએ:

અમે જે સ્ટ્રીમ સાથે કનેક્ટ થવા જઈ રહ્યા છીએ તે પસંદ કરો (એરલાઇન_ટિકિટ):

આગળ, તમારે નવી IAM ભૂમિકા જોડવાની જરૂર છે જેથી કરીને એપ્લિકેશન સ્ટ્રીમમાંથી વાંચી શકે અને સ્ટ્રીમમાં લખી શકે. આ કરવા માટે, ઍક્સેસ પરવાનગી બ્લોકમાં કંઈપણ બદલવું નહીં તે પૂરતું છે:

ચાલો હવે સ્ટ્રીમમાં ડેટા સ્કીમા શોધવાની વિનંતી કરીએ; આ કરવા માટે, "ડિસ્કવર સ્કીમા" બટન પર ક્લિક કરો. પરિણામે, IAM ભૂમિકા અપડેટ કરવામાં આવશે (એક નવું બનાવવામાં આવશે) અને સ્ટ્રીમમાં પહેલેથી જ આવી ગયેલા ડેટામાંથી સ્કીમા શોધ શરૂ કરવામાં આવશે:

હવે તમારે 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 Data Analytics માં, તમે સ્ટ્રીમ્સ (STREAMs) અને પમ્પ્સ (PUMPs) સાથે કામ કરો છો—સતત ઇન્સર્ટ વિનંતીઓ જે એપ્લિકેશનમાં એક સ્ટ્રીમમાંથી ડેટાને બીજી સ્ટ્રીમમાં દાખલ કરે છે.

ઉપર પ્રસ્તુત SQL ક્વેરી પાંચ હજાર રુબેલ્સથી ઓછી કિંમતે એરોફ્લોટ ટિકિટો માટે શોધ કરે છે. આ શરતોને પૂર્ણ કરતા તમામ રેકોર્ડ્સ DESTINATION_SQL_STREAM સ્ટ્રીમમાં મૂકવામાં આવશે.

ગંતવ્ય બ્લોકમાં, સ્પેશિયલ_સ્ટ્રીમ સ્ટ્રીમ પસંદ કરો અને ઇન-એપ્લિકેશન સ્ટ્રીમ નામ DESTINATION_SQL_STREAM ડ્રોપ-ડાઉન સૂચિમાં:

તમામ મેનિપ્યુલેશન્સનું પરિણામ નીચેના ચિત્ર જેવું જ કંઈક હોવું જોઈએ:

SNS વિષય બનાવવો અને સબ્સ્ક્રાઇબ કરવું

સિમ્પલ નોટિફિકેશન સર્વિસ પર જાઓ અને ત્યાં એરલાઈન્સ નામ સાથે નવો વિષય બનાવો:

આ વિષય પર સબ્સ્ક્રાઇબ કરો અને મોબાઇલ ફોન નંબર સૂચવો કે જેના પર SMS સૂચનાઓ મોકલવામાં આવશે:

DynamoDB માં ટેબલ બનાવો

તેમની એરલાઇન_ટિકિટ સ્ટ્રીમમાંથી કાચો ડેટા સ્ટોર કરવા માટે, ચાલો તે જ નામ સાથે ડાયનામોડીબીમાં એક ટેબલ બનાવીએ. અમે પ્રાથમિક કી તરીકે record_id નો ઉપયોગ કરીશું:

લેમ્બડા ફંક્શન કલેક્ટર બનાવવું

ચાલો કલેક્ટર નામનું લેમ્બડા ફંક્શન બનાવીએ, જેનું કાર્ય એરલાઇન_ટિકેટ સ્ટ્રીમનું મતદાન કરવાનું રહેશે અને, જો ત્યાં નવા રેકોર્ડ જોવા મળે, તો આ રેકોર્ડ્સને ડાયનામોડીબી ટેબલમાં દાખલ કરો. દેખીતી રીતે, ડિફૉલ્ટ રાઇટ્સ ઉપરાંત, આ લેમ્બડા પાસે કાઇનેસિસ ડેટા સ્ટ્રીમની રીડ એક્સેસ અને ડાયનામોડીબી પર લખવાની એક્સેસ હોવી જોઈએ.

કલેક્ટર લેમ્બડા ફંક્શન માટે IAM ભૂમિકા બનાવવી
પ્રથમ, ચાલો લેમ્બડા માટે એક નવી IAM ભૂમિકા બનાવીએ જેનું નામ છે Lambda-TicketsProcessingRole:

પરીક્ષણ ઉદાહરણ માટે, નીચેની ચિત્રમાં બતાવ્યા પ્રમાણે, પૂર્વ-રૂપરેખાંકિત 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()

લેમ્બડા ફંક્શન નોટિફાયર બનાવવું

બીજું લેમ્બડા ફંક્શન, જે બીજા સ્ટ્રીમ (સ્પેશિયલ_સ્ટ્રીમ) ને મોનિટર કરશે અને SNS ને સૂચના મોકલશે, તે જ રીતે બનાવવામાં આવ્યું છે. તેથી, આ લેમ્બડા પાસે કાઇનેસિસમાંથી વાંચવા અને આપેલા SNS વિષય પર સંદેશા મોકલવાની ઍક્સેસ હોવી આવશ્યક છે, જે પછી SNS સેવા દ્વારા આ વિષયના તમામ સબ્સ્ક્રાઇબર્સને મોકલવામાં આવશે (ઇમેઇલ, SMS, વગેરે).

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 ઇન્સ્ટોલ અને ગોઠવેલ છે, કારણ કે... Terraform ~/.aws/credentials ફાઇલમાં ઓળખપત્રો શોધશે.

ટેરાફોર્મ હાલમાં ક્લાઉડમાં આપણા માટે શું બનાવી રહ્યું છે તે જોવા માટે સમગ્ર ઇન્ફ્રાસ્ટ્રક્ચરને જમાવવા પહેલાં પ્લાન કમાન્ડને ચલાવવાની સારી પ્રથા છે:

terraform.exe plan

તમને સૂચનાઓ મોકલવા માટે ફોન નંબર દાખલ કરવા માટે સંકેત આપવામાં આવશે. આ તબક્કે તેને દાખલ કરવું જરૂરી નથી.

પ્રોગ્રામના ઓપરેશન પ્લાનનું વિશ્લેષણ કર્યા પછી, અમે સંસાધનો બનાવવાનું શરૂ કરી શકીએ છીએ:

terraform.exe apply

આ આદેશ મોકલ્યા પછી, તમને ફરીથી ફોન નંબર દાખલ કરવા માટે કહેવામાં આવશે; જ્યારે ખરેખર ક્રિયાઓ કરવા વિશે પ્રશ્ન બતાવવામાં આવે ત્યારે "હા" ડાયલ કરો. આ તમને સમગ્ર ઈન્ફ્રાસ્ટ્રક્ચર સેટ કરવા, EC2 નું તમામ જરૂરી રૂપરેખાંકન હાથ ધરવા, લેમ્બડા ફંક્શન્સ વગેરે ગોઠવવાની મંજૂરી આપશે.

ટેરાફોર્મ કોડ દ્વારા તમામ સંસાધનો સફળતાપૂર્વક બનાવવામાં આવ્યા પછી, તમારે કાઇનેસિસ એનાલિટિક્સ એપ્લિકેશનની વિગતોમાં જવાની જરૂર છે (કમનસીબે, કોડમાંથી આ કેવી રીતે કરવું તે મને મળ્યું નથી).

એપ્લિકેશન લોંચ કરો:

આ પછી, તમારે ડ્રોપ-ડાઉન સૂચિમાંથી પસંદ કરીને એપ્લિકેશનમાં સ્ટ્રીમ નામ સ્પષ્ટપણે સેટ કરવું આવશ્યક છે:

હવે બધું જવા માટે તૈયાર છે.

એપ્લિકેશન પરીક્ષણ

તમે સિસ્ટમને મેન્યુઅલી અથવા ટેરાફોર્મ કોડ દ્વારા કેવી રીતે જમાવ્યું છે તે ધ્યાનમાં લીધા વિના, તે તે જ કાર્ય કરશે.

અમે SSH મારફતે EC2 વર્ચ્યુઅલ મશીનમાં લૉગ ઇન કરીએ છીએ જ્યાં કાઇનેસિસ એજન્ટ ઇન્સ્ટોલ કરેલું છે અને api_caller.py સ્ક્રિપ્ટ ચલાવીએ છીએ

sudo ./api_caller.py TOKEN

તમારે ફક્ત તમારા નંબર પર એક SMSની રાહ જોવાની છે:

SMS - લગભગ 1 મિનિટમાં તમારા ફોન પર એક સંદેશ આવે છે:

તે પછીના, વધુ વિગતવાર વિશ્લેષણ માટે ડાયનામોડીબી ડેટાબેઝમાં રેકોર્ડ્સ સાચવવામાં આવ્યા હતા કે કેમ તે જોવાનું બાકી છે. એરલાઇન_ટિકિટ કોષ્ટકમાં લગભગ નીચેનો ડેટા છે:

નિષ્કર્ષ

કરવામાં આવેલ કાર્ય દરમિયાન, એમેઝોન કિનેસિસ પર આધારિત ઓનલાઈન ડેટા પ્રોસેસિંગ સિસ્ટમ બનાવવામાં આવી હતી. કિનેસિસ ડેટા સ્ટ્રીમ્સ અને એસક્યુએલ આદેશોનો ઉપયોગ કરીને રીઅલ-ટાઇમ એનાલિટિક્સ કિનેસિસ એનાલિટિક્સ સાથે જોડાણમાં કાઇનેસિસ એજન્ટનો ઉપયોગ કરવા માટેના વિકલ્પો તેમજ અન્ય AWS સેવાઓ સાથે એમેઝોન કિનેસિસની ક્રિયાપ્રતિક્રિયાને ધ્યાનમાં લેવામાં આવી હતી.

અમે ઉપરોક્ત સિસ્ટમને બે રીતે તૈનાત કરી છે: એક તેના બદલે લાંબી મેન્યુઅલ અને ટેરાફોર્મ કોડમાંથી ઝડપી.

બધા પ્રોજેક્ટ સ્ત્રોત કોડ ઉપલબ્ધ છે મારા GitHub રીપોઝીટરીમાં, હું તમને તેની સાથે પરિચિત થવાનું સૂચન કરું છું.

હું લેખની ચર્ચા કરવામાં ખુશ છું, હું તમારી ટિપ્પણીઓની રાહ જોઉં છું. હું રચનાત્મક ટીકાની આશા રાખું છું.

હું તમને સફળતાની ઇચ્છા કરું છું!

સોર્સ: www.habr.com