Apache Arrow को साथ स्तम्भ डेटा स्ट्रिमिङ

लेखको अनुवाद पाठ्यक्रमका विद्यार्थीहरूको लागि विशेष गरी तयार गरिएको थियो डाटा इन्जिनियर.

पछिल्ला केही हप्ताहरूमा, हामी नोङ लि मा थपियो Apache तीर बाइनरी स्ट्रिम ढाँचा, पहिले नै अवस्थित अनियमित पहुँच/IPC फाइल ढाँचाको पूरक। हामीसँग Java र C++ कार्यान्वयन र पाइथन बाइन्डिङहरू छन्। यस लेखमा, म ढाँचाले कसरी काम गर्छ भनेर वर्णन गर्नेछु र तपाइँ कसरी पान्डा डाटाफ्रेमको लागि धेरै उच्च डेटा थ्रुपुट प्राप्त गर्न सक्नुहुन्छ भनेर देखाउनेछु।

स्तम्भ डेटा स्ट्रिमिङ

एरो प्रयोगकर्ताहरूबाट मैले प्राप्त गरेको एउटा सामान्य प्रश्न भनेको पङ्क्ति- वा रेकर्ड-उन्मुख ढाँचाबाट स्तम्भ ढाँचामा ठूला ट्याबुलर डेटासेटहरू माइग्रेट गर्ने उच्च लागत हो। बहु-गीगाबाइट डेटासेटहरूको लागि, मेमोरीमा वा डिस्कमा ट्रान्सपोजिङ भारी हुन सक्छ।

स्ट्रिमिङ डेटाका लागि, स्रोत डेटा पङ्क्ति होस् वा स्तम्भ, एउटा विकल्प भनेको पङ्क्तिहरूको सानो ब्याचहरू पठाउनु हो, प्रत्येकमा आन्तरिक रूपमा स्तम्भ लेआउट समावेश हुन्छ।

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 GB डाटा लेख्न चाहन्छौं, प्रत्येकमा 1 MB खण्डहरू, कुल 1024 खण्डहरूको लागि। पहिले, १६ स्तम्भहरूसँग पहिलो 1MB डाटा फ्रेम सिर्जना गरौं:

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

यो सबै, निस्सन्देह, राम्रो छ, तर तपाईं प्रश्न हुन सक्छ। यो कति छिटो हुन्छ? chunk साइजले पान्डा डाटाफ्रेम पुन: प्राप्ति कार्यसम्पादनलाई कसरी असर गर्छ?

स्ट्रिमिङ प्रदर्शन

स्ट्रिमिङ खण्डको आकार घट्दै जाँदा, पाण्डामा सन्निहित स्तम्भ डाटाफ्रेमको पुनर्निर्माणको लागत अकुशल क्यास पहुँच योजनाहरूको कारण बढ्छ। त्यहाँ C++ डेटा संरचनाहरू र arrays र तिनीहरूको मेमोरी बफरहरूसँग काम गर्ने केही ओभरहेड पनि छ।

माथिको रूपमा 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