Apache NIFI - অনুশীলনে সুযোগের একটি সংক্ষিপ্ত বিবরণ

ভূমিকা

এটি তাই ঘটেছে যে আমার বর্তমান কাজের জায়গায় আমাকে এই প্রযুক্তির সাথে পরিচিত হতে হয়েছিল। আমি একটু ব্যাকগ্রাউন্ড দিয়ে শুরু করব। পরবর্তী সমাবেশে, আমাদের দলকে বলা হয়েছিল যে আমাদের সাথে একটি সংহতি তৈরি করতে হবে পরিচিত সিস্টেম. ইন্টিগ্রেশনের অর্থ হল যে এই সুপরিচিত সিস্টেমটি HTTP এর মাধ্যমে একটি নির্দিষ্ট শেষ পয়েন্টে আমাদের অনুরোধ পাঠাবে, এবং অদ্ভুতভাবে যথেষ্ট, আমরা একটি SOAP বার্তার আকারে প্রতিক্রিয়াগুলি ফেরত পাঠাব। সবকিছু সহজ এবং তুচ্ছ বলে মনে হচ্ছে। এর থেকে যা পাওয়া যায় তা হল…

কাজের

3টি পরিষেবা তৈরি করুন। প্রথমটি হল ডাটাবেস আপডেট সার্ভিস। এই পরিষেবাটি, তৃতীয় পক্ষের সিস্টেম থেকে নতুন ডেটা প্রাপ্তির পরে, ডাটাবেসে ডেটা আপডেট করে এবং পরবর্তী সিস্টেমে স্থানান্তর করার জন্য CSV ফর্ম্যাটে একটি ফাইল তৈরি করে। দ্বিতীয় পরিষেবার শেষ পয়েন্টকে বলা হয় - FTP এর মাধ্যমে পরিবহন পরিষেবা, যা স্থানান্তরিত ফাইলটি গ্রহণ করে, এটিকে যাচাই করে এবং FTP এর মাধ্যমে ফাইল স্টোরেজে রাখে। তৃতীয় পরিষেবা - গ্রাহকের কাছে ডেটা স্থানান্তর পরিষেবা, প্রথম দুটির সাথে অ্যাসিঙ্ক্রোনাসভাবে কাজ করে। এটি একটি ফাইল পাওয়ার জন্য একটি তৃতীয় পক্ষের বাহ্যিক সিস্টেম থেকে একটি অনুরোধ পায়, যা উপরে আলোচনা করা হয়েছে, একটি প্রস্তুত প্রতিক্রিয়া ফাইল নেয়, এটিকে সংশোধন করে (আইডি, বিবরণ, লিঙ্কটোফাইল ক্ষেত্রগুলি আপডেট করে) এবং একটি আকারে একটি প্রতিক্রিয়া পাঠায় SOAP বার্তা। অর্থাৎ, সাধারণভাবে, চিত্রটি নিম্নরূপ: প্রথম দুটি পরিষেবা তাদের কাজ শুরু করে যখন আপডেট করার জন্য ডেটা আসে। তৃতীয় পরিষেবাটি ক্রমাগত কাজ করে কারণ তথ্যের অনেক গ্রাহক রয়েছে, প্রতি মিনিটে ডেটার জন্য প্রায় 1000 অনুরোধ। পরিষেবাগুলি সর্বদা উপলব্ধ এবং তাদের উদাহরণগুলি বিভিন্ন পরিবেশে অবস্থিত, যেমন পরীক্ষা, ডেমো, প্রিপ্রোড এবং প্রোড৷ এই পরিষেবাগুলি কীভাবে কাজ করে তার একটি চিত্র নীচে দেওয়া হল৷ আমি এখনই ব্যাখ্যা করি যে অপ্রয়োজনীয় জটিলতা এড়াতে কিছু বিবরণ সরলীকৃত করা হয়েছে।

প্রযুক্তিগত গভীরতা

সমস্যার সমাধানের পরিকল্পনা করার সময়, আমরা প্রথমে স্প্রিং ফ্রেমওয়ার্ক, Nginx ব্যালেন্সার, পোস্টগ্রেস ডাটাবেস এবং অন্যান্য প্রযুক্তিগত এবং খুব প্রযুক্তিগত জিনিস ব্যবহার করে জাভাতে অ্যাপ্লিকেশন তৈরি করার সিদ্ধান্ত নিয়েছিলাম। যেহেতু একটি প্রযুক্তিগত সমাধান বিকাশের সময় এই সমস্যাটি সমাধান করার জন্য অন্যান্য পন্থা বিবেচনা করা সম্ভব করে তুলেছিল, চেহারাটি Apache NIFI প্রযুক্তির উপর পড়েছিল, নির্দিষ্ট চেনাশোনাগুলিতে ফ্যাশনেবল। আমি অবিলম্বে বলতে হবে যে এই প্রযুক্তিটি আমাদের এই 3টি পরিষেবা লক্ষ্য করার অনুমতি দিয়েছে। এই নিবন্ধটি একটি ফাইল পরিবহন পরিষেবা এবং ভোক্তাদের কাছে একটি ডেটা স্থানান্তর পরিষেবার বিকাশ বর্ণনা করবে, তবে, যদি নিবন্ধটি আসে তবে আমি ডাটাবেসে ডেটা আপডেট পরিষেবা সম্পর্কে লিখব।

এটা কি?

NIFI দ্রুত সমান্তরাল লোডিং এবং ডেটা প্রক্রিয়াকরণের জন্য একটি বিতরণকৃত আর্কিটেকচার, উত্স এবং রূপান্তরের জন্য প্রচুর সংখ্যক প্লাগ-ইন, কনফিগারেশন সংস্করণ এবং আরও অনেক কিছু। একটি চমৎকার বোনাস হল যে এটি ব্যবহার করা খুব সহজ। তুচ্ছ প্রক্রিয়া যেমন getFile, sendHttpRequest এবং অন্যান্যগুলিকে বর্গ হিসাবে উপস্থাপন করা যেতে পারে। প্রতিটি বর্গক্ষেত্র একটি নির্দিষ্ট প্রক্রিয়ার প্রতিনিধিত্ব করে, যার মিথস্ক্রিয়া নীচের চিত্রে দেখা যায়। প্রক্রিয়া সেটআপ মিথস্ক্রিয়া সম্পর্কে আরও বিস্তারিত ডকুমেন্টেশন লেখা হয়েছে এখানে , যারা রাশিয়ান ভাষায় - তাদের জন্য এখানে. ডকুমেন্টেশন নিখুঁতভাবে বর্ণনা করে যে কীভাবে NIFI আনপ্যাক এবং চালাতে হয়, সেইসাথে কীভাবে প্রক্রিয়াগুলি তৈরি করতে হয়, সেগুলিও বর্গক্ষেত্র।
একটি নিবন্ধ লেখার ধারণাটি একটি দীর্ঘ অনুসন্ধান এবং প্রাপ্ত তথ্যকে সচেতন কিছুতে গঠন করার পরে, সেইসাথে ভবিষ্যতের বিকাশকারীদের জন্য জীবনকে আরও সহজ করার ইচ্ছার জন্ম হয়েছিল।

উদাহরণ

বর্গাকার কিভাবে একে অপরের সাথে যোগাযোগ করে তার একটি উদাহরণ বিবেচনা করা হয়। সাধারণ স্কিমটি বেশ সহজ: আমরা একটি HTTP অনুরোধ পাই (তত্ত্ব অনুসারে, অনুরোধের অংশে একটি ফাইল সহ। NIFI এর ক্ষমতা প্রদর্শন করতে, এই উদাহরণে, অনুরোধটি স্থানীয় FH থেকে একটি ফাইল পাওয়ার প্রক্রিয়া শুরু করে), তারপরে আমরা একটি প্রতিক্রিয়া ফেরত পাঠাই যে অনুরোধটি গৃহীত হয়েছে, সমান্তরালভাবে, FH থেকে একটি ফাইল পাওয়ার প্রক্রিয়া এবং তারপর FTP এর মাধ্যমে FH-এ সরানোর প্রক্রিয়া। এটি ব্যাখ্যা করার মতো যে প্রক্রিয়াগুলি তথাকথিত ফ্লোফাইলের মাধ্যমে একে অপরের সাথে যোগাযোগ করে। এটি NIFI-এর মৌলিক সত্তা যা বৈশিষ্ট্য এবং বিষয়বস্তু সঞ্চয় করে। বিষয়বস্তু - স্ট্রীম ফাইল দ্বারা প্রতিনিধিত্ব করা হয় যে ডেটা. অর্থাৎ, মোটামুটিভাবে বলতে গেলে, আপনি যদি একটি বর্গক্ষেত্র থেকে একটি ফাইল গ্রহণ করেন এবং এটি অন্য বর্গক্ষেত্রে স্থানান্তর করেন তবে আপনার ফাইলটি হবে বিষয়বস্তু।

আপনি দেখতে পাচ্ছেন, এই ছবিটি সামগ্রিক প্রক্রিয়া দেখায়। HandleHttpRequest - অনুরোধ গ্রহণ করে, ReplaceText - একটি প্রতিক্রিয়া বডি তৈরি করে, HandleHttpResponse - একটি প্রতিক্রিয়া দেয়। FetchFile - ফাইল স্টোরেজ থেকে একটি ফাইল গ্রহণ করে এবং এটি PutSftp স্কোয়ারে স্থানান্তর করে - এই ফাইলটিকে FTP-এ, নির্দিষ্ট ঠিকানায় রাখে। এখন এই প্রক্রিয়া সম্পর্কে আরো.

এই ক্ষেত্রে, অনুরোধ সবকিছুর শুরু। এর কনফিগারেশন অপশন দেখি।

স্ট্যান্ডার্ডএইচটিপি কনটেক্সটম্যাপ ব্যতীত এখানে সবকিছুই খুব তুচ্ছ - এটি এমন একটি পরিষেবা যা আপনাকে অনুরোধ পাঠাতে এবং গ্রহণ করতে দেয়। আরো বিস্তারিত জানার জন্য এবং এমনকি উদাহরণ সহ, আপনি দেখতে পারেন - এখানে

এর পরে, আসুন ReplaceText স্কোয়ার কনফিগারেশন বিকল্পগুলি দেখি। এটি প্রতিস্থাপন মূল্যের দিকে মনোযোগ দেওয়া মূল্যবান - এটি প্রতিক্রিয়া হিসাবে ব্যবহারকারীকে ফেরত দেওয়া হবে। সেটিংসে, আপনি লগিং স্তর সামঞ্জস্য করতে পারেন, আপনি লগগুলি দেখতে পারেন {যেখানে nifi আনপ্যাক করা হয়েছিল}/nifi-1.9.2/logs, এছাড়াও ব্যর্থতা / সাফল্যের পরামিতি রয়েছে - এই প্যারামিটারগুলির উপর ভিত্তি করে, আপনি প্রক্রিয়াটিকে নিয়ন্ত্রণ করতে পারেন সম্পূর্ণ অর্থাৎ, সফল পাঠ্য প্রক্রিয়াকরণের ক্ষেত্রে, ব্যবহারকারীকে একটি প্রতিক্রিয়া পাঠানোর প্রক্রিয়াটি বলা হবে, এবং অন্য ক্ষেত্রে, আমরা কেবল ব্যর্থ প্রক্রিয়াটির প্রতিশ্রুতি দিই।

HandleHttpResponse বৈশিষ্ট্যগুলিতে বিশেষ আকর্ষণীয় কিছু নেই, যখন প্রতিক্রিয়া সফলভাবে তৈরি করা হয়েছিল তখন স্ট্যাটাস ব্যতীত।

আমরা অনুরোধের প্রতিক্রিয়াটি বের করেছি - আসুন ফাইলটি গ্রহণ করা এবং এটি FTP সার্ভারে স্থাপন করা যাক৷ FetchFile - সেটিংসে নির্দিষ্ট পথ থেকে একটি ফাইল গ্রহণ করে এবং পরবর্তী প্রক্রিয়ায় স্থানান্তর করে।

এবং তারপর বর্গাকার PutSftp - ফাইলটিকে ফাইল স্টোরেজে রাখে। কনফিগারেশন অপশন নিচে দেখা যাবে.

এটি প্রতিটি বর্গক্ষেত্র একটি পৃথক প্রক্রিয়া যে চালু করা আবশ্যক যে মনোযোগ দিতে মূল্য। আমরা সবচেয়ে সহজ উদাহরণ বিবেচনা করেছি যার জন্য কোন জটিল কাস্টমাইজেশনের প্রয়োজন নেই। এর পরে, আমরা প্রক্রিয়াটিকে আরও জটিল বিবেচনা করব, যেখানে আমরা খাঁজে একটু লিখব।

আরও জটিল উদাহরণ

SOAP বার্তা পরিবর্তন করার প্রক্রিয়ার মাধ্যমে গ্রাহকের কাছে ডেটা স্থানান্তর পরিষেবাটি আরও জটিল করে তোলা হয়েছে। সামগ্রিক প্রক্রিয়াটি নীচের চিত্রে দেখানো হয়েছে।

এখানে, ধারণাটিও খুব জটিল নয়: আমরা ভোক্তার কাছ থেকে একটি অনুরোধ পেয়েছি যে তার ডেটা প্রয়োজন, একটি প্রতিক্রিয়া পাঠিয়েছি যে তিনি একটি বার্তা পেয়েছেন, প্রতিক্রিয়া ফাইল গ্রহণের প্রক্রিয়া শুরু করেছেন, তারপর নির্দিষ্ট যুক্তি দিয়ে এটি সম্পাদনা করেছেন এবং তারপরে স্থানান্তরিত করেছেন। সার্ভারে একটি SOAP বার্তা আকারে গ্রাহকের কাছে ফাইলটি।

আমি মনে করি যে স্কোয়ারগুলি আমরা উপরে দেখেছি তা আবার বর্ণনা করা মূল্যবান নয় - এখনই নতুনগুলির দিকে এগিয়ে যাওয়া যাক৷ আপনি যদি কোনো ফাইল সম্পাদনা করতে চান এবং ReplaceText-এর মতো সাধারণ বর্গক্ষেত্র উপযুক্ত না হয়, তাহলে আপনাকে নিজের স্ক্রিপ্ট লিখতে হবে। এটি ExecuteGroogyScript বর্গ ব্যবহার করে করা যেতে পারে। এর সেটিংস নীচে দেখানো হয়েছে।

এই বর্গক্ষেত্রে স্ক্রিপ্ট লোড করার জন্য দুটি বিকল্প আছে। প্রথমটি একটি স্ক্রিপ্ট সহ একটি ফাইল আপলোড করে। দ্বিতীয়টি হল স্ক্রিপ্টবডিতে স্ক্রিপ্ট ঢোকানো। যতদূর আমি জানি, executeScript স্কোয়ারটি বেশ কয়েকটি PL সমর্থন করে - তাদের মধ্যে একটি গ্রুভি। আমি জাভা বিকাশকারীদের হতাশ করব - আপনি এই ধরনের স্কোয়ারে জাভাতে স্ক্রিপ্ট লিখতে পারবেন না। যারা সত্যিই চান তাদের জন্য, আপনাকে আপনার নিজস্ব কাস্টম স্কোয়ার তৈরি করতে হবে এবং এটি NIFI সিস্টেমে ফেলে দিতে হবে। এই পুরো অপারেশনটি একটি দফের সাথে বরং দীর্ঘ নাচের সাথে রয়েছে, যা আমরা এই নিবন্ধে মোকাবেলা করব না। আমি গ্রোভি ভাষা বেছে নিয়েছি। নীচে একটি পরীক্ষা স্ক্রিপ্ট যা সহজভাবে SOAP বার্তায় আইডি আপডেট করে। এটা লক্ষ্য করা গুরুত্বপূর্ণ। আপনি ফ্লোফাইল থেকে একটি ফাইল নিন এবং এটি আপডেট করুন, ভুলে যাবেন না যে আপনার এটির প্রয়োজন, আপডেট করুন, এটি সেখানে আবার রাখুন। এটাও লক্ষণীয় যে সব লাইব্রেরি সংযুক্ত নয়। এটি চালু হতে পারে যে আপনাকে এখনও একটি libs আমদানি করতে হবে৷ খারাপ দিক হল এই স্কোয়ারের স্ক্রিপ্টটি ডিবাগ করা বেশ কঠিন। NIFI JVM এর সাথে সংযোগ করার এবং ডিবাগিং প্রক্রিয়া শুরু করার একটি উপায় রয়েছে। ব্যক্তিগতভাবে, আমি একটি স্থানীয় অ্যাপ্লিকেশন চালিয়েছি এবং সেশন থেকে একটি ফাইল পাওয়ার অনুকরণ করেছি। ডিবাগিংও স্থানীয়ভাবে করা হয়েছিল। স্ক্রিপ্টটি লোড করার সময় যে ত্রুটিগুলি আসে তা গুগলের কাছে বেশ সহজ এবং লগে 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