Apache Arrow વડે કૉલમ ડેટા સ્ટ્રીમિંગ

લેખનો અનુવાદ કોર્સના વિદ્યાર્થીઓ માટે ખાસ તૈયાર કરવામાં આવ્યો હતો ડેટા એન્જિનિયર.

છેલ્લા કેટલાક અઠવાડિયામાં અમારી પાસે છે નોંગ લિ ઉમેરવામાં અપાચે એરો બાઈનરી સ્ટ્રીમિંગ ફોર્મેટ, હાલના રેન્ડમ એક્સેસ/આઈપીસી ફાઇલ ફોર્મેટને પૂરક બનાવે છે. અમારી પાસે Java અને C++ અમલીકરણો અને Python બાઈન્ડિંગ્સ છે. આ લેખમાં, હું સમજાવીશ કે ફોર્મેટ કેવી રીતે કાર્ય કરે છે અને બતાવીશ કે તમે કેવી રીતે પાંડા ડેટાફ્રેમ માટે ખૂબ જ ઉચ્ચ ડેટા થ્રુપુટ પ્રાપ્ત કરી શકો છો.

સ્ટ્રીમિંગ કૉલમ ડેટા

એરો યુઝર્સ તરફથી મને મળેલો એક સામાન્ય પ્રશ્ન એ પંક્તિ- અથવા રેકોર્ડ-ઓરિએન્ટેડ ફોર્મેટમાંથી કૉલમ-ઓરિએન્ટેડ ફોર્મેટમાં ટેબ્યુલર ડેટાના મોટા સેટને સ્થાનાંતરિત કરવાની ઊંચી કિંમત છે. મલ્ટી-ગીગાબાઈટ ડેટાસેટ્સ માટે, મેમરીમાં અથવા ડિસ્ક પર સ્થાનાંતરિત કરવું એ જબરજસ્ત કાર્ય હોઈ શકે છે.

ડેટાને સ્ટ્રીમ કરવા માટે, ભલે સ્રોત ડેટા પંક્તિ હોય કે કૉલમ, એક વિકલ્પ એ છે કે પંક્તિઓના નાના બૅચેસ મોકલો, જેમાં દરેક અંદર સ્તંભાકાર લેઆઉટ ધરાવે છે.

અપાચે એરોમાં, ટેબલના ભાગનું પ્રતિનિધિત્વ કરતી ઇન-મેમરી કૉલમ એરેના સંગ્રહને રેકોર્ડ બેચ કહેવામાં આવે છે. લોજિકલ ટેબલના એક ડેટા સ્ટ્રક્ચરનું પ્રતિનિધિત્વ કરવા માટે, રેકોર્ડની ઘણી બેચ એકત્રિત કરી શકાય છે.

હાલના "રેન્ડમ એક્સેસ" ફાઈલ ફોર્મેટમાં, અમે મેટાડેટાને રેકોર્ડ કરીએ છીએ જેમાં ટેબલ સ્કીમા હોય છે અને ફાઈલના અંતે બ્લોક સ્થાનો હોય છે, જેનાથી તમે ડેટા સેટમાંથી રેકોર્ડના કોઈપણ બેચ અથવા કોઈપણ કૉલમને અત્યંત સસ્તામાં પસંદ કરી શકો છો. સ્ટ્રીમિંગ ફોર્મેટમાં, અમે સંદેશાઓની શ્રેણી મોકલીએ છીએ: એક રૂપરેખા, અને પછી રેકોર્ડની એક અથવા વધુ બેચ.

વિવિધ ફોર્મેટ્સ કંઈક આના જેવા દેખાય છે:

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 GB ડેટા લખવા માંગીએ છીએ, જેમાં 1024 MB દરેક હિસ્સાનો સમાવેશ થાય છે. શરૂ કરવા માટે, ચાલો 1 કૉલમ સાથે પ્રથમ 16 MB ડેટા ફ્રેમ બનાવીએ:

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 1MB હિસ્સામાંથી 1024GB ડેટાફ્રેમ પુનઃસ્થાપિત કરવા માટે અસરકારક થ્રુપુટ 1 GB/s છે. જો આપણે મોટા અથવા નાના ભાગોનો ઉપયોગ કરીએ તો શું થાય છે? આ પરિણામો છે:

કામગીરી નોંધપાત્ર રીતે 256K થી 64K હિસ્સામાં ઘટી છે. મને આશ્ચર્ય થયું કે 1 MB હિસ્સા પર 16 MB હિસ્સા કરતાં વધુ ઝડપથી પ્રક્રિયા કરવામાં આવી હતી. વધુ ઊંડાણપૂર્વક અભ્યાસ કરવા અને સમજવા યોગ્ય છે કે શું આ એક સામાન્ય વિતરણ છે કે પછી કંઈક બીજું છે.

ફોર્મેટના વર્તમાન અમલીકરણમાં, ડેટા સૈદ્ધાંતિક રીતે સંકુચિત નથી, તેથી મેમરીમાં કદ અને "વાયરોમાં" લગભગ સમાન છે. ભવિષ્યમાં, કમ્પ્રેશન એક વધારાનો વિકલ્પ બની શકે છે.

પરિણામ

કોલમર ડેટા સ્ટ્રીમિંગ એ મોટા ડેટા સેટ્સને કોલમર એનાલિટિક્સ ટૂલ્સમાં ફીડ કરવાની અસરકારક રીત હોઈ શકે છે જેમ કે નાના ભાગોમાં પાંડા. ડેટા સેવાઓ કે જે પંક્તિ-ઓરિએન્ટેડ સ્ટોરેજનો ઉપયોગ કરે છે તે તમારા પ્રોસેસરના 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