🏟️ 👎🏻 👿 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 चंक्स के लिए। सबसे पहले, आइए १६ स्तंभों के साथ पहला १ एमबी डाटा फ्रेम बनाएँ:

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 एमबी के लिए, जैसा कि ऊपर उल्लेख किया गया है, मेरे लैपटॉप पर (क्वाड-कोर एक्सोन E3-1505M) यह पता चला है:

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

यह पता चलता है कि प्रभावी बैंडविड्थ 7.75 जीबी / एस है, जो 1 जीबी डेटा का नाम 1 एमबी के प्रत्येक 1 एमबी के 1 जीबी डेटा को बहाल करने के लिए है। यदि हम बड़े या छोटे विखंडू का उपयोग करते हैं तो क्या होगा? यहाँ परिणाम हैं:

प्रदर्शन 256K से 64K विखंडू में महत्वपूर्ण रूप से गिरता है। मुझे आश्चर्य हुआ कि 16 एमबी की तुलना में 1 एमबी का हिस्सा तेजी से संसाधित हुआ। अधिक गहन अध्ययन करना और समझना आवश्यक है कि क्या यह एक सामान्य वितरण है या यदि कुछ और इसे प्रभावित करता है।

प्रारूप के वर्तमान कार्यान्वयन में, डेटा सिद्धांत रूप में संकुचित नहीं है, इसलिए मेमोरी और तारों में आकार लगभग समान है। भविष्य में, संपीड़न एक अतिरिक्त विकल्प बन सकता है।

संपूर्ण

स्‍तंभ कॉलम डेटा को बड़े डेटा सेट को कॉलम विश्लेषण टूल, जैसे पांडा, छोटे विखंडू का उपयोग करके स्थानांतरित करने का एक प्रभावी तरीका हो सकता है। लाइन-ओरिएंटेड स्टोरेज का उपयोग करने वाली डेटा सेवाएँ उन छोटे डेटा विखंडनों को स्थानांतरित और स्थानांतरित कर सकती हैं जो आपके प्रोसेसर के 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)

Apache Arrow का उपयोग करके कॉलम डेटा को स्ट्रीम करना

स्तंभ डेटा स्ट्रीमिंग

PyArrow स्ट्रीमिंग: एप्लिकेशन

स्ट्रीमिंग प्रदर्शन

संपूर्ण

More articles: