ការរួមបញ្ចូល Aviasales API ជាមួយ Amazon Kinesis និងភាពសាមញ្ញគ្មានម៉ាស៊ីនមេ

ហេហេ!

តើអ្នកចូលចិត្តយន្តហោះហោះទេ? ខ្ញុំស្រលាញ់វា ប៉ុន្តែក្នុងអំឡុងពេលឯកោខ្លួនឯង ខ្ញុំក៏ធ្លាក់ក្នុងអន្លង់ស្នេហ៍ជាមួយនឹងការវិភាគទិន្នន័យអំពីសំបុត្រយន្តហោះពីធនធានដ៏ល្បីមួយគឺ Aviasales ។

ថ្ងៃនេះយើងនឹងវិភាគការងាររបស់ Amazon Kinesis បង្កើតប្រព័ន្ធស្ទ្រីមជាមួយការវិភាគពេលវេលាជាក់ស្តែង ដំឡើងមូលដ្ឋានទិន្នន័យ Amazon DynamoDB NoSQL ជាកន្លែងផ្ទុកទិន្នន័យសំខាន់ និងរៀបចំការជូនដំណឹង SMS សម្រាប់សំបុត្រដែលគួរឱ្យចាប់អារម្មណ៍។

ព័ត៌មានលម្អិតគឺស្ថិតនៅក្រោមការកាត់! ទៅ!

សេចក្តីណែនាំ

ជាឧទាហរណ៍ យើងត្រូវការចូលប្រើ Aviasales API. ការចូលប្រើវាត្រូវបានផ្តល់ជូនដោយមិនគិតថ្លៃ និងគ្មានការរឹតបន្តឹង អ្នកគ្រាន់តែត្រូវចុះឈ្មោះនៅក្នុងផ្នែក "អ្នកអភិវឌ្ឍន៍" ដើម្បីទទួលបាននិមិត្តសញ្ញា API របស់អ្នកដើម្បីចូលប្រើទិន្នន័យ។

គោលបំណងសំខាន់នៃអត្ថបទនេះគឺដើម្បីផ្តល់ការយល់ដឹងទូទៅអំពីការប្រើប្រាស់ការផ្សាយព័ត៌មាននៅក្នុង AWS យើងពិចារណាថាទិន្នន័យដែលបានបញ្ជូនមកវិញដោយ API ដែលបានប្រើគឺមិនទាន់សម័យយ៉ាងតឹងរ៉ឹង និងត្រូវបានបញ្ជូនពីឃ្លាំងសម្ងាត់ដែលជា បង្កើតឡើងដោយផ្អែកលើការស្វែងរកដោយអ្នកប្រើប្រាស់នៃគេហទំព័រ Aviasales.ru និង Jetradar.com រយៈពេល 48 ម៉ោងចុងក្រោយ។

Kinesis-agent ដែលបានដំឡើងនៅលើម៉ាស៊ីនផលិត បានទទួលតាមរយៈ API នឹងញែក និងបញ្ជូនទិន្នន័យដោយស្វ័យប្រវត្តិទៅស្ទ្រីមដែលចង់បានតាមរយៈ Kinesis Data Analytics ។ កំណែដើមនៃស្ទ្រីមនេះនឹងត្រូវបានសរសេរដោយផ្ទាល់ទៅហាង។ ការផ្ទុកទិន្នន័យឆៅដែលបានដាក់ពង្រាយនៅក្នុង DynamoDB នឹងអនុញ្ញាតឱ្យមានការវិភាគសំបុត្រកាន់តែស៊ីជម្រៅតាមរយៈឧបករណ៍ BI ដូចជា AWS Quick Sight ជាដើម។

យើងនឹងពិចារណាជម្រើសពីរសម្រាប់ដាក់ពង្រាយហេដ្ឋារចនាសម្ព័ន្ធទាំងមូល៖

• សៀវភៅណែនាំ - តាមរយៈ AWS Management Console;
• ហេដ្ឋារចនាសម្ព័ន្ធពីកូដ Terraform គឺសម្រាប់ម៉ាស៊ីនស្វ័យប្រវត្តិខ្ជិល។

ស្ថាបត្យកម្មនៃប្រព័ន្ធអភិវឌ្ឍន៍

សមាសធាតុដែលបានប្រើ៖

• Aviasales API - ទិន្នន័យដែលបានបញ្ជូនមកវិញដោយ API នេះនឹងត្រូវបានប្រើសម្រាប់ការងារជាបន្តបន្ទាប់ទាំងអស់។
• ឧទាហរណ៍អ្នកផលិត EC2 - ម៉ាស៊ីននិម្មិតធម្មតានៅក្នុងពពកដែលស្ទ្រីមទិន្នន័យបញ្ចូលនឹងត្រូវបានបង្កើត៖
  • ភ្នាក់ងារ Kinesis គឺជាកម្មវិធី Java ដែលបានដំឡើងក្នុងមូលដ្ឋាននៅលើម៉ាស៊ីន ដែលផ្តល់នូវវិធីងាយស្រួលក្នុងការប្រមូល និងបញ្ជូនទិន្នន័យទៅកាន់ Kinesis (Kinesis Data Stream ឬ Kinesis Firehose)។ ភ្នាក់ងារត្រួតពិនិត្យជាបន្តបន្ទាប់នូវសំណុំឯកសារនៅក្នុងថតដែលបានបញ្ជាក់ ហើយបញ្ជូនទិន្នន័យថ្មីទៅ Kinesis ។
  • API Caller Script - ស្គ្រីប Python ដែលធ្វើសំណើទៅកាន់ API ហើយដាក់ការឆ្លើយតបទៅក្នុងថតឯកសារដែលត្រូវបានត្រួតពិនិត្យដោយភ្នាក់ងារ Kinesis ។
• ស្ទ្រីមទិន្នន័យ Kinesis - សេវាកម្មស្ទ្រីមទិន្នន័យតាមពេលវេលាជាក់ស្តែងជាមួយនឹងសមត្ថភាពធ្វើមាត្រដ្ឋានធំទូលាយ។
• ការវិភាគ Kinesis គឺជាសេវាកម្មគ្មានម៉ាស៊ីនមេ ដែលជួយសម្រួលដល់ការវិភាគនៃការផ្សាយទិន្នន័យក្នុងពេលវេលាជាក់ស្តែង។ Amazon Kinesis Data Analytics កំណត់រចនាសម្ព័ន្ធធនធានកម្មវិធី និងធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិ ដើម្បីគ្រប់គ្រងបរិមាណនៃទិន្នន័យចូល។
• AWS Lambda - សេវាកម្មដែលអនុញ្ញាតឱ្យអ្នកដំណើរការកូដដោយមិនចាំបាច់បម្រុងទុក ឬដំឡើងម៉ាស៊ីនមេ។ ថាមពលកុំព្យូទ័រទាំងអស់ត្រូវបានធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិសម្រាប់ការហៅនីមួយៗ។
• ក្រុមហ៊ុន Amazon DynamoDB - មូលដ្ឋានទិន្នន័យនៃគូតម្លៃ និងឯកសារដែលផ្តល់ភាពយឺតយ៉ាវតិចជាង 10 មីលីវិនាទី នៅពេលដំណើរការនៅមាត្រដ្ឋានណាមួយ។ នៅពេលប្រើ DynamoDB អ្នកមិនចាំបាច់ផ្តល់ បំណះ ឬគ្រប់គ្រងម៉ាស៊ីនមេណាមួយឡើយ។ DynamoDB ធ្វើមាត្រដ្ឋានតារាងដោយស្វ័យប្រវត្តិ ដើម្បីកែតម្រូវបរិមាណធនធានដែលមាន និងរក្សាដំណើរការខ្ពស់។ គ្មានការគ្រប់គ្រងប្រព័ន្ធត្រូវបានទាមទារ;
• ក្រុមហ៊ុន Amazon SNS - សេវាកម្មដែលគ្រប់គ្រងយ៉ាងពេញលេញសម្រាប់ការផ្ញើសារដោយប្រើគំរូអ្នកបោះពុម្ពផ្សាយ-ជាវ (Pub/Sub) ដែលអ្នកអាចញែកសេវាកម្មខ្នាតតូច ប្រព័ន្ធចែកចាយ និងកម្មវិធីដែលគ្មានម៉ាស៊ីនមេ។ 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 រួចហើយ។ ខ្ញុំចង់ទាក់ទាញការយកចិត្តទុកដាក់ជាពិសេសភ្លាមៗចំពោះការពិតដែលថា Kinesis និងការផ្ញើការជូនដំណឹងតាមរយៈសារ SMS មិនត្រូវបានរាប់បញ្ចូលក្នុងឆ្នាំ កម្រិតឥតគិតថ្លៃ (ការប្រើប្រាស់ដោយឥតគិតថ្លៃ). ប៉ុន្តែទោះបីជានេះក៏ដោយ ដោយគិតលុយពីរបីដុល្លារ វាពិតជាអាចទៅរួចក្នុងការសាងសង់ប្រព័ន្ធដែលបានស្នើឡើង ហើយលេងជាមួយវា។ ហើយជាការពិតណាស់កុំភ្លេចលុបធនធានទាំងអស់បន្ទាប់ពីពួកគេលែងត្រូវការ។

ជាសំណាងល្អ មុខងារ DynamoDb និង lambda នឹងមិនគិតថ្លៃសម្រាប់យើងទេ ប្រសិនបើយើងបំពេញតាមដែនកំណត់ឥតគិតថ្លៃប្រចាំខែរបស់យើង។ ឧទាហរណ៍សម្រាប់ DynamoDB៖ អង្គផ្ទុកទិន្នន័យ 25 GB, 25 WCU/RCU និង 100 លានសំណួរ។ និងការហៅមុខងារ lambda មួយលានដងក្នុងមួយខែ។

ការដាក់ពង្រាយប្រព័ន្ធដោយដៃ

ការដំឡើង Kinesis Data Stream

តោះទៅសេវាកម្ម Kinesis Data Stream ហើយបង្កើតស្ទ្រីមថ្មីពីរ មួយ shard សម្រាប់នីមួយៗ។

អ្វី​ទៅ​ជា shard ?
Shard គឺជាអង្គភាពផ្ទេរទិន្នន័យមូលដ្ឋាននៃស្ទ្រីម Amazon Kinesis ។ ផ្នែកមួយផ្តល់នូវការផ្ទេរទិន្នន័យបញ្ចូលក្នុងល្បឿន 1 MB/s និងការផ្ទេរទិន្នន័យលទ្ធផលក្នុងល្បឿន 2 MB/s ។ ផ្នែកមួយគាំទ្ររហូតដល់ 1000 PUT ធាតុក្នុងមួយវិនាទី។ នៅពេលបង្កើតស្ទ្រីមទិន្នន័យ អ្នកត្រូវបញ្ជាក់ចំនួនផ្នែកដែលត្រូវការ។ ឧទាហរណ៍ អ្នកអាចបង្កើតស្ទ្រីមទិន្នន័យដែលមានពីរផ្នែក។ ស្ទ្រីមទិន្នន័យនេះនឹងផ្តល់នូវការផ្ទេរទិន្នន័យបញ្ចូលក្នុងល្បឿន 2 MB/s និងការផ្ទេរទិន្នន័យលទ្ធផលក្នុងល្បឿន 4 MB/s ដែលគាំទ្ររហូតដល់ 2000 PUT records ក្នុងមួយវិនាទី។

ការ​ស្ទ្រីម​កាន់​តែ​ច្រើន​នៅ​ក្នុង​ការ​ផ្សាយ​របស់​អ្នក លំហូរ​របស់​វា​កាន់​តែ​ធំ។ ជាគោលការណ៍នេះជារបៀបដែលលំហូរត្រូវបានធ្វើមាត្រដ្ឋាន - ដោយបន្ថែមបំណែក។ ប៉ុន្តែ​បើ​អ្នក​មាន​សំបក​កាន់​តែ​ច្រើន តម្លៃ​កាន់​តែ​ខ្ពស់។ shard នីមួយៗមានតម្លៃ 1,5 សេនក្នុងមួយម៉ោង និងបន្ថែម 1.4 សេនសម្រាប់រាល់ការបញ្ជូន PUT រាប់លានគ្រឿង។

តោះបង្កើតស្ទ្រីមថ្មីជាមួយឈ្មោះ សំបុត្រយន្តហោះ, 1 shard នឹងគ្រប់គ្រាន់សម្រាប់គាត់:

ឥឡូវ​នេះ​យើង​បង្កើត​ខ្សែ​ស្រឡាយ​មួយ​ទៀត​ជាមួយ​នឹង​ឈ្មោះ ស្ទ្រីមពិសេស:

ការរៀបចំអ្នកផលិត

ដើម្បីវិភាគកិច្ចការមួយ វាគ្រប់គ្រាន់ក្នុងការប្រើឧទាហរណ៍ EC2 ធម្មតាជាអ្នកផលិតទិន្នន័យ។ វាមិនចាំបាច់ជាម៉ាស៊ីននិម្មិតដែលមានតម្លៃថ្លៃខ្លាំងនោះទេ កន្លែង t2.micro នឹងដំណើរការបានល្អ។

ចំណាំសំខាន់៖ ឧទាហរណ៍ អ្នកគួរតែប្រើរូបភាព - Amazon Linux AMI 2018.03.0 វាមានការកំណត់តិចជាងមុនសម្រាប់ការបើកដំណើរការភ្នាក់ងារ Kinesis យ៉ាងរហ័ស។

ចូលទៅកាន់សេវា EC2 បង្កើតម៉ាស៊ីននិម្មិតថ្មី ជ្រើសរើស AMI ដែលចង់បានជាមួយប្រភេទ t2.micro ដែលត្រូវបានដាក់បញ្ចូលក្នុង Free Tier៖

ដើម្បីឱ្យម៉ាស៊ីននិម្មិតដែលបានបង្កើតថ្មីអាចធ្វើអន្តរកម្មជាមួយសេវាកម្ម Kinesis វាត្រូវតែផ្តល់សិទ្ធិដើម្បីធ្វើដូច្នេះបាន។ មធ្យោបាយដ៏ល្អបំផុតដើម្បីធ្វើវាគឺកំណត់តួនាទី IAM ។ ដូច្នេះនៅលើជំហានទី 3: កំណត់រចនាសម្ព័ន្ធព័ត៌មានលម្អិតនៃអេក្រង់ អ្នកគួរតែជ្រើសរើស បង្កើតតួនាទី IAM ថ្មី។:

ការបង្កើតតួនាទី IAM សម្រាប់ EC2

នៅក្នុងបង្អួចដែលបើក សូមជ្រើសរើសថាយើងកំពុងបង្កើតតួនាទីថ្មីសម្រាប់ EC2 ហើយចូលទៅកាន់ផ្នែកការអនុញ្ញាត៖

ដោយប្រើឧទាហរណ៍នៃការបណ្តុះបណ្តាល យើងមិនចាំបាច់ចូលទៅក្នុងភាពស្មុគស្មាញទាំងអស់នៃការកំណត់រចនាសម្ព័ន្ធជាក្រឡានៃសិទ្ធិធនធាន ដូច្នេះយើងនឹងជ្រើសរើសគោលការណ៍ដែលបានកំណត់ជាមុនដោយ Amazon៖ AmazonKinesisFullAccess និង CloudWatchFullAccess ។

ចូរផ្តល់ឈ្មោះដ៏មានអត្ថន័យសម្រាប់តួនាទីនេះ ឧទាហរណ៍៖ EC2-KinesisStreams-FullAccess។ លទ្ធផលគួរតែដូចក្នុងរូបភាពខាងក្រោម៖

បន្ទាប់ពីបង្កើតតួនាទីថ្មីនេះ កុំភ្លេចភ្ជាប់វាទៅម៉ាស៊ីននិម្មិតដែលបានបង្កើត៖

យើងមិនផ្លាស់ប្តូរអ្វីផ្សេងទៀតនៅលើអេក្រង់នេះ ហើយបន្តទៅបង្អួចបន្ទាប់។

ការកំណត់ដ្រាយវ៍រឹងអាចត្រូវបានទុកជាលំនាំដើម ក៏ដូចជាស្លាក (ទោះបីជាវាជាការអនុវត្តល្អក្នុងការប្រើស្លាក យ៉ាងហោចណាស់ក៏ផ្តល់ឈ្មោះឱ្យឧទាហរណ៍ និងបង្ហាញពីបរិស្ថាន)។

ឥឡូវនេះយើងស្ថិតនៅលើជំហានទី 6៖ កំណត់រចនាសម្ព័ន្ធផ្ទាំងក្រុមសុវត្ថិភាព ដែលអ្នកត្រូវបង្កើតថ្មីមួយ ឬបញ្ជាក់ក្រុមសុវត្ថិភាពដែលមានស្រាប់របស់អ្នក ដែលអនុញ្ញាតឱ្យអ្នកភ្ជាប់តាមរយៈ ssh (ច្រក 22) ទៅកាន់ instance ។ ជ្រើសរើសប្រភព -> IP របស់ខ្ញុំនៅទីនោះ ហើយអ្នកអាចចាប់ផ្តើមឧទាហរណ៍។

ដរាបណាវាប្តូរទៅស្ថានភាពដែលកំពុងដំណើរការ អ្នកអាចព្យាយាមភ្ជាប់ទៅវាតាមរយៈ ssh ។

ដើម្បីអាចធ្វើការជាមួយ Kinesis Agent បន្ទាប់ពីភ្ជាប់ទៅម៉ាស៊ីនដោយជោគជ័យ អ្នកត្រូវតែបញ្ចូលពាក្យបញ្ជាខាងក្រោមនៅក្នុងស្ថានីយ៖

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"]
         }
      ]
    }
  ]
}

ដូចដែលអាចមើលឃើញពីឯកសារកំណត់រចនាសម្ព័ន្ធ ភ្នាក់ងារនឹងត្រួតពិនិត្យឯកសារដែលមានផ្នែកបន្ថែម .log នៅក្នុងថត /var/log/airline_tickets/ ញែកពួកវា និងផ្ទេរពួកវាទៅស្ទ្រីម airline_tickets។

យើងចាប់ផ្តើមសេវាកម្មឡើងវិញ ហើយត្រូវប្រាកដថាវាដំណើរការហើយ៖

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

ហើយយើងមើលលទ្ធផលនៃការងារនៅក្នុងកំណត់ហេតុភ្នាក់ងារ និងនៅលើផ្ទាំងត្រួតពិនិត្យនៅក្នុងស្ទ្រីមទិន្នន័យ airline_tickets៖

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

ដូចដែលអ្នកអាចឃើញអ្វីគ្រប់យ៉ាងដំណើរការហើយភ្នាក់ងារ Kinesis បញ្ជូនទិន្នន័យដោយជោគជ័យទៅស្ទ្រីម។ ឥឡូវនេះ ចូរយើងកំណត់រចនាសម្ព័ន្ធអ្នកប្រើប្រាស់។

ដំឡើង Kinesis Data Analytics

ចូរបន្តទៅផ្នែកកណ្តាលនៃប្រព័ន្ធទាំងមូល - បង្កើតកម្មវិធីថ្មីមួយនៅក្នុង Kinesis Data Analytics ដែលមានឈ្មោះថា kinesis_analytics_airlines_app៖

Kinesis Data Analytics អនុញ្ញាតឱ្យអ្នកធ្វើការវិភាគទិន្នន័យតាមពេលវេលាជាក់ស្តែងពី Kinesis Stream ដោយប្រើភាសា SQL ។ វាគឺជាសេវាកម្មធ្វើមាត្រដ្ឋានដោយស្វ័យប្រវត្តិយ៉ាងពេញលេញ (មិនដូច Kinesis Stream) ដែល៖

1. អនុញ្ញាតឱ្យអ្នកបង្កើតស្ទ្រីមថ្មី (ស្ទ្រីមលទ្ធផល) ដោយផ្អែកលើការស្នើសុំទិន្នន័យប្រភព។
2. ផ្តល់នូវការស្ទ្រីមជាមួយនឹងកំហុសដែលបានកើតឡើងខណៈពេលដែលកម្មវិធីកំពុងដំណើរការ (Error Stream);
3. អាចកំណត់គ្រោងការណ៍ទិន្នន័យបញ្ចូលដោយស្វ័យប្រវត្តិ (វាអាចត្រូវបានកំណត់ឡើងវិញដោយដៃប្រសិនបើចាំបាច់) ។

នេះមិនមែនជាសេវាកម្មថោកទេ - 0.11 ដុល្លារក្នុងមួយម៉ោងនៃការងារ ដូច្នេះអ្នកគួរតែប្រើវាដោយប្រុងប្រយ័ត្ន ហើយលុបវាចោលនៅពេលអ្នកបានបញ្ចប់។

តោះភ្ជាប់កម្មវិធីទៅប្រភពទិន្នន័យ៖

ជ្រើសរើសការផ្សាយដែលយើងនឹងភ្ជាប់ទៅ (airline_tickets)៖

បន្ទាប់មក អ្នកត្រូវភ្ជាប់តួនាទី IAM ថ្មីដើម្បីឱ្យកម្មវិធីអាចអានពីស្ទ្រីម និងសរសេរទៅស្ទ្រីម។ ដើម្បីធ្វើដូច្នេះ វាគ្រប់គ្រាន់ហើយដែលមិនត្រូវផ្លាស់ប្តូរអ្វីទាំងអស់នៅក្នុងប្លុកការអនុញ្ញាតចូលប្រើ៖

ឥឡូវ​នេះ​យើង​ស្នើ​សុំ​ការ​រក​ឃើញ​គ្រោងការណ៍​ទិន្នន័យ​នៅ​ក្នុង​ការ​ផ្សាយ; ដើម្បី​ធ្វើ​វា​ចុច​លើ​ប៊ូតុង "Discover schema"។ ជាលទ្ធផល តួនាទី 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 Data Analytics អ្នកធ្វើការជាមួយស្ទ្រីម (ស្ទ្រីម) និងស្នប់ (PUMPs) - សំណើបញ្ចូលជាបន្តបន្ទាប់ដែលបញ្ចូលទិន្នន័យពីស្ទ្រីមមួយនៅក្នុងកម្មវិធីមួយទៅក្នុងស្ទ្រីមផ្សេងទៀត។

សំណួរ SQL ដែលបានបង្ហាញខាងលើស្វែងរកសំបុត្រ Aeroflot ក្នុងតម្លៃក្រោមប្រាំពាន់រូប្លិ៍។ កំណត់ត្រាទាំងអស់ដែលបំពេញលក្ខខណ្ឌទាំងនេះនឹងត្រូវបានដាក់នៅក្នុងស្ទ្រីម DESTINATION_SQL_STREAM ។

នៅក្នុងប្លុកទិសដៅ ជ្រើសរើសស្ទ្រីម special_stream ហើយក្នុងបញ្ជីទម្លាក់ចុះឈ្មោះស្ទ្រីមក្នុងកម្មវិធី DESTINATION_SQL_STREAM៖

លទ្ធផលនៃឧបាយកលទាំងអស់គួរតែមានលក្ខណៈស្រដៀងនឹងរូបភាពខាងក្រោម៖

ការបង្កើត និងជាវប្រធានបទ SNS

ចូលទៅកាន់សេវាជូនដំណឹងសាមញ្ញ ហើយបង្កើតប្រធានបទថ្មីនៅទីនោះដោយដាក់ឈ្មោះក្រុមហ៊ុនអាកាសចរណ៍៖

ជាវប្រធានបទនេះ ហើយបញ្ជាក់លេខទូរស័ព្ទដែលការជូនដំណឹង SMS នឹងត្រូវបានផ្ញើ៖

បង្កើតតារាងនៅក្នុង DynamoDB

ដើម្បីរក្សាទុកទិន្នន័យឆៅពីការស្ទ្រីម airline_tickets របស់ពួកគេ ចូរយើងបង្កើតតារាងនៅក្នុង DynamoDB ដែលមានឈ្មោះដូចគ្នា។ យើងនឹងប្រើ record_id ជាសោចម្បង៖

ការបង្កើតឧបករណ៍ប្រមូលមុខងារ lambda

តោះបង្កើតមុខងារ lambda ហៅថា Collector ដែលភារកិច្ចរបស់វានឹងគឺដើម្បីស្ទង់មើលការផ្សាយ airline_tickets ហើយប្រសិនបើកំណត់ត្រាថ្មីត្រូវបានរកឃើញនៅទីនោះ សូមបញ្ចូលកំណត់ត្រាទាំងនេះទៅក្នុងតារាង DynamoDB។ ជាក់ស្តែង បន្ថែមពីលើសិទ្ធិលំនាំដើម lambda នេះត្រូវតែមានសិទ្ធិអានទៅកាន់ស្ទ្រីមទិន្នន័យ Kinesis និងសរសេរសិទ្ធិចូលប្រើ DynamoDB ។

ការបង្កើតតួនាទី IAM សម្រាប់មុខងារ lambda អ្នកប្រមូល
ដំបូង យើងបង្កើតតួនាទី IAM ថ្មីសម្រាប់ lambda ដែលមានឈ្មោះថា Lambda-TicketsProcessingRole៖

សម្រាប់ឧទាហរណ៍សាកល្បង គោលការណ៍ AmazonKinesisReadOnlyAccess និង AmazonDynamoDBFullAccess ដែលបានកំណត់ជាមុនគឺសមរម្យណាស់ ដូចបង្ហាញក្នុងរូបភាពខាងក្រោម៖

lambda នេះគួរតែត្រូវបានបើកដំណើរការដោយកេះពី Kinesis នៅពេលដែលធាតុថ្មីចូលទៅក្នុង 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

មុខងារ lambda ទីពីរ ដែលនឹងត្រួតពិនិត្យការផ្សាយទីពីរ (special_stream) និងផ្ញើការជូនដំណឹងទៅ SNS ត្រូវបានបង្កើតតាមរបៀបស្រដៀងគ្នា។ ដូច្នេះ lambda នេះត្រូវតែមានសិទ្ធិចូលអានពី Kinesis និងផ្ញើសារទៅកាន់ប្រធានបទ SNS ដែលបានផ្តល់ឱ្យ ដែលបន្ទាប់មកនឹងត្រូវបានផ្ញើដោយសេវាកម្ម SNS ទៅកាន់អតិថិជនទាំងអស់នៃប្រធានបទនេះ (អ៊ីមែល សារ SMS ។ល។)។

ការបង្កើតតួនាទី IAM
ដំបូង យើងបង្កើតតួនាទី IAM Lambda-KinesisAlarm សម្រាប់ lambda នេះ ហើយបន្ទាប់មកកំណត់តួនាទីនេះទៅ alarm_notifier lambda ដែលកំពុងត្រូវបានបង្កើត៖

lambda នេះគួរតែដំណើរការនៅលើ trigger សម្រាប់កំណត់ត្រាថ្មីដើម្បីចូលទៅក្នុង special_stream ដូច្នេះអ្នកត្រូវកំណត់រចនាសម្ព័ន្ធ trigger តាមរបៀបដូចដែលយើងបានធ្វើសម្រាប់ Collector 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 គឺជាឧបករណ៍ប្រភពបើកចំហដ៏ងាយស្រួលសម្រាប់ដាក់ពង្រាយហេដ្ឋារចនាសម្ព័ន្ធពីកូដ។ វាមានវាក្យសម្ព័ន្ធផ្ទាល់ខ្លួនរបស់វា ដែលងាយស្រួលរៀន និងមានឧទាហរណ៍ជាច្រើនអំពីរបៀប និងអ្វីដែលត្រូវប្រើ។ កម្មវិធីនិពន្ធ Atom ឬ Visual Studio Code មានកម្មវិធីជំនួយងាយស្រួលជាច្រើនដែលធ្វើឱ្យការធ្វើការជាមួយ Terraform កាន់តែងាយស្រួល។

អ្នកអាចទាញយកការចែកចាយ ពីទីនេះ. ការវិភាគលម្អិតនៃសមត្ថភាព Terraform ទាំងអស់គឺហួសពីវិសាលភាពនៃអត្ថបទនេះ ដូច្នេះយើងនឹងកំណត់ខ្លួនយើងទៅនឹងចំណុចសំខាន់ៗ។

របៀបចាប់ផ្តើម

កូដពេញលេញនៃគម្រោងគឺ នៅក្នុងឃ្លាំងរបស់ខ្ញុំ. យើងក្លូនឃ្លាំងសម្រាប់ខ្លួនយើង។ មុនពេលចាប់ផ្តើម អ្នកត្រូវតែប្រាកដថាអ្នកបានដំឡើង និងកំណត់រចនាសម្ព័ន្ធ AWS CLI ព្រោះ... Terraform នឹងស្វែងរកព័ត៌មានសម្ងាត់នៅក្នុងឯកសារ ~/.aws/credentials ។

ការអនុវត្តល្អគឺត្រូវដំណើរការពាក្យបញ្ជាផែនការ មុនពេលដាក់ពង្រាយហេដ្ឋារចនាសម្ព័ន្ធទាំងមូល ដើម្បីមើលអ្វីដែល Terraform កំពុងបង្កើតសម្រាប់ពួកយើងនៅលើពពក៖

terraform.exe plan

អ្នកនឹងត្រូវបានជម្រុញឱ្យបញ្ចូលលេខទូរស័ព្ទ ដើម្បីផ្ញើការជូនដំណឹងទៅ។ វាមិនចាំបាច់ក្នុងការបញ្ចូលវានៅដំណាក់កាលនេះទេ។

ដោយបានវិភាគផែនការប្រតិបត្តិការរបស់កម្មវិធី យើងអាចចាប់ផ្តើមបង្កើតធនធាន៖

terraform.exe apply

បន្ទាប់ពីផ្ញើពាក្យបញ្ជានេះ អ្នកនឹងត្រូវបានសួរម្តងទៀតឱ្យបញ្ចូលលេខទូរស័ព្ទ សូមចុច "បាទ" នៅពេលដែលសំណួរអំពីការអនុវត្តសកម្មភាពជាក់ស្តែងត្រូវបានបង្ហាញ។ វានឹងអនុញ្ញាតឱ្យអ្នករៀបចំហេដ្ឋារចនាសម្ព័ន្ធទាំងមូល អនុវត្តការកំណត់រចនាសម្ព័ន្ធចាំបាច់ទាំងអស់នៃ EC2 ដាក់ពង្រាយមុខងារ lambda ជាដើម។

បន្ទាប់ពីធនធានទាំងអស់ត្រូវបានបង្កើតដោយជោគជ័យតាមរយៈកូដ Terraform អ្នកត្រូវចូលទៅក្នុងព័ត៌មានលម្អិតនៃកម្មវិធី Kinesis Analytics (ជាអកុសល ខ្ញុំមិនបានរកឃើញវិធីធ្វើវាដោយផ្ទាល់ពីកូដទេ)។

បើកដំណើរការកម្មវិធី៖

បន្ទាប់ពីនេះ អ្នកត្រូវតែកំណត់ឈ្មោះស្ទ្រីមក្នុងកម្មវិធីយ៉ាងច្បាស់ ដោយជ្រើសរើសពីបញ្ជីទម្លាក់ចុះ៖

ឥឡូវនេះអ្វីគ្រប់យ៉ាងគឺត្រៀមខ្លួនជាស្រេចដើម្បីទៅ។

ការធ្វើតេស្តកម្មវិធី

ដោយមិនគិតពីរបៀបដែលអ្នកដាក់ពង្រាយប្រព័ន្ធដោយដៃ ឬតាមរយៈកូដ Terraform វានឹងដំណើរការដូចគ្នា។

យើងចូលតាមរយៈ SSH ទៅកាន់ម៉ាស៊ីននិម្មិត EC2 ដែល Kinesis Agent ត្រូវបានដំឡើង ហើយដំណើរការស្គ្រីប api_caller.py

sudo ./api_caller.py TOKEN

អ្វីដែលអ្នកត្រូវធ្វើគឺរង់ចាំសារ SMS ទៅកាន់លេខរបស់អ្នក៖

សារ SMS - សារមកដល់ទូរស័ព្ទរបស់អ្នកក្នុងរយៈពេលជិត 1 នាទី៖

វានៅសល់ដើម្បីមើលថាតើកំណត់ត្រាត្រូវបានរក្សាទុកក្នុងមូលដ្ឋានទិន្នន័យ DynamoDB សម្រាប់ការវិភាគលម្អិតបន្ថែមទៀតឬអត់។ តារាង airline_tickets មានទិន្នន័យប្រហែលខាងក្រោម៖

សេចក្តីសន្និដ្ឋាន

នៅក្នុងដំណើរការនៃការងារដែលបានធ្វើ ប្រព័ន្ធដំណើរការទិន្នន័យអនឡាញត្រូវបានបង្កើតឡើងដោយផ្អែកលើ Amazon Kinesis ។ ជម្រើសសម្រាប់ការប្រើប្រាស់ភ្នាក់ងារ Kinesis ដោយភ្ជាប់ជាមួយ Kinesis Data Stream និងការវិភាគពេលវេលាជាក់ស្តែង Kinesis Analytics ដោយប្រើពាក្យបញ្ជា SQL ក៏ដូចជាអន្តរកម្មរបស់ Amazon Kinesis ជាមួយសេវាកម្ម AWS ផ្សេងទៀតត្រូវបានគេពិចារណា។

យើងបានដាក់ពង្រាយប្រព័ន្ធខាងលើតាមពីរវិធី៖ សៀវភៅដៃវែងជាង និងមួយរហ័សពីកូដ Terraform ។

កូដប្រភពគម្រោងទាំងអស់មាន នៅក្នុងឃ្លាំង GitHub របស់ខ្ញុំខ្ញុំស្នើឱ្យអ្នកស្គាល់ខ្លួនឯងជាមួយវា។

ខ្ញុំរីករាយក្នុងការពិភាក្សាអត្ថបទ ខ្ញុំទន្ទឹងរង់ចាំយោបល់របស់អ្នក។ ខ្ញុំសង្ឃឹមលើការរិះគន់ក្នុងន័យស្ថាបនា។

ខ្ញុំសូមជូនពរឱ្យអ្នកទទួលបានជោគជ័យ!

ប្រភព: www.habr.com