Google ক্লাউড ডেটাফ্লো সহ নোটবুক ব্যবহার করা

1. ভূমিকা

গুগল ক্লাউড ডেটাফ্লো

শেষ আপডেট: 2023-জুলাই-5

Dataflow কি?

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

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

গতির সাথে ডেটা বিশ্লেষণ স্ট্রিমিং

ডেটাফ্লো কম ডেটা লেটেন্সি সহ দ্রুত, সরলীকৃত স্ট্রিমিং ডেটা পাইপলাইন বিকাশ সক্ষম করে।

ক্রিয়াকলাপ এবং পরিচালনা সহজতর করুন

দলগুলিকে সার্ভার ক্লাস্টারগুলি পরিচালনার পরিবর্তে প্রোগ্রামিং-এ ফোকাস করার অনুমতি দিন কারণ ডেটাফ্লো-এর সার্ভারহীন পদ্ধতি ডেটা ইঞ্জিনিয়ারিং কাজের চাপ থেকে অপারেশনাল ওভারহেড সরিয়ে দেয়৷

মালিকানার মোট খরচ কমান

রিসোর্স অটোস্কেলিং খরচ-অপ্টিমাইজ করা ব্যাচ প্রসেসিং ক্ষমতার সাথে যুক্ত মানে ডেটাফ্লো অতিরিক্ত খরচ ছাড়াই আপনার মৌসুমী এবং স্পাইকি কাজের চাপ পরিচালনা করার জন্য কার্যত সীমাহীন ক্ষমতা প্রদান করে।

মূল বৈশিষ্ট্য

স্বয়ংক্রিয় রিসোর্স ম্যানেজমেন্ট এবং ডাইনামিক ওয়ার্ক রিব্যালেন্সিং

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

অনুভূমিক অটোস্কেলিং

সর্বোত্তম থ্রুপুট ফলাফলের জন্য কর্মী সংস্থানগুলির অনুভূমিক অটোস্কেলিংয়ের ফলে সামগ্রিক মূল্য-থেকে-পারফরম্যান্স আরও ভাল হয়।

ব্যাচ প্রক্রিয়াকরণের জন্য নমনীয় সম্পদ সময় নির্ধারণের মূল্য

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

এর অংশ হিসেবে আপনি কি চালাবেন

JupyterLab নোটবুকগুলির সাথে Apache Beam ইন্টারেক্টিভ রানার ব্যবহার করে আপনাকে পুনরাবৃত্তিমূলকভাবে পাইপলাইনগুলি বিকাশ করতে দেয়, আপনার পাইপলাইন গ্রাফ পরিদর্শন করতে এবং একটি রিড-ইভাল-প্রিন্ট-লুপ (REPL) ওয়ার্কফ্লোতে পৃথক PC সংগ্রহগুলিকে পার্স করতে দেয়৷ এই Apache Beam নোটবুকগুলি Vertex AI Workbench- এর মাধ্যমে উপলব্ধ করা হয়েছে, একটি পরিচালিত পরিষেবা যা সর্বশেষ ডেটা সায়েন্স এবং মেশিন লার্নিং ফ্রেমওয়ার্কগুলির সাথে নোটবুক ভার্চুয়াল মেশিনগুলিকে আগে থেকে ইনস্টল করে।

এই কোডল্যাবটি Apache Beam নোটবুক দ্বারা প্রবর্তিত কার্যকারিতার উপর দৃষ্টি নিবদ্ধ করে।

আপনি কি শিখবেন

• কীভাবে একটি নোটবুক উদাহরণ তৈরি করবেন
• একটি মৌলিক পাইপলাইন তৈরি করা হচ্ছে
• সীমাহীন উৎস থেকে ডেটা পড়া
• ডেটা ভিজ্যুয়ালাইজ করা
• নোটবুক থেকে একটি ডেটাফ্লো কাজ চালু করা হচ্ছে
• একটি নোটবুক সংরক্ষণ করা হচ্ছে

আপনি কি প্রয়োজন হবে

• বিলিং সক্ষম সহ একটি Google ক্লাউড প্ল্যাটফর্ম প্রকল্প৷
• Google ক্লাউড ডেটাফ্লো এবং Google ক্লাউড পাবসাব সক্ষম।

2. সেট আপ করা হচ্ছে

1. ক্লাউড কনসোলে, প্রকল্প নির্বাচক পৃষ্ঠায়, একটি ক্লাউড প্রকল্প নির্বাচন করুন বা তৈরি করুন।

দ্রষ্টব্য : আপনি যদি এই পদ্ধতিতে আপনার তৈরি করা সংস্থানগুলি রাখার পরিকল্পনা না করেন তবে একটি বিদ্যমান প্রকল্প নির্বাচন করার পরিবর্তে একটি নতুন প্রকল্প তৈরি করুন৷ আপনি এই পদক্ষেপগুলি শেষ করার পরে, আপনি প্রকল্পের সাথে যুক্ত সমস্ত সংস্থান মুছে ফেলতে পারেন।

প্রকল্প নির্বাচক পৃষ্ঠায় যান

নিশ্চিত করুন যে আপনি নিম্নলিখিত API গুলি সক্ষম করেছেন:

• ডেটাফ্লো API
• ক্লাউড পাব/সাব API
• কম্পিউট ইঞ্জিন
• নোটবুক API

আপনি API এর এবং পরিষেবা পৃষ্ঠায় চেক করে এটি যাচাই করতে পারেন।

API এর এবং পরিষেবা পৃষ্ঠাতে যান

এই নির্দেশিকায়, আমরা একটি পাব/সাবস্ক্রিপশন থেকে ডেটা পড়ব, তাই নিশ্চিত করুন যে কম্পিউট ইঞ্জিন ডিফল্ট পরিষেবা অ্যাকাউন্টে সম্পাদকের ভূমিকা রয়েছে, অথবা এটিকে পাব/সাব এডিটরের ভূমিকা মঞ্জুর করুন ৷

3. Apache Beam নোটবুক দিয়ে শুরু করা

একটি Apache Beam নোটবুক ইনস্ট্যান্স চালু করা হচ্ছে

1. কনসোলে ডেটাফ্লো চালু করুন:

কনসোলে ডেটাফ্লোতে যান

2. বাম-হাতের মেনু ব্যবহার করে ওয়ার্কবেঞ্চ পৃষ্ঠাটি নির্বাচন করুন।
3. নিশ্চিত করুন যে আপনি ব্যবহারকারী-পরিচালিত নোটবুক ট্যাবে আছেন।
4. টুলবারে, নতুন নোটবুকে ক্লিক করুন।
5. Apache Beam > GPU ছাড়া নির্বাচন করুন।
6. নতুন নোটবুক পৃষ্ঠায়, নোটবুক VM-এর জন্য একটি সাবনেটওয়ার্ক নির্বাচন করুন এবং তৈরি করুন ক্লিক করুন।
7. লিঙ্কটি সক্রিয় হয়ে গেলে JupyterLab খুলুন ক্লিক করুন। Vertex AI Workbench একটি নতুন Apache Beam নোটবুক উদাহরণ তৈরি করে।

দ্রষ্টব্য : একটি ব্যবহারকারী-পরিচালিত নোটবুক ইনস্ট্যান্স খোলার পরে, উদাহরণ ফোল্ডারে উদাহরণ নোটবুক পাওয়া যায়। নিম্নলিখিত বর্তমানে উপলব্ধ:

• শব্দ গণনা
• স্ট্রিমিং শব্দ গণনা
• এনওয়াইসি ট্যাক্সি রাইড ডেটা স্ট্রিমিং
• ডেটাফ্লো শব্দ গণনা
• নোটবুকে Apache Beam SQL
• স্কেলে ইন্টারেক্টিভ ফ্লিঙ্ক
• Apache Beam সহ GPU ব্যবহার করুন
• ডেটা ভিজ্যুয়ালাইজ করুন

এই নোটবুকগুলিতে ব্যাখ্যামূলক পাঠ্য এবং মন্তব্য করা কোড ব্লকগুলি অন্তর্ভুক্ত রয়েছে যাতে আপনি Apache Beam ধারণা এবং API ব্যবহার বুঝতে সহায়তা করেন।

4. পাইপলাইন তৈরি করা

একটি নোটবুক উদাহরণ তৈরি করা হচ্ছে

ফাইল > নতুন > নোটবুকে নেভিগেট করুন এবং Apache Beam 2.47 বা তার পরের একটি কার্নেল নির্বাচন করুন।

দ্রষ্টব্য : Apache Beam নোটবুকগুলি Apache Beam SDK-এর মাস্টার শাখার বিপরীতে নির্মিত। এর মানে হল যে নোটবুক UI এ দেখানো কার্নেলের সর্বশেষ সংস্করণটি SDK-এর সাম্প্রতিক প্রকাশিত সংস্করণের চেয়ে এগিয়ে থাকতে পারে।

আপনার নোটবুকে কোড যোগ করা শুরু করুন

• আপনার নোটবুকের একটি নতুন কক্ষের মধ্যে প্রতিটি বিভাগ থেকে কোডটি অনুলিপি করুন এবং আটকান
• সেল চালান

JupyterLab নোটবুকগুলির সাথে Apache Beam ইন্টারেক্টিভ রানার ব্যবহার করে আপনাকে পুনরাবৃত্তিমূলকভাবে পাইপলাইনগুলি বিকাশ করতে দেয়, আপনার পাইপলাইন গ্রাফ পরিদর্শন করতে এবং একটি রিড-ইভাল-প্রিন্ট-লুপ (REPL) ওয়ার্কফ্লোতে পৃথক PC সংগ্রহগুলিকে পার্স করতে দেয়৷

Apache Beam আপনার নোটবুকের উদাহরণে ইনস্টল করা আছে, তাই আপনার নোটবুকে interactive_runner এবং interactive_beam মডিউলগুলি অন্তর্ভুক্ত করুন।

import apache_beam as beam
from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib

যদি আপনার নোটবুক অন্যান্য Google পরিষেবা ব্যবহার করে, তাহলে নিম্নলিখিত আমদানি বিবৃতি যোগ করুন:

from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth

ইন্টারঅ্যাক্টিভিটি বিকল্পগুলি সেট করা

নিম্নলিখিত ডেটা ক্যাপচারের সময়কাল 60 সেকেন্ডে সেট করে। আপনি যদি দ্রুত পুনরাবৃত্তি করতে চান তবে এটিকে একটি কম সময়সীমাতে সেট করুন, উদাহরণস্বরূপ '10s'।

ib.options.recording_duration = '60s'

অতিরিক্ত ইন্টারেক্টিভ বিকল্পের জন্য, interactive_beam.options ক্লাস দেখুন।

একটি InteractiveRunner অবজেক্ট ব্যবহার করে পাইপলাইন শুরু করুন।

# Setting up the Apache Beam pipeline options.
options = pipeline_options.PipelineOptions()

# Set the pipeline mode to stream the data from Pub/Sub.
options.view_as(pipeline_options.StandardOptions).streaming = True

# Sets the project to the default project in your current Google Cloud environment.
# The project will be used for creating a subscription to the Pub/Sub topic.
_, options.view_as(GoogleCloudOptions).project = google.auth.default()

# The Google Cloud PubSub topic for this example.
topic = "projects/pubsub-public-data/topics/shakespeare-kinglear"

p = beam.Pipeline(InteractiveRunner(), options=options)

ডেটা পড়া এবং ভিজ্যুয়ালাইজ করা

নিম্নলিখিত উদাহরণটি একটি Apache Beam পাইপলাইন দেখায় যা প্রদত্ত Pub/Sub বিষয়ের সদস্যতা তৈরি করে এবং সদস্যতা থেকে পড়ে।

words = p | "read" >> beam.io.ReadFromPubSub(topic=topic)

পাইপলাইন উৎস থেকে জানালা দিয়ে শব্দ গণনা করে। এটি স্থির উইন্ডো তৈরি করে যার প্রতিটি উইন্ডো 10 সেকেন্ড সময়কালের।

windowed_words = (words
   | "window" >> beam.WindowInto(beam.window.FixedWindows(10)))

ডেটা উইন্ডো করার পরে, শব্দগুলি উইন্ডো দ্বারা গণনা করা হয়।

windowed_word_counts = (windowed_words
   | "count" >> beam.combiners.Count.PerElement())

ডেটা ভিজ্যুয়ালাইজ করা

show() পদ্ধতিটি নোটবুকে ফলাফলপ্রাপ্ত PC সংগ্রহকে কল্পনা করে।

ib.show(windowed_word_counts, include_window_info=True)

আপনার ডেটার ভিজ্যুয়ালাইজেশন প্রদর্শন করতে, show() পদ্ধতিতে visualize_data=True পাস করুন। একটি নতুন সেল যোগ করুন:

ib.show(windowed_word_counts, include_window_info=True, visualize_data=True)

আপনি আপনার ভিজ্যুয়ালাইজেশনে একাধিক ফিল্টার প্রয়োগ করতে পারেন। নিম্নলিখিত ভিজ্যুয়ালাইজেশন আপনাকে লেবেল এবং অক্ষ দ্বারা ফিল্টার করতে দেয়:

দ্রষ্টব্য : স্ট্রিমিং পাইপলাইন প্রোটোটাইপ করার সময় রিপ্লেবিলিটি নিশ্চিত করতে, শো() পদ্ধতি কলগুলি ডিফল্টরূপে ক্যাপচার করা ডেটা পুনরায় ব্যবহার করে। interactive_beam.options.enable_capture_replay = False সেট করে, আপনি এই আচরণ পরিবর্তন করতে পারেন এবং show() পদ্ধতিতে সর্বদা নতুন ডেটা আনতে পারেন। এছাড়াও, আপনি যদি আপনার নোটবুকে একটি দ্বিতীয় সীমাহীন উত্স যোগ করেন, তবে পূর্ববর্তী অসীম উৎস থেকে ডেটা বাতিল করা হয়।

5. একটি পান্ডাস ডেটাফ্রেম ব্যবহার করা

Apache Beam নোটবুকের আরেকটি দরকারী ভিজ্যুয়ালাইজেশন হল একটি পান্ডাস ডেটাফ্রেম । নিম্নলিখিত উদাহরণটি প্রথমে শব্দগুলিকে ছোট হাতের অক্ষরে রূপান্তর করে এবং তারপর প্রতিটি শব্দের ফ্রিকোয়েন্সি গণনা করে।

windowed_lower_word_counts = (windowed_words
   | "to lower case" >> beam.Map(lambda word: word.lower())
   | "count lowered" >> beam.combiners.Count.PerElement())

collect() পদ্ধতি একটি পান্ডাস ডেটাফ্রেমে আউটপুট প্রদান করে।

ib.collect(windowed_lower_word_counts, include_window_info=True)

দ্রষ্টব্য : নোটবুক বিকাশে একটি সেল সম্পাদনা এবং পুনরায় কার্যকর করা একটি সাধারণ অভ্যাস। আপনি যখন Apache Beam নোটবুকে একটি ঘর সম্পাদনা করেন এবং পুনরায় সম্পাদন করেন, তখন সেলটি মূল কক্ষের কোডের উদ্দেশ্যমূলক ক্রিয়াটিকে পূর্বাবস্থায় ফিরিয়ে আনে না। উদাহরণস্বরূপ, যদি একটি সেল একটি পাইপলাইনে একটি PTransform যোগ করে, সেই সেলটি পুনরায় কার্যকর করা হলে পাইপলাইনে একটি অতিরিক্ত PTransform যোগ হবে। আপনি যদি রাজ্যটি পরিষ্কার করতে চান, কার্নেলটি পুনরায় চালু করুন এবং কোষগুলি পুনরায় চালু করুন।

6. (ঐচ্ছিক) আপনার নোটবুক থেকে ডেটাফ্লো কাজ চালু করা

1. ডেটাফ্লোতে কাজ চালানোর জন্য, আপনার অতিরিক্ত অনুমতির প্রয়োজন। কম্পিউট ইঞ্জিন ডিফল্ট পরিষেবা অ্যাকাউন্টে সম্পাদকের ভূমিকা রয়েছে তা নিশ্চিত করুন বা এটিকে নিম্নলিখিত IAM ভূমিকাগুলি মঞ্জুর করুন :

• ডেটাফ্লো অ্যাডমিন
• ডেটাফ্লো কর্মী
• স্টোরেজ অ্যাডমিন এবং
• পরিষেবা অ্যাকাউন্ট ব্যবহারকারী (roles/iam.serviceAccountUser)

ডকুমেন্টেশনে ভূমিকা সম্পর্কে আরও দেখুন।

2. (ঐচ্ছিক) ডেটাফ্লো কাজগুলি চালানোর জন্য আপনার নোটবুক ব্যবহার করার আগে, কার্নেলটি পুনরায় চালু করুন, সমস্ত কক্ষ পুনরায় চালু করুন এবং আউটপুট যাচাই করুন।
3. নিম্নলিখিত আমদানি বিবৃতি সরান:

from apache_beam.runners.interactive.interactive_runner import InteractiveRunner
import apache_beam.runners.interactive.interactive_beam as ib

4. নিম্নলিখিত আমদানি বিবৃতি যোগ করুন:

from apache_beam.runners import DataflowRunner

5. নিম্নলিখিত রেকর্ডিং সময়কাল বিকল্পটি সরান:

ib.options.recording_duration = '60s'

6. আপনার পাইপলাইন বিকল্পগুলিতে নিম্নলিখিত যোগ করুন। আপনার ইতিমধ্যেই মালিকানাধীন একটি বালতিতে নির্দেশ করতে আপনাকে ক্লাউড স্টোরেজ অবস্থান সামঞ্জস্য করতে হবে, অথবা আপনি এই উদ্দেশ্যে একটি নতুন বালতি তৈরি করতে পারেন। এছাড়াও আপনি us-central1 থেকে অঞ্চলের মান পরিবর্তন করতে পারেন।

# Set the Google Cloud region to run Dataflow.
options.view_as(GoogleCloudOptions).region = 'us-central1'

# Choose a Cloud Storage location.
dataflow_gcs_location = 'gs://<change me>/dataflow'

# Choose a location for your output files
output_gcs_location = '%s/results/' % dataflow_gcs_location


# Set the staging location. This location is used to stage the
# Dataflow pipeline and SDK binary.
options.view_as(GoogleCloudOptions).staging_location = '%s/staging' % dataflow_gcs_location

# Set the temporary location. This location is used to store temporary files
# or intermediate results before outputting to the sink.
options.view_as(GoogleCloudOptions).temp_location = '%s/temp' % dataflow_gcs_location

7. beam.Pipeline() এর কনস্ট্রাক্টরে, DataflowRunner দিয়ে InteractiveRunner প্রতিস্থাপন করুন। p হল আপনার পাইপলাইন তৈরির পাইপলাইন অবজেক্ট।

p = beam.Pipeline(DataflowRunner(), options=options)

8. আপনার কোড থেকে ইন্টারেক্টিভ কল সরান. উদাহরণস্বরূপ, আপনার কোড থেকে show() , collect() , head() , show_graph() , এবং watch() সরান।
9. কোনো ফলাফল দেখতে সক্ষম হতে আপনাকে একটি সিঙ্ক যোগ করতে হবে। পূর্ববর্তী বিভাগে আমরা নোটবুকের ফলাফলগুলি ভিজ্যুয়ালাইজ করছিলাম, কিন্তু এবার, আমরা এই নোটবুকের বাইরে কাজটি চালাচ্ছি - ডেটাফ্লোতে৷ অতএব, আমাদের ফলাফলের জন্য আমাদের একটি বাহ্যিক অবস্থান প্রয়োজন। এই উদাহরণে, আমরা GCS (Google ক্লাউড স্টোরেজ) টেক্সট ফাইলগুলিতে ফলাফল লিখব। যেহেতু এটি একটি স্ট্রিমিং পাইপলাইন, ডেটা উইন্ডো সহ, আমরা প্রতি উইন্ডোতে একটি টেক্সট ফাইল তৈরি করতে চাই। এটি অর্জন করতে, আপনার পাইপলাইনে নিম্নলিখিত পদক্ষেপগুলি যুক্ত করুন:

result = (windowed_lower_word_counts
    | "decode words and format" >> beam.Map(lambda x: f"{x[0].decode('utf-8')}: {x[1]}")
    | "write events to file" >> beam.io.fileio.WriteToFiles(path=output_gcs_location, shards=1, max_writers_per_bundle=0))

10. আপনার পাইপলাইন কোডের শেষে p.run() যোগ করুন।
11. আপনি সমস্ত পরিবর্তনগুলিকে অন্তর্ভুক্ত করেছেন তা নিশ্চিত করতে এখন আপনার নোটবুকের কোড পর্যালোচনা করুন৷ এটি এর মতো দেখতে হবে:

import apache_beam as beam
from apache_beam.options import pipeline_options
from apache_beam.options.pipeline_options import GoogleCloudOptions
import google.auth
from apache_beam.runners import DataflowRunner

# Setting up the Apache Beam pipeline options.
options = pipeline_options.PipelineOptions()

# Set the pipeline mode to stream the data from Pub/Sub.
options.view_as(pipeline_options.StandardOptions).streaming = True

# Sets the project to the default project in your current Google Cloud environment.
# The project will be used for creating a subscription to the Pub/Sub topic.
_, options.view_as(GoogleCloudOptions).project = google.auth.default()

# Set the Google Cloud region to run Dataflow.
options.view_as(GoogleCloudOptions).region = 'us-central1'

# Choose a Cloud Storage location.
dataflow_gcs_location = 'gs://<change me>/dataflow'

# Choose a location for your output files
output_gcs_location = '%s/results/' % dataflow_gcs_location


# Set the staging location. This location is used to stage the
# Dataflow pipeline and SDK binary.
options.view_as(GoogleCloudOptions).staging_location = '%s/staging' % dataflow_gcs_location

# Set the temporary location. This location is used to store temporary files
# or intermediate results before outputting to the sink.
options.view_as(GoogleCloudOptions).temp_location = '%s/temp' % dataflow_gcs_location



# The Google Cloud PubSub topic for this example.
topic = "projects/pubsub-public-data/topics/shakespeare-kinglear"

p = beam.Pipeline(DataflowRunner(), options=options)
words = p | "read" >> beam.io.ReadFromPubSub(topic=topic)
windowed_words = (words
   | "window" >> beam.WindowInto(beam.window.FixedWindows(10)))

windowed_words_counts = (windowed_words
   | "count" >> beam.combiners.Count.PerElement())

windowed_lower_word_counts = (windowed_words
   | "to lower case" >> beam.Map(lambda word: word.lower())
   | "count lowered" >> beam.combiners.Count.PerElement())

result = (windowed_lower_word_counts
    | "decode words and format" >> beam.Map(lambda x: f"{x[0].decode('utf-8')}: {x[1]}")
    | "write events to file" >> beam.io.fileio.WriteToFiles(path=output_gcs_location, shards=1, max_writers_per_bundle=0))

p.run()

12. কোষ চালান।
13. আপনি নিম্নলিখিত একটি অনুরূপ আউটপুট দেখতে হবে:

<DataflowPipelineResult <Job
 clientRequestId: '20230623100011457336-8998'
 createTime: '2023-06-23T10:00:33.447347Z'
 currentStateTime: '1970-01-01T00:00:00Z'
 id: '2023-06-23_03_00_33-11346237320103246437'
 location: 'us-central1'
 name: 'beamapp-root-0623075553-503897-boh4u4wb'
 projectId: 'your-project-id'
 stageStates: []
 startTime: '2023-06-23T10:00:33.447347Z'
 steps: []
 tempFiles: []
 type: TypeValueValuesEnum(JOB_TYPE_STREAMING, 2)> at 0x7fc7e4084d60>

14. কাজটি চলছে কিনা তা যাচাই করতে, Dataflow-এর জন্য চাকরির পৃষ্ঠায় যান। আপনার তালিকায় একটি নতুন চাকরি দেখতে হবে। কাজটি ডেটা প্রক্রিয়াকরণ শুরু করতে প্রায় ~5-10 মিনিট সময় নেবে৷
15. ডেটা প্রসেসিং হয়ে গেলে, ক্লাউড স্টোরেজ- এ যান এবং ডাটাফ্লো ফলাফলগুলি সংরক্ষণ করে এমন ডিরেক্টরিতে নেভিগেট করুন (আপনার সংজ্ঞায়িত output_gcs_location )। প্রতি উইন্ডোতে একটি ফাইল সহ আপনাকে পাঠ্য ফাইলগুলির একটি তালিকা দেখতে হবে।
16. ফাইল ডাউনলোড করুন এবং বিষয়বস্তু পরিদর্শন করুন. এটিতে তাদের গণনার সাথে যুক্ত করা শব্দের তালিকা থাকা উচিত। বিকল্পভাবে, ফাইলগুলি পরিদর্শন করতে কমান্ড লাইন ইন্টারফেস ব্যবহার করুন। আপনি আপনার নোটবুকের একটি নতুন ঘরে নিম্নলিখিতগুলি চালিয়ে এটি করতে পারেন:

! gsutils cat gs://<your-bucket-here>/results/<file-name>

17. আপনি এটির অনুরূপ আউটপুট দেখতে পাবেন:

Safer: 1

trust: 1

mercy: 1

harms: 1

far: 2

fear: 1

than: 1

take: 1

me: 1

goneril: 1

still: 1

away: 1

let: 1

too: 2

the: 1

18. তাই তো! আপনার তৈরি করা কাজটি পরিষ্কার করতে এবং বন্ধ করতে ভুলবেন না (এই কোডল্যাবের চূড়ান্ত ধাপ দেখুন)।

একটি ইন্টারেক্টিভ নোটবুকে কীভাবে এই রূপান্তরটি সম্পাদন করতে হয় তার একটি উদাহরণের জন্য, আপনার নোটবুকের উদাহরণে ডেটাফ্লো ওয়ার্ড কাউন্ট নোটবুকটি দেখুন।

বিকল্পভাবে, আপনি একটি এক্সিকিউটেবল স্ক্রিপ্ট হিসাবে আপনার নোটবুক রপ্তানি করতে পারেন, পূর্ববর্তী পদক্ষেপগুলি ব্যবহার করে জেনারেট করা .py ফাইলটি সংশোধন করতে পারেন এবং তারপরে ডেটাফ্লো পরিষেবাতে আপনার পাইপলাইন স্থাপন করতে পারেন ৷

7. আপনার নোটবুক সংরক্ষণ করা হচ্ছে

আপনার তৈরি নোটবুকগুলি আপনার চলমান নোটবুক উদাহরণে স্থানীয়ভাবে সংরক্ষিত হয়। আপনি যদি ডেভেলপমেন্টের সময় নোটবুক ইন্সট্যান্স রিসেট বা বন্ধ করেন, তাহলে সেই নতুন নোটবুকগুলি যতক্ষণ না /home/jupyter ডিরেক্টরির অধীনে তৈরি করা হয় ততক্ষণ পর্যন্ত টিকে থাকে। যাইহোক, যদি একটি নোটবুক উদাহরণ মুছে ফেলা হয়, সেই নোটবুকগুলিও মুছে ফেলা হয়।

ভবিষ্যতে ব্যবহারের জন্য আপনার নোটবুকগুলিকে রাখতে, সেগুলিকে স্থানীয়ভাবে আপনার ওয়ার্কস্টেশনে ডাউনলোড করুন, সেগুলিকে GitHub-এ সংরক্ষণ করুন , অথবা একটি ভিন্ন ফাইল বিন্যাসে রপ্তানি করুন৷

8. পরিষ্কার করা

আপনি আপনার Apache Beam নোটবুক ইন্সট্যান্স ব্যবহার করা শেষ করার পরে, নোটবুক ইনস্ট্যান্স বন্ধ করে এবং স্ট্রিমিং কাজ বন্ধ করে Google ক্লাউডে আপনার তৈরি সংস্থানগুলি পরিষ্কার করুন, যদি আপনি একটি চালান।

বিকল্পভাবে, আপনি যদি এই কোডল্যাবের একমাত্র উদ্দেশ্যের জন্য একটি প্রকল্প তৈরি করে থাকেন, তাহলে আপনি প্রকল্পটি সম্পূর্ণরূপে বন্ধ করে দিতে পারেন।