Parallel Processing Large File In Python

প্লেটো দ্বারা প্রকাশিত

অনুসরণকারী: 0

পাইথনে প্যারালাল প্রসেসিং বড় ফাইল
লেখকের ছবি

সমান্তরাল প্রক্রিয়াকরণের জন্য, আমরা আমাদের কাজকে সাব-ইউনিটে ভাগ করি। এটি প্রোগ্রাম দ্বারা প্রক্রিয়াকৃত কাজের সংখ্যা বৃদ্ধি করে এবং সামগ্রিক প্রক্রিয়াকরণের সময় হ্রাস করে।

উদাহরণস্বরূপ, যদি আপনি একটি বড় CSV ফাইল নিয়ে কাজ করেন এবং আপনি একটি একক কলাম পরিবর্তন করতে চান। আমরা ফাংশনে একটি অ্যারে হিসাবে ডেটা ফিড করব এবং এটি উপলব্ধ সংখ্যার উপর ভিত্তি করে একসাথে একাধিক মান সমান্তরাল প্রক্রিয়া করবে শ্রমিকদের. এই কর্মীরা আপনার প্রসেসরের মধ্যে কোরের সংখ্যার উপর ভিত্তি করে।

বিঃদ্রঃ: একটি ছোট ডেটাসেটে সমান্তরাল প্রক্রিয়াকরণ ব্যবহার করে প্রক্রিয়াকরণের সময় উন্নত হবে না।

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

বিঃদ্রঃ: আমরা পরীক্ষার জন্য কাগল নোটবুক ব্যবহার করছি। প্রক্রিয়াকরণের সময় মেশিন থেকে মেশিনে পরিবর্তিত হতে পারে।

আমরা ব্যবহার করা হবে মার্কিন দুর্ঘটনা (2016 – 2021) Kaggle থেকে ডেটাসেট যা 2.8 মিলিয়ন রেকর্ড এবং 47টি কলাম নিয়ে গঠিত।

আমরা আমদানি করব multiprocessing, joblib, এবং tqdm উন্নত সমান্তরাল প্রক্রিয়াকরণ, pandas উন্নত ডেটা ইনজেশন, এবং re, nltk, এবং string উন্নত পাঠ্য প্রক্রিয়াকরণ.

# সমান্তরাল কম্পিউটিং
আমদানি মাল্টিপ্রসেসিং as mp
থেকে joblib আমদানি সমান্তরাল, বিলম্বিত
থেকে tqdm.notebook আমদানি tqdm # ডেটা ইনজেশন 
আমদানি পান্ডাস as pd # পাঠ্য প্রক্রিয়াকরণ 
আমদানি re থেকে nltk.corpus আমদানি স্টপওয়ার্ড
আমদানি স্ট্রিং

আমরা ডানে ঝাঁপ দেওয়ার আগে, এর সেট করা যাক n_workers দ্বিগুণ করে cpu_count(). আপনি দেখতে পাচ্ছেন, আমাদের 8 জন কর্মী রয়েছে।

n_workers = 2 * mp.cpu_count() প্রিন্ট(f"{n_workers} কর্মী উপলব্ধ") >>> 8 জন কর্মী পাওয়া যায়

পরবর্তী ধাপে, আমরা ব্যবহার করে বড় CSV ফাইল ইনজেস্ট করব পান্ডাস read_csv ফাংশন তারপরে, ডেটাফ্রেমের আকৃতি, কলামের নাম এবং প্রক্রিয়াকরণের সময় মুদ্রণ করুন।

বিঃদ্রঃ: জুপিটার এর জাদু ফাংশন %%time প্রদর্শন করতে পারে CPU বার এবং প্রাচীর সময় প্রক্রিয়া শেষে।

%%time file_name="../input/us-accidents/US_Accidents_Dec21_updated.csv" df = pd.read_csv(file_name) প্রিন্ট(f"শেপ:{df.shape}nn কলামের নাম:n{df.columns}n")

আউটপুট

আকৃতি:(2845342, 47) কলামের নাম: সূচক(['ID', 'তীব্রতা', 'Start_Time', 'End_Time', 'Start_Lat', 'Start_Lng', 'End_Lat', 'End_Lng', 'দূরত্ব(mi) ', 'বিবরণ', 'সংখ্যা', 'রাস্তা', 'পার্শ্ব', 'শহর', 'কাউন্টি', 'রাজ্য', 'জিপকোড', 'দেশ', 'টাইমজোন', 'এয়ারপোর্ট_কোড', 'ওয়েদার_টাইমস্ট্যাম্প', 'তাপমাত্রা(F)', 'Wind_Chill(F)', 'আর্দ্রতা(%)', 'চাপ(in)', 'দৃশ্যমানতা(mi)', 'Wind_Direction', 'Wind_Speed(mph)', 'Recipitation(in) )', 'আবহাওয়া_পরিস্থিতি', 'সুবিধা', 'বাম্প', 'ক্রসিং', 'গিভ_ওয়ে', 'জংশন', 'নো_এক্সিট', 'রেলওয়ে', 'রাউন্ডঅবাউট', 'স্টেশন', 'স্টপ', 'ট্রাফিক_ক্যালিং' , 'ট্রাফিক_সিগন্যাল', 'টার্নিং_লুপ', 'সানরাইজ_সানসেট', 'সিভিল_টোয়াইলাইট', 'নটিক্যাল_টোয়াইলাইট', 'অ্যাস্ট্রোনমিক্যাল_টোয়াইলাইট'], dtype='অবজেক্ট') CPU বার: ব্যবহারকারী 33.9 s, sys: 3.93s, 37.9 মোট সময়: 46.9। : XNUMX সেকেন্ড

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

Def পরিষ্কার_পাঠ্য(পাঠ্য): # স্টপ শব্দগুলি সরান stops = stopwords.words("ইংরেজি") টেক্সট = " ". join([শব্দ উন্নত  শব্দ in text.split() if শব্দ না in থামে]) # বিশেষ অক্ষর সরান text = text.translate(str.maketrans('', '', স্ট্রিং.বিরামচিহ্ন)) # অতিরিক্ত স্থান অপসারণ text = re.sub(' +',' ', টেক্সট) প্রত্যাবর্তন পাঠ

সিরিয়াল প্রক্রিয়াকরণের জন্য, আমরা পান্ডা ব্যবহার করতে পারি .apply() ফাংশন, কিন্তু আপনি যদি অগ্রগতি বার দেখতে চান তবে আপনাকে সক্রিয় করতে হবে tqdm উন্নত পান্ডাস এবং তারপর ব্যবহার করুন .progress_apply() ফাংশন.

আমরা 2.8 মিলিয়ন রেকর্ড প্রক্রিয়া করতে যাচ্ছি এবং ফলাফলটিকে আবার "বিবরণ" কলাম কলামে সংরক্ষণ করতে যাচ্ছি।

%%time tqdm.pandas() df['Description'] = df['Description'].progress_apply(clean_text)

আউটপুট

এর জন্য সময় লেগেছে ৯ মিনিট ৫ সেকেন্ড হাই-এন্ড প্রসেসর থেকে সিরিয়াল প্রক্রিয়া 2.8 মিলিয়ন সারি।

100% 2845342/2845342 [09:05<00:00, 5724.25it/s] CPU সময়: ব্যবহারকারী 8min 14s, sys: 53.6 s, মোট: 9min 7s ওয়াল টাইম: 9min 5s

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

আমরা একটি মাল্টিপ্রসেসিং তৈরি করব পুল সঙ্গে 8 শ্রমিক এবং ব্যবহার করুন মানচিত্র প্রক্রিয়া শুরু করার ফাংশন। অগ্রগতি বার প্রদর্শন করতে, আমরা ব্যবহার করছি tqdm.

মানচিত্র ফাংশন দুটি বিভাগ নিয়ে গঠিত। প্রথমটির জন্য ফাংশন প্রয়োজন এবং দ্বিতীয়টির জন্য একটি আর্গুমেন্ট বা আর্গুমেন্টের তালিকা প্রয়োজন৷

পড়ে আরও জানুন ডকুমেন্টেশন.

%% সময় p = mp.Pool(n_workers) df['Description'] = p.map(clean_text,tqdm(df['Description']))

আউটপুট

আমরা আমাদের প্রক্রিয়াকরণের সময় প্রায় উন্নত করেছি 3X. প্রক্রিয়াকরণ সময় থেকে বাদ 9 মিনিট 5 সেকেন্ড থেকে 3 মিনিট 51 সেকেন্ড.

100% 2845342/2845342 [02:58<00:00, 135646.12it/s] CPU বার: ব্যবহারকারী 5.68 s, sys: 1.56 s, মোট: 7.23 s Wall time: 3min 51s

আমরা এখন সমান্তরাল প্রক্রিয়াকরণ করার জন্য আরেকটি পাইথন প্যাকেজ সম্পর্কে জানব। এই বিভাগে, আমরা joblib এর ব্যবহার করব সমান্তরাল এবং বিলম্বিত প্রতিলিপি করতে মানচিত্র ফাংশন.

সমান্তরাল দুটি আর্গুমেন্ট প্রয়োজন: n_jobs = 8 এবং ব্যাকএন্ড = মাল্টিপ্রসেসিং।
তারপর, আমরা যোগ করব পরিষ্কার_পাঠ্য থেকে বিলম্বিত ফাংশন.
একটি সময়ে একটি একক মান খাওয়ানোর জন্য একটি লুপ তৈরি করুন৷

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

প্রস্তাবিত: ব্যবহার করে ব্যতিক্রম হ্যান্ডলিং যোগ করুন try: এবং except:

Def পাঠ্য_সমান্তরাল_পরিষ্কার(অ্যারে): ফলাফল = সমান্তরাল(n_jobs=n_workers, backend="multiprocessing")( delayed(clean_text) (টেক্সট) উন্নত  পাঠ in tqdm(অ্যারে) ) প্রত্যাবর্তন ফল

"বিবরণ" কলাম যোগ করুন text_parallel_clean().

%%সময় df['Description'] = text_parallel_clean(df['description'])

আউটপুট

আমাদের ফাংশনটি মাল্টিপ্রসেস করার চেয়ে 13 সেকেন্ড বেশি সময় নেয় পুল। অথচ, সমান্তরাল এর চেয়ে 4 মিনিট 59 সেকেন্ড দ্রুত ক্রমিক প্রক্রিয়াকরণ।

100% 2845342/2845342 [04:03<00:00, 10514.98it/s] CPU বার: ব্যবহারকারী 44.2 s, sys: 2.92 s, মোট: 47.1 s Wall time: 4min 4s

বড় ফাইলগুলিকে ব্যাচে বিভক্ত করে এবং সমান্তরালভাবে প্রক্রিয়াকরণ করার একটি ভাল উপায় রয়েছে। চলুন শুরু করা যাক একটি ব্যাচ ফাংশন তৈরি করে যা একটি চালাবে clean_function মান একক ব্যাচে.

ব্যাচ প্রসেসিং ফাংশন

Def proc_batch(ব্যাচ): প্রত্যাবর্তন [ ক্লিন_টেক্সট (টেক্সট) উন্নত  পাঠ in ব্যাচ ]

ফাইলটিকে ব্যাচগুলিতে বিভক্ত করা

নীচের ফাংশনটি কর্মীদের সংখ্যার উপর ভিত্তি করে ফাইলটিকে একাধিক ব্যাচে বিভক্ত করবে। আমাদের ক্ষেত্রে, আমরা 8 ব্যাচ পেতে.

Def ব্যাচ ফাইল(অ্যারে, n_workers): file_len = len(array) batch_size = round(file_len / n_workers) ব্যাচ = [ array[ix:ix+batch_size] উন্নত  ix in tqdm(পরিসীমা(0, ফাইল_লেন, ব্যাচ_সাইজ)) ] প্রত্যাবর্তন ব্যাচ ব্যাচ = batch_file(df['Description'],n_workers) >>> 100% 8/8 [00:00<00:00, 280.01it/s]

সমান্তরাল ব্যাচ প্রক্রিয়াকরণ চলছে

অবশেষে, আমরা ব্যবহার করব সমান্তরাল এবং বিলম্বিত ব্যাচগুলি প্রক্রিয়া করতে।

বিঃদ্রঃ: মানগুলির একটি একক অ্যারে পেতে, আমাদের নীচে দেখানো হিসাবে তালিকা বোঝা চালাতে হবে।

%%time batch_output = সমান্তরাল(n_jobs=n_workers,backend="multiprocessing")( বিলম্বিত(proc_batch) (ব্যাচ) উন্নত  দল in tqdm(ব্যাচ) ) df['Description'] = [j উন্নত  i in ব্যাচ_আউটপুট উন্নত  j in i]

আউটপুট

আমরা প্রক্রিয়াকরণের সময় উন্নত করেছি। এই কৌশলটি জটিল তথ্য প্রক্রিয়াকরণ এবং গভীর শিক্ষার মডেল প্রশিক্ষণের জন্য বিখ্যাত।

100% 8/8 [00:00<00:00, 2.19it/s] CPU বার: ব্যবহারকারী 3.39 s, sys: 1.42 s, মোট: 4.81 s Wall time: 3min 56s

tqdm মাল্টিপ্রসেসিংকে পরবর্তী স্তরে নিয়ে যায়। এটা সহজ এবং শক্তিশালী. আমি প্রত্যেক ডেটা বিজ্ঞানীর কাছে এটি সুপারিশ করব।

চেক আউট ডকুমেন্টেশন মাল্টিপ্রসেসিং সম্পর্কে আরও জানতে।

সার্জারির process_map প্রয়োজন:

ফাংশন নাম
ডেটাফ্রেম কলাম
সর্বোচ্চ_কর্মী
chucksize ব্যাচ আকার অনুরূপ. আমরা কর্মীদের সংখ্যা ব্যবহার করে ব্যাচের আকার গণনা করব বা আপনি আপনার পছন্দের উপর ভিত্তি করে নম্বর যোগ করতে পারেন।

%% সময়
থেকে tqdm.contrib.concurrent আমদানি process_map ব্যাচ = round(len(df)/n_workers) df["Description"] = process_map( clean_text, df["Description"], max_workers=n_workers, chunksize=batch )

আউটপুট

কোডের একটি একক লাইন দিয়ে, আমরা সেরা ফলাফল পেতে পারি।

100% 2845342/2845342 [03:48<00:00, 1426320.93it/s] CPU বার: ব্যবহারকারী 7.32 s, sys: 1.97 s, মোট: 9.29 s Wall time: 3min 51s

আপনাকে একটি ভারসাম্য খুঁজে বের করতে হবে এবং আপনার ক্ষেত্রে সবচেয়ে ভালো কাজ করে এমন কৌশলটি নির্বাচন করতে হবে। এটি সিরিয়াল প্রসেসিং, সমান্তরাল বা ব্যাচ প্রসেসিং হতে পারে। আপনি যদি একটি ছোট, কম জটিল ডেটাসেটের সাথে কাজ করেন তবে সমান্তরাল প্রক্রিয়াকরণটি ব্যাকফায়ার করতে পারে।

এই মিনি-টিউটোরিয়ালে, আমরা পাইথনের বিভিন্ন প্যাকেজ এবং কৌশল সম্পর্কে শিখেছি যা আমাদের ডেটা ফাংশনগুলিকে সমান্তরালভাবে প্রক্রিয়া করার অনুমতি দেয়।

আপনি যদি শুধুমাত্র একটি ট্যাবুলার ডেটাসেট নিয়ে কাজ করেন এবং আপনার প্রক্রিয়াকরণ কর্মক্ষমতা উন্নত করতে চান, তাহলে আমি আপনাকে চেষ্টা করার পরামর্শ দেব দাস্ক, ডেটা টেবিল, এবং রেপিডস

উল্লেখ

আবিদ আলী আওয়ান (@1 আবিদালিয়াওয়ান) একজন প্রত্যয়িত ডেটা সায়েন্টিস্ট পেশাদার যিনি মেশিন লার্নিং মডেল তৈরি করতে পছন্দ করেন। বর্তমানে, তিনি মেশিন লার্নিং এবং ডেটা সায়েন্স টেকনোলজিতে বিষয়বস্তু তৈরি এবং প্রযুক্তিগত ব্লগ লেখার উপর মনোযোগ নিবদ্ধ করছেন। আবিদ টেকনোলজি ম্যানেজমেন্টে স্নাতকোত্তর ডিগ্রি এবং টেলিকমিউনিকেশন ইঞ্জিনিয়ারিংয়ে স্নাতক ডিগ্রি অর্জন করেছেন। তার দৃষ্টিভঙ্গি মানসিক অসুস্থতার সাথে সংগ্রামরত শিক্ষার্থীদের জন্য একটি গ্রাফ নিউরাল নেটওয়ার্ক ব্যবহার করে একটি AI পণ্য তৈরি করা।