הזרמת נתוני עמודות עם Apache Arrow

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

במהלך השבועות האחרונים יש לנו נונג לי נוסף ל חץ אפאצ'י פורמט סטרימינג בינארי, המשלים את פורמט הקובץ הקיים בגישה אקראית/IPC. יש לנו יישומי Java ו-C++ וקשרים לפייתון. במאמר זה, אסביר כיצד הפורמט עובד ואראה כיצד ניתן להשיג תפוקת נתונים גבוהה מאוד עבור DataFrame של pandas.

הזרמת נתוני עמודות

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

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

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

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

הפורמטים השונים נראים בערך כך:

הזרמת נתונים ב- PyArrow: אפליקציה

כדי להראות לך איך זה עובד, אני אצור מערך נתונים לדוגמה המייצג נתח זרם יחיד:

import time
import numpy as np
import pandas as pd
import pyarrow as pa

def generate_data(total_size, ncols):
    nrows = int(total_size / ncols / np.dtype('float64').itemsize)
    return pd.DataFrame({
        'c' + str(i): np.random.randn(nrows)
        for i in range(ncols)
    })	

כעת, נניח שאנו רוצים לכתוב 1 ג'יגה-בייט של נתונים, המורכבים מנתחים של 1 מגה-בייט כל אחד, עבור סך של 1024 נתחים. כדי להתחיל, בואו ניצור את מסגרת הנתונים הראשונה של 1 MB עם 16 עמודות:

KILOBYTE = 1 << 10
MEGABYTE = KILOBYTE * KILOBYTE
DATA_SIZE = 1024 * MEGABYTE
NCOLS = 16

df = generate_data(MEGABYTE, NCOLS)

ואז אני ממיר אותם ל pyarrow.RecordBatch:

batch = pa.RecordBatch.from_pandas(df)

עכשיו אני אצור זרם פלט שיכתוב ל-RAM וייצור StreamWriter:

sink = pa.InMemoryOutputStream()
stream_writer = pa.StreamWriter(sink, batch.schema)

לאחר מכן נכתוב 1024 נתחים, שיסתכמו בסופו של דבר במערך נתונים של 1GB:

for i in range(DATA_SIZE // MEGABYTE):
    stream_writer.write_batch(batch)

מכיוון שכתבנו ל-RAM, נוכל לקבל את כל הזרם במאגר אחד:

In [13]: source = sink.get_result()

In [14]: source
Out[14]: <pyarrow.io.Buffer at 0x7f2df7118f80>

In [15]: source.size
Out[15]: 1074750744

מכיוון שהנתונים האלה נמצאים בזיכרון, קריאת קבוצות של רשומות חץ היא פעולת אפס העתקה. אני פותח את StreamReader, מקריא נתונים pyarrow.Table, ולאחר מכן המר אותם ל DataFrame pandas:

In [16]: reader = pa.StreamReader(source)

In [17]: table = reader.read_all()

In [18]: table
Out[18]: <pyarrow.table.Table at 0x7fae8281f6f0>

In [19]: df = table.to_pandas()

In [20]: df.memory_usage().sum()
Out[20]: 1073741904

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

ביצועי סטרימינג

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

עבור 1 MB, כמו לעיל, במחשב הנייד שלי (Quad-core Xeon E3-1505M) מתברר:

In [20]: %timeit pa.StreamReader(source).read_all().to_pandas()
10 loops, best of 3: 129 ms per loop

מסתבר שהתפוקה האפקטיבית היא 7.75 GB/s לשחזור DataFrame של 1GB מ-1024 נתחי 1MB. מה קורה אם אנו משתמשים בנתחים גדולים יותר או קטנים יותר? אלו התוצאות:

הביצועים יורדים משמעותית מ-256K ל-64K נתחים. הופתעתי ש-1 MB chunks עובדו מהר יותר מאשר 16 MB chunks. כדאי לערוך מחקר מעמיק יותר ולהבין האם מדובר בהתפלגות נורמלית או שמא יש משהו אחר במשחק.

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

סך הכל

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

קוד מלא

import time
import numpy as np
import pandas as pd
import pyarrow as pa

def generate_data(total_size, ncols):
    nrows = total_size / ncols / np.dtype('float64').itemsize
    return pd.DataFrame({
        'c' + str(i): np.random.randn(nrows)
        for i in range(ncols)
    })

KILOBYTE = 1 << 10
MEGABYTE = KILOBYTE * KILOBYTE
DATA_SIZE = 1024 * MEGABYTE
NCOLS = 16

def get_timing(f, niter):
    start = time.clock_gettime(time.CLOCK_REALTIME)
    for i in range(niter):
        f()
    return (time.clock_gettime(time.CLOCK_REALTIME) - start) / NITER

def read_as_dataframe(klass, source):
    reader = klass(source)
    table = reader.read_all()
    return table.to_pandas()
NITER = 5
results = []

CHUNKSIZES = [16 * KILOBYTE, 64 * KILOBYTE, 256 * KILOBYTE, MEGABYTE, 16 * MEGABYTE]

for chunksize in CHUNKSIZES:
    nchunks = DATA_SIZE // chunksize
    batch = pa.RecordBatch.from_pandas(generate_data(chunksize, NCOLS))

    sink = pa.InMemoryOutputStream()
    stream_writer = pa.StreamWriter(sink, batch.schema)

    for i in range(nchunks):
        stream_writer.write_batch(batch)

    source = sink.get_result()

    elapsed = get_timing(lambda: read_as_dataframe(pa.StreamReader, source), NITER)

    result = (chunksize, elapsed)
    print(result)
    results.append(result)

מקור: www.habr.com