Apache Arrow सह स्तंभ डेटा प्रवाहित करणे

लेखाचा अनुवाद विशेषतः अभ्यासक्रमाच्या विद्यार्थ्यांसाठी तयार केला होता डेटा अभियंता.

गेल्या काही आठवड्यांत, आम्ही नॉन्ग ली मध्ये जोडले अपाचे बाण बायनरी स्ट्रीम फॉरमॅट, आधीपासून अस्तित्वात असलेल्या यादृच्छिक प्रवेश/IPC फाइल स्वरूपनाला पूरक. आमच्याकडे Java आणि C++ अंमलबजावणी आणि Python बाइंडिंग आहेत. या लेखात, मी फॉरमॅट कसे कार्य करते हे स्पष्ट करेन आणि पांडा डेटाफ्रेमसाठी तुम्ही खूप उच्च डेटा थ्रूपुट कसे मिळवू शकता ते मी दाखवीन.

स्तंभ डेटा प्रवाहित करणे

एरो वापरकर्त्यांकडून मला पडलेला एक सामान्य प्रश्न म्हणजे मोठ्या सारणी डेटासेटला पंक्ती- किंवा रेकॉर्ड-ओरिएंटेड फॉरमॅटमधून कॉलम फॉरमॅटमध्ये स्थलांतरित करण्याची उच्च किंमत. मल्टी-गीगाबाइट डेटासेटसाठी, मेमरीमध्ये किंवा डिस्कवर ट्रान्सपोज करणे जबरदस्त असू शकते.

प्रवाहित डेटासाठी, स्त्रोत डेटा पंक्ती किंवा स्तंभ असो, एक पर्याय म्हणजे पंक्तींच्या लहान बॅच पाठवणे, प्रत्येकामध्ये अंतर्गत स्तंभ लेआउट आहे.

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 MB भागांचा समावेश असलेला 1024 GB डेटा लिहायचा आहे. प्रथम, 1 स्तंभांसह पहिली 16MB डेटा फ्रेम तयार करूया:

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

हे सर्व नक्कीच चांगले आहे, परंतु तुम्हाला प्रश्न असू शकतात. ते किती वेगाने घडते? भाग आकार पांडा डेटाफ्रेम पुनर्प्राप्ती कार्यक्षमतेवर कसा परिणाम करतो?

प्रवाहित कार्यप्रदर्शन

स्ट्रीमिंग भागाचा आकार कमी होत असताना, अकार्यक्षम कॅशे ऍक्सेस योजनांमुळे पांडांमध्ये संलग्न स्तंभीय डेटाफ्रेमची पुनर्रचना करण्याची किंमत वाढते. C++ डेटा स्ट्रक्चर्स आणि अॅरे आणि त्यांच्या मेमरी बफरसह काम करण्यापासून काही ओव्हरहेड देखील आहे.

वरीलप्रमाणे 1 MB साठी, माझ्या लॅपटॉपवर (क्वाड-कोर Xeon E3-1505M) असे दिसून आले:

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

असे दिसून आले की 7.75 1 MB भागांमधून 1024 GB डेटाफ्रेम पुनर्संचयित करण्यासाठी प्रभावी थ्रूपुट 1 Gb/s आहे. आपण मोठे किंवा लहान भाग वापरल्यास काय होईल? तुम्हाला मिळणारे परिणाम येथे आहेत:

कार्यप्रदर्शन 256K ते 64K भागांमध्ये लक्षणीयरीत्या घसरते. मला आश्चर्य वाटले की 1MB भागांवर 16MB भागांपेक्षा जलद प्रक्रिया केली गेली. अधिक सखोल अभ्यास करणे आणि हे सामान्य वितरण आहे की आणखी काहीतरी गुंतलेले आहे हे समजून घेणे योग्य आहे.

स्वरूपाच्या सध्याच्या अंमलबजावणीमध्ये, डेटा तत्त्वतः संकुचित केला जात नाही, म्हणून मेमरीमधील आकार आणि "वायरवर" अंदाजे समान आहे. भविष्यात कॉम्प्रेशन हा एक पर्याय बनू शकतो.

परिणाम

स्तंभ डेटा प्रवाहित करणे हा मोठ्या डेटासेटला स्तंभ विश्लेषण साधनांमध्ये हस्तांतरित करण्याचा एक प्रभावी मार्ग असू शकतो जसे की लहान भागांमध्ये पांडा. डेटा सेवा ज्या रो-ओरिएंटेड स्टोरेज वापरतात त्या तुमच्या प्रोसेसरच्या 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