হ্যালো সবাই!

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

NiFi সাইট টু সাইট (S2S) হল NiFi দৃষ্টান্তগুলির মধ্যে ডেটা স্থানান্তর করার একটি নিরাপদ, অত্যন্ত কাস্টমাইজযোগ্য উপায়৷ দেখুন কিভাবে S2S কাজ করে ডকুমেন্টেশন এবং S2S দেখার অনুমতি দেওয়ার জন্য আপনার NiFi ইন্সট্যান্স সেট আপ করার কথা মনে রাখা গুরুত্বপূর্ণ এখানে.

যখন S2S ব্যবহার করে ডেটা স্থানান্তরের কথা আসে, তখন একটি উদাহরণকে ক্লায়েন্ট বলা হয়, দ্বিতীয়টি একটি সার্ভার। ক্লায়েন্ট ডেটা পাঠায়, সার্ভার তা গ্রহণ করে। তাদের মধ্যে ডেটা স্থানান্তর সেট আপ করার দুটি উপায়:

1. ধাক্কা. রিমোট প্রসেস গ্রুপ (RPG) ব্যবহার করে ক্লায়েন্ট ইন্সট্যান্স থেকে ডেটা পাঠানো হয়। সার্ভারের উদাহরণে, ইনপুট পোর্ট ব্যবহার করে ডেটা প্রাপ্ত হয়
2. টান. সার্ভার আরপিজি ব্যবহার করে ডেটা গ্রহণ করে, ক্লায়েন্ট আউটপুট পোর্ট ব্যবহার করে পাঠায়।

রোলিংয়ের জন্য ফ্লো অ্যাপাচি রেজিস্ট্রিতে সংরক্ষণ করা হয়।

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

শুরুতে, যখন N একটি ছোট সংখ্যা হয়, তখন প্রবাহ বিতরণ করা হয় এবং একটি যুক্তিসঙ্গত সময়ে হাতে আপডেট করা হয়।

কিন্তু N বৃদ্ধির সাথে সাথে আরও সমস্যা রয়েছে:

1. প্রবাহ আপডেট করতে আরও সময় লাগে। আপনাকে সব সার্ভারে যেতে হবে
2. টেমপ্লেট আপডেট করার সময় ত্রুটি আছে। এখানে তারা আপডেট, কিন্তু এখানে তারা ভুলে গেছে
3. বিপুল সংখ্যক অনুরূপ অপারেশন করার সময় মানব ত্রুটি

এই সমস্ত আমাদের এই সত্যে নিয়ে আসে যে প্রক্রিয়াটি স্বয়ংক্রিয় করা প্রয়োজন। আমি এই সমস্যা সমাধানের জন্য নিম্নলিখিত উপায় চেষ্টা করেছি:

1. NiFi এর পরিবর্তে MiNiFi ব্যবহার করুন
2. নিফাই সিএলআই
3. NiPyAPI

MiNiFi ব্যবহার করা হচ্ছে

ApacheMiNify Apache NiFi এর একটি সাবপ্রজেক্ট। MiNiFy হল একটি কমপ্যাক্ট এজেন্ট যা NiFi-এর মতো একই প্রসেসর ব্যবহার করে, যা আপনাকে NiFi-এর মতো একই প্রবাহ তৈরি করতে দেয়। MiNiFy-এর ফ্লো কনফিগারেশনের জন্য গ্রাফিকাল ইন্টারফেস না থাকার কারণে অন্যান্য জিনিসের মধ্যে এজেন্টের হালকাতা অর্জন করা হয়। MiNiFy এর একটি গ্রাফিক্যাল ইন্টারফেসের অভাবের অর্থ হল মিনিফাইতে প্রবাহ সরবরাহের সমস্যা সমাধান করা প্রয়োজন। যেহেতু MiNiFy সক্রিয়ভাবে IOT-তে ব্যবহার করা হয়, তাই অনেকগুলি উপাদান রয়েছে এবং চূড়ান্ত মিনিফাই দৃষ্টান্তগুলিতে প্রবাহ সরবরাহ করার প্রক্রিয়াটি অবশ্যই স্বয়ংক্রিয় হতে হবে। একটি পরিচিত কাজ, তাই না?

আরেকটি সাবপ্রজেক্ট, MiNiFi C2 সার্ভার, এই সমস্যা সমাধানে সাহায্য করবে। এই পণ্যটি স্থাপনার আর্কিটেকচারের কেন্দ্রীয় বিন্দু হতে উদ্দিষ্ট। পরিবেশ কীভাবে কনফিগার করবেন - এতে বর্ণিত হয়েছে এই নিবন্ধটি হ্যাব্রে এবং তথ্য সমস্যা সমাধানের জন্য যথেষ্ট। MiNiFi C2 সার্ভারের সাথে স্বয়ংক্রিয়ভাবে এর কনফিগারেশন আপডেট করে। এই পদ্ধতির একমাত্র ত্রুটি হল যে আপনাকে C2 সার্ভারে টেমপ্লেট তৈরি করতে হবে, রেজিস্ট্রিতে একটি সাধারণ প্রতিশ্রুতি যথেষ্ট নয়।

উপরের নিবন্ধে বর্ণিত বিকল্পটি কাজ করছে এবং বাস্তবায়ন করা কঠিন নয়, তবে আমাদের নিম্নলিখিতগুলি ভুলে যাওয়া উচিত নয়:

1. minifi-এ nifi থেকে সমস্ত প্রসেসর নেই
2. Minifi-এর CPU সংস্করণগুলি NiFi-এর CPU সংস্করণের চেয়ে পিছিয়ে আছে।

লেখার সময়, NiFi এর সর্বশেষ সংস্করণ 1.9.2। সর্বশেষ MiNiFi সংস্করণের প্রসেসর সংস্করণ 1.7.0। MiNiFi এ প্রসেসর যোগ করা যেতে পারে, কিন্তু NiFi এবং MiNiFi প্রসেসরের মধ্যে সংস্করণের অসঙ্গতির কারণে এটি কাজ নাও করতে পারে।

নিফাই সিএলআই

দ্বারা বিচার বর্ণনা অফিসিয়াল ওয়েবসাইটে টুল, এটি ফ্লো ডেলিভারি বা প্রক্রিয়া পরিচালনার ক্ষেত্রে NiFI এবং NiFi রেজিস্ট্রির মধ্যে মিথস্ক্রিয়া স্বয়ংক্রিয় করার জন্য একটি টুল। শুরু করতে এই টুল ডাউনলোড করুন. এখানে থেকে.

ইউটিলিটি চালান

./bin/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  /  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' to see a list of available commands, use tab to auto-complete.

আমাদের রেজিস্ট্রি থেকে প্রয়োজনীয় প্রবাহ লোড করার জন্য, আমাদের ঝুড়ির শনাক্তকারী (বালতি শনাক্তকারী) এবং প্রবাহ নিজেই (প্রবাহ শনাক্তকারী) জানতে হবে। এই ডেটা cli বা NiFi রেজিস্ট্রি ওয়েব ইন্টারফেসের মাধ্যমে প্রাপ্ত করা যেতে পারে। ওয়েব ইন্টারফেস এই মত দেখায়:

CLI ব্যবহার করে, আপনি এটি করেন:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

রেজিস্ট্রি থেকে আমদানি প্রক্রিয়া গ্রুপ চালান:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

একটি গুরুত্বপূর্ণ বিষয় হল যে কোনও nifi দৃষ্টান্ত হোস্ট হিসাবে নির্দিষ্ট করা যেতে পারে যার উপর আমরা প্রক্রিয়া গ্রুপটি রোল করি।

বন্ধ প্রসেসরের সাথে প্রসেস গ্রুপ যোগ করা হয়েছে, সেগুলি শুরু করা দরকার

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

দুর্দান্ত, প্রসেসরগুলি শুরু হয়েছে। যাইহোক, সমস্যার শর্ত অনুযায়ী, অন্যান্য উদাহরণে ডেটা পাঠাতে আমাদের NiFi দৃষ্টান্ত প্রয়োজন। ধরা যাক সার্ভারে ডেটা স্থানান্তর করার জন্য পুশ পদ্ধতিটি বেছে নেওয়া হয়েছিল। ডেটা স্থানান্তর সংগঠিত করার জন্য, যুক্ত করা রিমোট প্রসেস গ্রুপে (আরপিজি) ডেটা স্থানান্তর সক্ষম করা প্রয়োজন, যা ইতিমধ্যে আমাদের প্রবাহে অন্তর্ভুক্ত রয়েছে।

CLI এবং অন্যান্য উত্সগুলিতে ডকুমেন্টেশনে, আমি ডেটা স্থানান্তর সক্ষম করার উপায় খুঁজে পাইনি। আপনি যদি এটি করতে জানেন তবে মন্তব্যে লিখুন।

যেহেতু আমাদের বাশ আছে এবং আমরা শেষ পর্যন্ত যেতে প্রস্তুত, আমরা একটি উপায় খুঁজে বের করব! আপনি এই সমস্যার সমাধান করতে NiFi API ব্যবহার করতে পারেন। আসুন নিম্নলিখিত পদ্ধতিটি ব্যবহার করি, আমরা উপরের উদাহরণগুলি থেকে আইডি গ্রহণ করি (আমাদের ক্ষেত্রে এটি 7f522a13-016e-1000-e504-d5b15587f2f3)। NiFi API পদ্ধতির বর্ণনা এখানে.

শরীরে, আপনাকে নিম্নলিখিত ফর্মের JSON পাস করতে হবে:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

"কাজ" করার জন্য যে প্যারামিটারগুলি অবশ্যই পূরণ করতে হবে:
অবস্থা - ডেটা স্থানান্তরের স্থিতি। ডেটা স্থানান্তর সক্ষম করতে ট্রান্সমিটিং উপলব্ধ, নিষ্ক্রিয় করতে স্টপড
সংস্করণ - প্রসেসর সংস্করণ

সংস্করণ তৈরি করার সময় ডিফল্ট 0 হবে, তবে এই পরামিতিগুলি পদ্ধতি ব্যবহার করে প্রাপ্ত করা যেতে পারে

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

NiPyAPI

NiPyAPI হল NiFi দৃষ্টান্তগুলির সাথে ইন্টারঅ্যাক্ট করার জন্য একটি পাইথন লাইব্রেরি। ডকুমেন্টেশন পৃষ্ঠা লাইব্রেরির সাথে কাজ করার জন্য প্রয়োজনীয় তথ্য রয়েছে। দ্রুত শুরু বর্ণনা করা হয়েছে প্রকল্প গিথুবে।

কনফিগারেশন রোল আউট করার জন্য আমাদের স্ক্রিপ্ট হল একটি পাইথন প্রোগ্রাম। চলুন কোডিং এগিয়ে যান.
পরবর্তী কাজের জন্য কনফিগারেশন সেট আপ করুন। আমাদের নিম্নলিখিত পরামিতিগুলির প্রয়োজন হবে:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' #путь до nifi-api инстанса, на котором разворачиваем process group
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' #путь до nifi-registry-api registry
nipyapi.config.registry_name = 'MyBeutifulRegistry' #название registry, как будет называться в инстансе nifi
nipyapi.config.bucket_name = 'BucketName' #название bucket, из которого подтягиваем flow
nipyapi.config.flow_name = 'FlowName' #название flow, которое подтягиваем

আরও আমি এই লাইব্রেরির পদ্ধতিগুলির নাম সন্নিবেশ করব, যা বর্ণনা করা হয়েছে এখানে.

আমরা ব্যবহার করে nifi উদাহরণের সাথে রেজিস্ট্রি সংযোগ করি

nipyapi.versioning.create_registry_client

এই ধাপে, আপনি একটি চেক যোগ করতে পারেন যে রেজিস্ট্রি ইতিমধ্যেই উদাহরণে যোগ করা হয়েছে, এর জন্য আপনি পদ্ধতিটি ব্যবহার করতে পারেন

nipyapi.versioning.list_registry_clients

ঝুড়িতে প্রবাহের জন্য আরও অনুসন্ধান করার জন্য আমরা বালতিটি খুঁজে পাই

nipyapi.versioning.get_registry_bucket

পাওয়া বালতি অনুযায়ী, আমরা প্রবাহ খুঁজছি

nipyapi.versioning.get_flow_in_bucket

পরবর্তী, এই প্রক্রিয়া গ্রুপ ইতিমধ্যে যোগ করা হয়েছে কিনা তা বোঝা গুরুত্বপূর্ণ। প্রক্রিয়া গোষ্ঠীটি স্থানাঙ্ক দ্বারা স্থাপন করা হয় এবং একটি পরিস্থিতির উদ্ভব হতে পারে যখন একটি দ্বিতীয়টি একটির উপরে চাপানো হয়। আমি পরীক্ষা করে দেখেছি, এটি হতে পারে 🙂 সমস্ত যোগ করা প্রক্রিয়া গ্রুপ পেতে, পদ্ধতিটি ব্যবহার করুন

nipyapi.canvas.list_all_process_groups

এবং তারপর আমরা অনুসন্ধান করতে পারি, উদাহরণস্বরূপ, নাম দ্বারা।

আমি টেমপ্লেট আপডেট করার প্রক্রিয়াটি বর্ণনা করব না, আমি কেবল বলব যে যদি টেমপ্লেটের নতুন সংস্করণে প্রসেসর যুক্ত করা হয়, তবে সারিগুলিতে বার্তাগুলির উপস্থিতিতে কোনও সমস্যা নেই। কিন্তু যদি প্রসেসরগুলি সরানো হয়, তাহলে সমস্যা দেখা দিতে পারে (নিফাই প্রসেসর অপসারণের অনুমতি দেয় না যদি এটির সামনে একটি বার্তা সারি জমে থাকে)। আপনি যদি এই সমস্যার সমাধান করতে আগ্রহী হন - আমাকে লিখুন, দয়া করে, আমরা এই বিষয়টি নিয়ে আলোচনা করব। নিবন্ধের শেষে পরিচিতি. চলুন একটি প্রসেস গ্রুপ যুক্ত করার ধাপে এগিয়ে যাই।

স্ক্রিপ্টটি ডিবাগ করার সময়, আমি একটি বৈশিষ্ট্য জুড়ে এসেছি যে প্রবাহের সর্বশেষ সংস্করণটি সর্বদা টানা হয় না, তাই আমি সুপারিশ করছি যে আপনি প্রথমে এই সংস্করণটি পরিষ্কার করুন:

nipyapi.versioning.get_latest_flow_ver

প্রক্রিয়া গ্রুপ স্থাপন:

nipyapi.versioning.deploy_flow_version

আমরা প্রসেসর শুরু করি:

nipyapi.canvas.schedule_process_group

সিএলআই সম্পর্কে ব্লকে লেখা ছিল যে রিমোট প্রসেস গ্রুপে ডেটা ট্রান্সফার স্বয়ংক্রিয়ভাবে সক্ষম হয় না? স্ক্রিপ্ট বাস্তবায়ন করার সময়, আমিও এই সমস্যার সম্মুখীন হয়েছিলাম। সেই সময়ে, আমি API ব্যবহার করে ডেটা স্থানান্তর শুরু করতে পারিনি এবং আমি NiPyAPI লাইব্রেরির বিকাশকারীকে লিখতে এবং পরামর্শ / সাহায্যের জন্য জিজ্ঞাসা করার সিদ্ধান্ত নিয়েছিলাম। বিকাশকারী আমাকে উত্তর দিয়েছেন, আমরা সমস্যাটি নিয়ে আলোচনা করেছি এবং তিনি লিখেছেন যে "কিছু পরীক্ষা করার" জন্য তার সময় প্রয়োজন। এবং এখন, কয়েকদিন পরে, একটি ইমেল আসে যেখানে একটি পাইথন ফাংশন লেখা আছে যা আমার স্টার্টআপ সমস্যার সমাধান করে !!! সেই সময়ে, NiPyAPI সংস্করণটি ছিল 0.13.3 এবং অবশ্যই, এটিতে এমন কিছুই ছিল না। কিন্তু 0.14.0 সংস্করণে, যা বেশ সম্প্রতি প্রকাশিত হয়েছিল, এই ফাংশনটি ইতিমধ্যে লাইব্রেরিতে অন্তর্ভুক্ত করা হয়েছে। সম্মেলন

nipyapi.canvas.set_remote_process_group_transmission

সুতরাং, NiPyAPI লাইব্রেরির সাহায্যে, আমরা রেজিস্ট্রি সংযুক্ত করেছি, প্রবাহ চালু করেছি এবং এমনকি প্রসেসর এবং ডেটা স্থানান্তর শুরু করেছি। তারপরে আপনি কোডটি চিরুনি করতে পারেন, সমস্ত ধরণের চেক যোগ করতে পারেন, লগিং করতে পারেন এবং এটিই। কিন্তু এটা সম্পূর্ণ ভিন্ন গল্প।

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

এখানেই শেষ. আমি NiFi-এ স্বয়ংক্রিয় প্রবাহ বিতরণের 3টি পন্থা বর্ণনা করেছি, যে সমস্যাগুলি একজন ডেভেলপার সম্মুখীন হতে পারে এবং স্বয়ংক্রিয় বিতরণের জন্য একটি কার্যকরী কোড প্রদান করেছে৷ আপনি যদি এই বিষয়ে আমার মতোই আগ্রহী হন - লিখুন!

উত্স: www.habr.com