Apache NIFI - ទិដ្ឋភាពទូទៅសង្ខេបនៃលក្ខណៈពិសេសក្នុងការអនុវត្ត

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

វាបានកើតឡើងដូច្នេះថានៅកន្លែងធ្វើការបច្ចុប្បន្នរបស់ខ្ញុំ ខ្ញុំត្រូវតែស្គាល់បច្ចេកវិទ្យានេះ។ ខ្ញុំនឹងចាប់ផ្តើមជាមួយផ្ទៃខាងក្រោយតិចតួច។ នៅក្នុងកិច្ចប្រជុំបន្ទាប់ ក្រុមការងាររបស់យើងត្រូវបានប្រាប់ថា យើងត្រូវបង្កើតការរួមបញ្ចូលជាមួយ ប្រព័ន្ធដែលគេស្គាល់. តាមរយៈការធ្វើសមាហរណកម្ម វាមានន័យថាប្រព័ន្ធដ៏ល្បីនេះនឹងបញ្ជូនយើងនូវសំណើតាមរយៈ HTTP ទៅកាន់ចំណុចបញ្ចប់ជាក់លាក់មួយ ហើយយើងពិតជាអាចផ្ញើការឆ្លើយតបមកវិញក្នុងទម្រង់ជាសារ SOAP ។ អ្វីគ្រប់យ៉ាងហាក់ដូចជាសាមញ្ញ និងតូចតាច។ ពីនេះវាដូចខាងក្រោមដែលអ្នកត្រូវការ ...

គោលបំណង

បង្កើតសេវាកម្ម 3 ។ ទីមួយក្នុងចំនោមពួកគេគឺសេវាធ្វើបច្ចុប្បន្នភាពមូលដ្ឋានទិន្នន័យ។ សេវាកម្មនេះ នៅពេលដែលទិន្នន័យថ្មីមកដល់ពីប្រព័ន្ធភាគីទីបី ធ្វើបច្ចុប្បន្នភាពទិន្នន័យនៅក្នុងមូលដ្ឋានទិន្នន័យ និងបង្កើតឯកសារក្នុងទម្រង់ CSV ដើម្បីផ្ទេរវាទៅប្រព័ន្ធបន្ទាប់។ ចំណុចបញ្ចប់នៃសេវាកម្មទីពីរត្រូវបានគេហៅថា - សេវាដឹកជញ្ជូន FTP ដែលទទួលឯកសារដែលបានផ្ទេរ ធ្វើឱ្យមានសុពលភាព ហើយដាក់វានៅក្នុងកន្លែងផ្ទុកឯកសារតាមរយៈ FTP ។ សេវាកម្មទីបី សេវាផ្ទេរទិន្នន័យអ្នកប្រើប្រាស់ ដំណើរការដោយអសមកាលជាមួយពីរដំបូង។ វាទទួលសំណើពីប្រព័ន្ធខាងក្រៅរបស់ភាគីទីបី ដើម្បីទទួលឯកសារដែលបានពិភាក្សាខាងលើ យកឯកសារឆ្លើយតបរួចរាល់ កែប្រែវា (ធ្វើបច្ចុប្បន្នភាពលេខសម្គាល់ ការពិពណ៌នា វាល linkToFile) និងផ្ញើការឆ្លើយតបក្នុងទម្រង់ជាសារ SOAP ។ នោះគឺរូបភាពរួមមានដូចខាងក្រោម៖ សេវាកម្មពីរដំបូងចាប់ផ្តើមការងាររបស់ពួកគេតែនៅពេលដែលទិន្នន័យសម្រាប់ការធ្វើបច្ចុប្បន្នភាពបានមកដល់។ សេវាកម្មទីបីដំណើរការឥតឈប់ឈរ ពីព្រោះមានអ្នកប្រើប្រាស់ព័ត៌មានច្រើន ប្រហែល 1000 សំណើសម្រាប់ទិន្នន័យក្នុងមួយនាទី។ សេវាកម្មអាចរកបានឥតឈប់ឈរ ហើយករណីរបស់ពួកគេមានទីតាំងនៅក្នុងបរិយាកាសផ្សេងៗគ្នា ដូចជាការធ្វើតេស្ត ការបង្ហាញ ការផលិតមុន និងផលិតផល។ ខាងក្រោមនេះគឺជាដ្យាក្រាមនៃរបៀបដែលសេវាកម្មទាំងនេះដំណើរការ។ ខ្ញុំសូមបញ្ជាក់ភ្លាមៗថាព័ត៌មានលម្អិតមួយចំនួនត្រូវបានធ្វើឱ្យសាមញ្ញ ដើម្បីជៀសវាងភាពស្មុគស្មាញដែលមិនចាំបាច់។

ការធ្វើឱ្យស៊ីជម្រៅផ្នែកបច្ចេកទេស

នៅពេលរៀបចំផែនការដំណោះស្រាយចំពោះបញ្ហា យើងបានសម្រេចចិត្តបង្កើតកម្មវិធីនៅក្នុង Java ដោយប្រើ Spring framework, Nginx balancer, Postgres database និងបច្ចេកទេសផ្សេងទៀត ហើយមិនមែនជារឿងបច្ចេកទេសនោះទេ។ ចាប់តាំងពីពេលបង្កើតដំណោះស្រាយបច្ចេកទេសបានអនុញ្ញាតឱ្យយើងពិចារណាវិធីសាស្រ្តផ្សេងទៀតក្នុងការដោះស្រាយបញ្ហានេះ ការសម្លឹងរបស់យើងបានធ្លាក់ទៅលើបច្ចេកវិទ្យា Apache NIFI ដែលជាម៉ូដទាន់សម័យនៅក្នុងរង្វង់ជាក់លាក់។ ខ្ញុំ​នឹង​និយាយ​ភ្លាម​ថា​បច្ចេកវិទ្យា​នេះ​អនុញ្ញាត​ឱ្យ​យើង​សម្គាល់​ឃើញ​សេវាកម្ម​ទាំង 3 នេះ។ អត្ថបទនេះនឹងរៀបរាប់ពីការអភិវឌ្ឍន៍សេវាកម្មដឹកជញ្ជូនឯកសារ និងសេវាកម្មផ្ទេរទិន្នន័យដល់អ្នកប្រើប្រាស់ ប៉ុន្តែប្រសិនបើអត្ថបទមានប្រយោជន៍ ខ្ញុំនឹងសរសេរអំពីសេវាកម្មសម្រាប់ធ្វើបច្ចុប្បន្នភាពទិន្នន័យក្នុងមូលដ្ឋានទិន្នន័យ។

តើវាជាអ្វី?

NIFI គឺជាស្ថាបត្យកម្មចែកចាយសម្រាប់ការផ្ទុក និងដំណើរការទិន្នន័យស្របគ្នាយ៉ាងលឿន កម្មវិធីជំនួយមួយចំនួនធំសម្រាប់ប្រភព និងការផ្លាស់ប្តូរ កំណែនៃការកំណត់ និងច្រើនទៀត។ ប្រាក់រង្វាន់ដ៏ល្អមួយគឺថាវាងាយស្រួលប្រើណាស់។ ដំណើរការមិនសំខាន់ដូចជា getFile, sendHttpRequest និងផ្សេងទៀតអាចត្រូវបានតំណាងជាការ៉េ។ ការ៉េនីមួយៗតំណាងឱ្យដំណើរការមួយ អន្តរកម្មដែលអាចមើលឃើញនៅក្នុងរូបភាពខាងក្រោម។ ឯកសារលម្អិតបន្ថែមទៀតអំពីអន្តរកម្មនៃការដំឡើងដំណើរការត្រូវបានសរសេរ នៅទីនេះ សម្រាប់អ្នកដែលនិយាយភាសារុស្សី - នៅទីនេះ. ឯកសារពិពណ៌នាយ៉ាងល្អឥតខ្ចោះអំពីរបៀបស្រាយ និងដំណើរការ NIFI ក៏ដូចជារបៀបបង្កើតដំណើរការ ដែលត្រូវបានគេស្គាល់ថាជាការ៉េ
គំនិតក្នុងការសរសេរអត្ថបទមួយបានកើតបន្ទាប់ពីការស្វែងរកដ៏យូរ និងរៀបចំរចនាសម្ព័ន្ធព័ត៌មានដែលទទួលបានទៅជាអ្វីមួយដែលដឹងខ្លួន ក៏ដូចជាបំណងប្រាថ្នាចង់ធ្វើឱ្យជីវិតកាន់តែងាយស្រួលសម្រាប់អ្នកអភិវឌ្ឍន៍នាពេលអនាគត។

ឧទាហរណ៍:

ឧទាហរណ៍នៃរបៀបដែលការ៉េធ្វើអន្តរកម្មគ្នាទៅវិញទៅមកត្រូវបានពិចារណា។ គ្រោងការណ៍ទូទៅគឺសាមញ្ញណាស់៖ យើងទទួលបានសំណើ HTTP (តាមទ្រឹស្តី ជាមួយនឹងឯកសារនៅក្នុងតួនៃសំណើ។ ដើម្បីបង្ហាញពីសមត្ថភាពរបស់ NIFI ក្នុងឧទាហរណ៍នេះ សំណើចាប់ផ្តើមដំណើរការទទួលឯកសារពីកន្លែងផ្ទុកឯកសារមូលដ្ឋាន។ ) បន្ទាប់មកយើងផ្ញើត្រឡប់មកវិញនូវការឆ្លើយតបដែលសំណើត្រូវបានទទួល ស្របគ្នានឹងដំណើរការនៃការទទួលឯកសារពី FH ហើយបន្ទាប់មកដំណើរការផ្លាស់ប្តូរវាតាមរយៈ FTP ទៅ FH ។ វាមានតម្លៃក្នុងការបញ្ជាក់ឱ្យច្បាស់ថាដំណើរការមានអន្តរកម្មគ្នាទៅវិញទៅមកតាមរយៈអ្វីដែលគេហៅថា flowFile ។ នេះគឺជាអង្គភាពមូលដ្ឋាននៅក្នុង NIFI ដែលរក្សាទុកគុណលក្ខណៈ និងខ្លឹមសារ។ មាតិកាគឺជាទិន្នន័យដែលត្រូវបានតំណាងដោយឯកសារស្ទ្រីម។ ពោល​គឺ​បើ​អ្នក​ទទួល​ឯកសារ​មួយ​ពី​ការ៉េ​មួយ​ហើយ​ផ្ទេរ​វា​ទៅ​មួយ​ទៀត មាតិកា​នឹង​ជា​ឯកសារ​របស់​អ្នក។

ដូចដែលអ្នកអាចឃើញរូបភាពនេះបង្ហាញពីដំណើរការទូទៅ។ HandleHttpRequest - ទទួលយកសំណើ ជំនួសអត្ថបទ - បង្កើតតួការឆ្លើយតប HandleHttpResponse - ផ្ញើការឆ្លើយតប។ FetchFile - ទទួលឯកសារពីកន្លែងផ្ទុកឯកសារ ផ្ទេរវាទៅការ៉េ PutSftp - ដាក់ឯកសារនេះនៅលើ FTP តាមអាសយដ្ឋានដែលបានបញ្ជាក់។ ឥឡូវនេះបន្ថែមទៀតអំពីដំណើរការនេះ។

ក្នុងករណីនេះការស្នើសុំគឺជាការចាប់ផ្តើមនៃអ្វីគ្រប់យ៉ាង។ សូមក្រឡេកមើលប៉ារ៉ាម៉ែត្រកំណត់រចនាសម្ព័ន្ធរបស់វា។

អ្វីគ្រប់យ៉ាងនៅទីនេះគឺតូចតាចណាស់ លើកលែងតែ StandardHttpContextMap - នេះគឺជាប្រភេទនៃសេវាកម្មដែលអនុញ្ញាតឱ្យអ្នកផ្ញើ និងទទួលសំណើ។ នៅក្នុងលម្អិតបន្ថែមទៀត និងសូម្បីតែជាមួយឧទាហរណ៍ អ្នកអាចមើលឃើញ - នៅទីនេះ

បន្ទាប់​មក យើង​មើល​ប៉ារ៉ាម៉ែត្រ​កំណត់​រចនាសម្ព័ន្ធ​ជំនួស​អត្ថបទ​នៃ​ការ៉េ។ វាគួរអោយយកចិត្តទុកដាក់ចំពោះ ReplacementValue - នេះគឺជាអ្វីដែលនឹងត្រូវប្រគល់ជូនអ្នកប្រើប្រាស់វិញក្នុងទម្រង់ជាការឆ្លើយតប។ នៅក្នុងការកំណត់ អ្នកអាចកែតម្រូវកម្រិតនៃការកត់ត្រា អ្នកអាចឃើញកំណត់ហេតុ {កន្លែងដែលអ្នកពន្លា nifi}/nifi-1.9.2/logs ក៏មានប៉ារ៉ាម៉ែត្របរាជ័យ/ជោគជ័យផងដែរ - ដោយផ្អែកលើប៉ារ៉ាម៉ែត្រទាំងនេះ អ្នកអាចគ្រប់គ្រងដំណើរការទាំងមូល . នោះគឺក្នុងករណីដែលដំណើរការអត្ថបទជោគជ័យ ដំណើរការនៃការផ្ញើការឆ្លើយតបទៅកាន់អ្នកប្រើប្រាស់នឹងត្រូវបានហៅ ហើយក្នុងករណីមួយទៀតយើងនឹងកត់ត្រាដំណើរការដែលមិនជោគជ័យ។

មិនមានអ្វីគួរឱ្យចាប់អារម្មណ៍ជាពិសេសនៅក្នុងលក្ខណៈសម្បត្តិ HandleHttpResponse លើកលែងតែស្ថានភាពនៅពេលដែលការឆ្លើយតបត្រូវបានបង្កើតដោយជោគជ័យ។

យើងបានតម្រៀបចេញនូវសំណើ និងការឆ្លើយតប - ចូរបន្តទៅការទទួលឯកសារ ហើយដាក់វានៅលើម៉ាស៊ីនមេ FTP ។ FetchFile - ទទួលឯកសារនៅផ្លូវដែលបានបញ្ជាក់ក្នុងការកំណត់ ហើយបញ្ជូនវាទៅដំណើរការបន្ទាប់។

ហើយបន្ទាប់មក PutSftp square - ដាក់ឯកសារនៅក្នុងកន្លែងផ្ទុកឯកសារ។ យើងអាចមើលឃើញប៉ារ៉ាម៉ែត្រកំណត់រចនាសម្ព័ន្ធខាងក្រោម។

វាគួរអោយយកចិត្តទុកដាក់ចំពោះការពិតដែលថាការ៉េនីមួយៗគឺជាដំណើរការដាច់ដោយឡែកដែលត្រូវតែចាប់ផ្តើម។ យើងបានក្រឡេកមើលឧទាហរណ៍សាមញ្ញបំផុតដែលមិនតម្រូវឱ្យមានការប្ដូរតាមបំណងដ៏ស្មុគស្មាញណាមួយឡើយ។ បន្ទាប់ យើងនឹងពិនិត្យមើលដំណើរការដែលស្មុគស្មាញបន្តិច ដែលយើងនឹងសរសេរបន្តិចបន្តួចនៅលើចង្អូរ។

ឧទាហរណ៍ស្មុគស្មាញជាង

សេវាកម្មផ្ទេរទិន្នន័យទៅអ្នកប្រើប្រាស់បានប្រែជាស្មុគស្មាញបន្តិច ដោយសារដំណើរការនៃការកែប្រែសារ SOAP ។ ដំណើរការទូទៅត្រូវបានបង្ហាញនៅក្នុងរូបភាពខាងក្រោម។

នៅទីនេះគំនិតនេះក៏មិនស្មុគស្មាញជាពិសេសដែរ៖ យើងបានទទួលសំណើពីអ្នកប្រើប្រាស់ថាគាត់ត្រូវការទិន្នន័យ ផ្ញើការឆ្លើយតបថាគាត់បានទទួលសារ ចាប់ផ្តើមដំណើរការទទួលឯកសារឆ្លើយតប បន្ទាប់មកកែសម្រួលវាដោយតក្កវិជ្ជាជាក់លាក់ ហើយបន្ទាប់មក ផ្ទេរឯកសារទៅអ្នកប្រើប្រាស់ក្នុងទម្រង់ជាសារ SOAP ទៅកាន់ម៉ាស៊ីនមេ។

ខ្ញុំ​គិត​ថា​មិន​ចាំបាច់​ពណ៌នា​ម្តងទៀត​នូវ​ការ៉េ​ដែល​យើង​បាន​ឃើញ​ខាង​លើ​នោះ​ទេ - ចូរ​ផ្លាស់ទី​ត្រង់​ទៅ​កន្លែង​ថ្មី។ ប្រសិនបើអ្នកត្រូវការកែសម្រួលឯកសារណាមួយ ហើយការេប្រភេទ ReplaceText ធម្មតាមិនសមស្របទេ អ្នកនឹងត្រូវសរសេរស្គ្រីបផ្ទាល់ខ្លួនរបស់អ្នក។ នេះអាចត្រូវបានធ្វើដោយប្រើ ExecuteGrogyScript square ។ ការកំណត់របស់វាត្រូវបានបង្ហាញខាងក្រោម។

មានជម្រើសពីរសម្រាប់ការផ្ទុកស្គ្រីបទៅក្នុងការ៉េនេះ។ ទីមួយគឺដោយការទាញយកឯកសារដែលមានស្គ្រីប។ ទីពីរគឺដោយការបញ្ចូលស្គ្រីបទៅក្នុង scriptBody ។ តាមដែលខ្ញុំដឹង ការ៉េ executeScript គាំទ្រភាសាជាច្រើន - មួយក្នុងចំណោមពួកគេគឺ groovy ។ ខ្ញុំនឹងខកចិត្តអ្នកអភិវឌ្ឍន៍ java - អ្នកមិនអាចសរសេរស្គ្រីបក្នុង java ក្នុងការ៉េបែបនេះបានទេ។ សម្រាប់អ្នកដែលពិតជាចង់ អ្នកត្រូវបង្កើតការ៉េផ្ទាល់ខ្លួនរបស់អ្នក ហើយបន្ថែមវាទៅក្នុងប្រព័ន្ធ NIFI ។ ប្រតិបត្តិការទាំងមូលនេះត្រូវបានអមដោយការរាំដ៏វែងមួយជាមួយនឹង tambourine ដែលយើងនឹងមិនដោះស្រាយនៅក្នុងអត្ថបទនេះទេ។ ខ្ញុំបានជ្រើសរើសភាសារអ៊ូរទាំ។ ខាងក្រោមគឺជាស្គ្រីបសាកល្បងដែលគ្រាន់តែបង្កើនការអាប់ដេតលេខសម្គាល់នៅក្នុងសារ SOAP ។ វាជាការសំខាន់ក្នុងការកត់សម្គាល់។ អ្នកយកឯកសារពី flowFile ហើយអាប់ដេតវា កុំភ្លេចថាអ្នកត្រូវដាក់វាវិញនៅទីនោះ ធ្វើបច្ចុប្បន្នភាព។ វាក៏គួរអោយកត់សំគាល់ផងដែរថាមិនមែនបណ្ណាល័យទាំងអស់ត្រូវបានរួមបញ្ចូលទេ។ វាអាចកើតឡើងដែលអ្នកនៅតែត្រូវនាំចូល libs មួយ។ គុណវិបត្តិមួយទៀតគឺថា ស្គ្រីបក្នុងការ៉េនេះពិបាកដោះស្រាយណាស់។ មានវិធីមួយដើម្បីភ្ជាប់ទៅ NIFI JVM ហើយចាប់ផ្តើមដំណើរការបំបាត់កំហុស។ ដោយផ្ទាល់ ខ្ញុំបានបើកដំណើរការកម្មវិធីក្នុងស្រុក និងក្លែងធ្វើការទទួលឯកសារពីវគ្គ។ ខ្ញុំ​ក៏​បាន​កែ​កំហុស​ក្នុង​ស្រុក​ដែរ។ កំហុសដែលលេចឡើងនៅពេលផ្ទុកស្គ្រីបគឺងាយស្រួលសម្រាប់ Google ហើយត្រូវបានសរសេរដោយ NIFI ខ្លួនវាទៅក្នុងកំណត់ហេតុ។

import org.apache.commons.io.IOUtils
import groovy.xml.XmlUtil
import java.nio.charset.*
import groovy.xml.StreamingMarkupBuilder

def flowFile = session.get()
if (!flowFile) return
try {
    flowFile = session.write(flowFile, { inputStream, outputStream ->
        String result = IOUtils.toString(inputStream, "UTF-8");
        def recordIn = new XmlSlurper().parseText(result)
        def element = recordIn.depthFirst().find {
            it.name() == 'id'
        }

        def newId = Integer.parseInt(element.toString()) + 1
        def recordOut = new XmlSlurper().parseText(result)
        recordOut.Body.ClientMessage.RequestMessage.RequestContent.content.MessagePrimaryContent.ResponseBody.id = newId

        def res = new StreamingMarkupBuilder().bind { mkp.yield recordOut }.toString()
        outputStream.write(res.getBytes(StandardCharsets.UTF_8))
} as StreamCallback)
     session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
    log.error("Error during processing of validate.groovy", e)
    session.transfer(flowFile, REL_FAILURE)
}

តាមពិតនេះគឺជាកន្លែងដែលការប្ដូរតាមបំណងនៃការ៉េបញ្ចប់។ បន្ទាប់មកឯកសារដែលបានធ្វើបច្ចុប្បន្នភាពត្រូវបានផ្ទេរទៅការ៉េដែលទទួលខុសត្រូវក្នុងការបញ្ជូនឯកសារទៅម៉ាស៊ីនមេ។ ខាងក្រោមនេះគឺជាការកំណត់សម្រាប់ការ៉េនេះ។

យើងពណ៌នាអំពីវិធីសាស្ត្រដែលសារ SOAP នឹងត្រូវបានបញ្ជូន។ យើងសរសេរកន្លែងណា។ បន្ទាប់អ្នកត្រូវបង្ហាញថានេះគឺជា SOAP ។

បន្ថែមលក្ខណៈសម្បត្តិមួយចំនួនដូចជាម៉ាស៊ីន និងសកម្មភាព (soapAction)។ យើងរក្សាទុកនិងពិនិត្យ។ អ្នកអាចមើលព័ត៌មានលម្អិតបន្ថែមអំពីរបៀបផ្ញើសំណើ SOAP នៅទីនេះ

យើងបានពិនិត្យមើលជម្រើសជាច្រើនសម្រាប់ការប្រើប្រាស់ដំណើរការ NIFI ។ តើ​ពួក​គេ​មាន​ទំនាក់​ទំនង​យ៉ាង​ណា ហើយ​អ្វី​ជា​ប្រយោជន៍​ពិត​ប្រាកដ​របស់​ពួក​គេ? ឧទាហរណ៍​ដែល​គេ​ចាត់​ទុក​គឺ​ជា​ការ​សាកល្បង ហើយ​ខុស​គ្នា​បន្តិច​ពី​អ្វី​ដែល​កើត​ឡើង​ពិត​ប្រាកដ​ក្នុង​ការ​ប្រយុទ្ធ។ ខ្ញុំសង្ឃឹមថាអត្ថបទនេះនឹងមានប្រយោជន៍តិចតួចសម្រាប់អ្នកអភិវឌ្ឍន៍។ សូម​អរគុណ​ចំពោះ​ការ​យកចិត្ត​ទុកដាក់​របស់​លោកអ្នក។ ប្រសិនបើអ្នកមានសំណួរសូមសរសេរ។ ខ្ញុំនឹងព្យាយាមឆ្លើយ។

ប្រភព: www.habr.com