עיבוד מקבילי קובץ גדול בפייתון

הועלה מחדש על ידי אפלטון

עוקב: 0

עיבוד מקביל לקובץ גדול ב-Python
תמונה מאת המחבר

לעיבוד מקביל, אנו מחלקים את המשימה שלנו ליחידות משנה. זה מגדיל את מספר המשרות המעובדות על ידי התוכנית ומפחית את זמן העיבוד הכולל.

לדוגמה, אם אתה עובד עם קובץ CSV גדול וברצונך לשנות עמודה בודדת. אנו נזין את הנתונים כמערך לפונקציה, והיא תעבד במקביל מספר ערכים בבת אחת על סמך מספר הזמינים עובדים. עובדים אלה מבוססים על מספר הליבות בתוך המעבד שלך.

הערה: שימוש בעיבוד מקביל במערך נתונים קטן יותר לא ישפר את זמן העיבוד.

בבלוג זה, נלמד כיצד להפחית את זמן העיבוד בקבצים גדולים באמצעות עיבוד רב, ג'וביב, ו tqdm חבילות פייתון. זהו מדריך פשוט שיכול לחול על כל קובץ, מסד נתונים, תמונה, וידאו ואודיו.

הערה: אנו משתמשים במחברת Kaggle עבור הניסויים. זמן העיבוד יכול להשתנות ממכונה למכונה.

אנו נשתמש ב- תאונות בארה"ב (2016 - 2021) מערך נתונים מ-Kaggle המורכב מ-2.8 מיליון רשומות ו-47 עמודות.

אנחנו נייבא multiprocessing, joblib, ו tqdm ל עיבוד מקביל, pandas ל בליעת נתונים, ו re, nltk, ו string ל עיבוד טקסט.

# מחשוב מקביל
לייבא עיבוד רב as mp
החל מ- ג'וביב לייבא מקביל, מושהה
החל מ- tqdm.notebook לייבא tqdm # הכנסת נתונים 
לייבא דובי פנדה as pd # עיבוד טקסט 
לייבא re החל מ- nltk.corpus לייבא מילות עצור
לייבא מחרוזת

לפני שנקפוץ ישר פנימה, בואו נתחיל n_workers על ידי הכפלה cpu_count(). כפי שאתה יכול לראות, יש לנו 8 עובדים.

n_workers = 2 * mp.cpu_count() print(f"{n_workers} עובדים זמינים") >>> 8 עובדים זמינים

בשלב הבא, נבלע קבצי CSV גדולים באמצעות ה דובי פנדה read_csv פוּנקצִיָה. לאחר מכן, הדפס את צורת מסגרת הנתונים, שם העמודות וזמן העיבוד.

הערה: פונקציית הקסם של יופיטר %%time יכול להציג זמני מעבד ו זמן קיר בסוף התהליך.

%%זְמַן
file_name="../input/us-accidents/US_Accidents_Dec21_updated.csv"
df = pd.read_csv(file_name) print(f"Shape:{df.shape}nnשמות עמודות:n{df.columns}n")

תְפוּקָה

Shape:(2845342, 47) שמות עמודות: אינדקס(['ID', 'Severity', 'Start_Time', 'End_Time', 'Start_Lat', 'Start_Lng', 'End_Lat', 'End_Lng', 'Distance(mi) ', 'תיאור', 'מספר', 'רחוב', 'צד', 'עיר', 'מחוז', 'מדינה', 'מיקוד', 'מדינה', 'אזור זמן', 'קוד_שדה תעופה', 'חותמת_מזג אוויר', 'טמפרטורה(F)', 'Wind_Chill(F)', 'Hummidity(%)', 'Pressure(in)', 'Visibility(mi)', 'Wind_Direction', 'Wind_Speed(mph)', 'Precipitation(in) )', 'תנאי_מזג אוויר', 'שירותים', 'בליטה', 'חצייה', 'פניית דרך', 'צומת', 'אין_יציאה', 'רכבת', 'כיכר', 'תחנה', 'עצירה', 'הרגעת_תנועה' , 'Signal_Traffic', 'Turning_Loop', 'Sunrise_Sunset', 'Civil_Twilight', 'Nautical_Twilight', 'Astronomical_Twilight'],
dtype='object') זמני מעבד: משתמש 33.9 שניות, מערכת הפעלה: 3.93 שניות, סה"כ: 37.9 שניות
זמן קיר: 46.9 שניות

השמיים clean_text היא פונקציה פשוטה לעיבוד וניקוי הטקסט. נקבל אנגלית מילות עצור באמצעות nltk.copus השתמש בו כדי לסנן מילות עצירה משורת הטקסט. לאחר מכן, נסיר מהמשפט תווים מיוחדים ורווחים נוספים. זו תהיה פונקציית הבסיס לקביעת זמן העיבוד עבורו סידורי, במקביל, ו קְבוּצָה מעבד.

def טקסט נקי(טֶקסט): # הסר מילות עצירה stops = stopwords.words("אנגלית") text = " ".join([word ל מילה in text.split() if מילה לֹא in עוצרים]) # הסר תווים מיוחדים text = text.translate(str.maketrans('', '', string.punctuation)) # הסרת הרווחים הנוספים text = re.sub(' +',' ', text) לַחֲזוֹר טֶקסט

לעיבוד סדרתי, נוכל להשתמש בפנדות .apply() פונקציה, אבל אם אתה רוצה לראות את סרגל ההתקדמות, עליך להפעיל tqdm ל דובי פנדה ואז השתמש ב- .progress_apply() פונקציה.

אנחנו הולכים לעבד את 2.8 מיליון הרשומות ולשמור את התוצאה בחזרה לעמודה "תיאור".

%%זְמַן
tqdm.pandas() df['Description'] = df['Description'].progress_apply(clean_text)

תְפוּקָה

זה לקח 9 דקות ו-5 שניות עבור high-end מעבד לתהליך סדרתי 2.8 מיליון שורות.

100% 2845342/2845342 [09:05<00:00, 5724.25it/s] זמני מעבד: משתמש 8 דקות 14 שניות, מערכת: 53.6 שניות, סה"כ: 9 דקות 7 שניות
זמן קיר: 9 דקות 5 שניות

ישנן דרכים שונות לעיבוד מקביל של הקובץ, ואנחנו הולכים ללמוד על כולן. ה multiprocessing היא חבילת python מובנית המשמשת בדרך כלל לעיבוד מקבילי של קבצים גדולים.

ניצור ריבוי עיבודים ברכה עם עובד 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] זמני מעבד: משתמש 5.68 שניות, מערכת הפעלה: 1.56 שניות, סה"כ: 7.23 שניות
זמן קיר: 3 דקות 51 שניות

כעת נלמד על חבילת Python נוספת לביצוע עיבוד מקביל. בחלק זה, נשתמש ב-joblib's מקביל ו מעוכב לשכפל את מַפָּה פונקציה.

המקביל דורש שני ארגומנטים: n_jobs = 8 ו-backend = multiprocessing.
לאחר מכן, נוסיף טקסט נקי אל ה מעוכב פונקציה.
צור לולאה כדי להזין ערך בודד בכל פעם.

התהליך שלהלן הוא די גנרי, ואתה יכול לשנות את הפונקציה והמערך שלך בהתאם לצרכים שלך. השתמשתי בו כדי לעבד אלפי קבצי אודיו ווידאו ללא כל בעיה.

מומלץ: הוסף טיפול חריג באמצעות try: ו except:

def text_parallel_clean(מערך): תוצאה = Parallel(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] זמני מעבד: משתמש 44.2 שניות, מערכת הפעלה: 2.92 שניות, סה"כ: 47.1 שניות
זמן קיר: 4 דקות 4 שניות

יש דרך טובה יותר לעבד קבצים גדולים על ידי פיצול אותם לקבוצות ועיבודם במקביל. נתחיל ביצירת פונקציית אצווה שתפעיל את a clean_function על אצווה אחת של ערכים.

פונקציית עיבוד אצווה

def proc_batch(קבוצה): לַחֲזוֹר [ clean_text(טקסט) ל טֶקסט in קבוצה ]

פיצול הקובץ לקבוצות

הפונקציה שלהלן תפצל את הקובץ למספר אצוות על סמך מספר העובדים. במקרה שלנו, אנחנו מקבלים 8 אצוות.

def batch_file(array,n_workers): file_len = len(array) batch_size = round(file_len / n_workers) batches = [ array[ix:ix+batch_size] ל ix in tqdm(range(0, file_len, batch_size)) ] לַחֲזוֹר batches batches = batch_file(df['Description'],n_workers) >>> 100% 8/8 [00:00<00:00, 280.01it/s]

הפעלת עיבוד אצווה מקביל

לבסוף, נשתמש מקביל ו מעוכב לעבד קבוצות.

הערה: כדי לקבל מערך יחיד של ערכים, עלינו להפעיל הבנת רשימה כפי שמוצג להלן.

%%זְמַן
batch_output = Parallel(n_jobs=n_workers,backend="multiprocessing")( delayed(proc_batch) (אצווה) ל קְבוּצָה in tqdm(batches) ) df['Description'] = [j ל i in batch_output ל j in i]

תְפוּקָה

שיפרנו את זמן העיבוד. טכניקה זו מפורסמת בעיבוד נתונים מורכבים והכשרת מודלים של למידה עמוקה.

100% 8/8 [00:00<00:00, 2.19it/s] זמני מעבד: משתמש 3.39 שניות, מערכת הפעלה: 1.42 שניות, סה"כ: 4.81 שניות
זמן קיר: 3 דקות 56 שניות

tqdm לוקח ריבוי עיבודים לשלב הבא. זה פשוט ועוצמתי. אמליץ עליו לכל מדען נתונים.

בדוק תיעוד למידע נוסף על ריבוי עיבודים.

השמיים process_map דורש:

שם פונקציה
עמודת Dataframe
max_workers
גודל chucksize דומה לגודל אצווה. אנו נחשב את גודל האצווה לפי מספר העובדים או שאתה יכול להוסיף את המספר לפי העדפתך.

%%זְמַן
החל מ- tqdm.contrib.concurrent לייבא מפת התהליך
batch = 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] זמני מעבד: משתמש 7.32 שניות, מערכת הפעלה: 1.97 שניות, סה"כ: 9.29 שניות
זמן קיר: 3 דקות 51 שניות

אתה צריך למצוא איזון ולבחור את הטכניקה המתאימה ביותר למקרה שלך. זה יכול להיות עיבוד סדרתי, מקביל או עיבוד אצווה. העיבוד המקביל עלול לגרום לאחור אם אתה עובד עם מערך נתונים קטן יותר ופחות מורכב.

במיני-מדריך זה, למדנו על חבילות וטכניקות שונות של Python המאפשרות לנו לעבד במקביל את פונקציות הנתונים שלנו.

אם אתה עובד רק עם מערך נתונים טבלאי ורוצה לשפר את ביצועי העיבוד שלך, אז אני מציע לך לנסות לוּחַ מַחווָנִים, טבלת נתונים, ו אֶשֶׁד

התייחסות

עביד עלי אוואן (@1abidaliawan) הוא איש מקצוע מוסמך של מדען נתונים שאוהב לבנות מודלים של למידת מכונה. נכון לעכשיו, הוא מתמקד ביצירת תוכן וכתיבת בלוגים טכניים על למידת מכונה וטכנולוגיות מדעי נתונים. עביד הוא בעל תואר שני בניהול טכנולוגיה ותואר ראשון בהנדסת טלקומוניקציה. החזון שלו הוא לבנות מוצר בינה מלאכותית באמצעות רשת עצבית גרפית עבור תלמידים הנאבקים במחלות נפש.