Veergude andmete voogedastus Apache Arrow abil

Artikli tõlge on ette valmistatud spetsiaalselt kursuse üliõpilastele „Andmeinsener“.

Viimase kuu jooksul oleme me Nong Li lisanud Apache Arrow binaarse voogesitusvormingu, täiustades juba olemasolevat juhusliku juurdepääsu/IP formaati. Meil on rakendused Java ja C++ keeles ning Python jaoks sidemed. Selles artiklis selgitan, kuidas see formaat töötab ja näitan, kuidas saavutada väga kõrget andme edastuskiiruset DataFrame pandas'e jaoks.

Veergude andmete voogedastus

Küsimus, mida Arrowi kasutajatelt tihti kuulen, on seotud suurte tabelandmete komplektide üleviimise kõrgete kuludega ridakujulisest või kirjekujulisest formaadist veergude formaati. Mitme gigabaidi suuruste andmehulkade korral võib mälu või ketta peal transponimine osutuda ületamatuks ülesandeks.

Andmete voogedastamiseks, sõltumata sellest, kas algandmed on ridakujulised või veergude formaadis, on üks valik saata väikeseid ridapakette, kus igaühes on veergude struktuur.

Apache Arrowis on mälus veergude massiivide kogum, mis esindab tabeli chunk'i, nimetatakse kirje pakkumiks (record batch). Üksikute andmestruktuuride esindamiseks saab kokku koguda mitmeid kirje pakkumeid.

Olemasolevas juhusliku juurdepääsu failiformaadis kirjutame metainfot, mis sisaldab tabeli skeemi ja plokkide asukohta faili lõppu, võimaldades teil väga odavalt valida mis tahes kirje paku või mis tahes veeru andmestikust. Voogesitusformaadis saadame sõnumite seeria: skeemi ja seejärel ühe või mitu kirje pakkumist.

Erinevad formaadid näevad välja umbes nii, nagu see joonis näitab:

Andmete voogedastus PyArrow'is: rakendus

Kuidas näidata, kuidas see töötab, loodan luua näidisandmestiku, mis esindab ühte voogu:

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)
    })	

Nüüd oletame, et tahame kirjutada 1 GB andmeid, mis koosnevad 1 MB suurustest chunk'idest, kokku 1024 chunk'i. Esiteks loome 1 MB suuruse andmeraami, kus on 16 veergu:

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

df = generate_data(MEGABYTE, NCOLS)

Seejärel konverteerin need pyarrow.RecordBatch:

batch = pa.RecordBatch.from_pandas(df)

Nüüd loon ma väljundi voo, mis kirjutab mällu ja loon StreamWriter:

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

Seejärel kirjutame 1024 lõiku, mis kokku moodustavad 1 GB andmekogumist:

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

Kuna kirjutasime RAM-i, saame kogu voog ühes puhvrisse kätte:

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

In [14]: source
Out[14]: 

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

Kuna need andmed on mälus, on Arrow'i kirje pakettide lugemine zero-copy operatsioon. Avan StreamReader'i, loen andmed pyarrow.Table, ja seejärel konverteerin need DataFrame pandas:

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

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

In [18]: table
Out[18]: 

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

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

Kõik see on muidugi kena, kuid teil võivad tekkida küsimused. Kui kiiresti see toimub? Kuidas lõike suurus mõjutab pandas DataFrame'i saamise jõudlust?

Andmevoo jõudlus

Lõike suuruse vähenedes suureneb andmevoo pideva veergude struktuuri rekonstrueerimise maksumus pandas'is, mistõttu on vahemälu juurdepääsu skeemid ebaefektiivsed. Samuti on mõningaid kulusid C++ andmestruktuuride ja massiivide ning nende mälupuhvriga töötamisel.

1 MB puhul, nagu eespool mainitud, on minu sülearvuti (Quad-core Xeon E3-1505M) puhul tulemuseks:

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

Seega on efektiivne läbilaskevõime 7,75 GB/s, et taastada 1 GB suurune DataFrame 1024 1 MB lõigust. Mis juhtub, kui kasutame suurema või väiksema suurusega lõike? Tulemused on:

Jõudlus väheneb märkimisväärselt 256K kuni 64K lõikude puhul. Mind üllatas, et 1 MB suurused lõigud töötati kiiremini kui 16 MB. Tasu on teha põhjalikum uurimistöö, et mõista, kas see on normaaljaotuste mõju või on siin veel mõni tegur.

Praeguses formaadis ei pakita andmeid põhimõtteliselt, seega on mälus suurus ja ülekanal, kus andmed liiguvad, ligikaudu sama. Tulevikus võib pakkimine saada lisavalikuks.

Kokkuvõte

Veergu põhinev andmete voogedastus võib olla tõhus viis suurte andmepakettide edastamiseks veergude analüüsi tööriistadesse, näiteks pandas, väikeste blokkgeneraatorite kaudu. Andmete teenused, mis kasutavad ridadele suunatud salvestust, saavad edastada ja transponida väikeseid andmeblokke, mis sobivad paremini teie protsessori L2 ja L3 vahemällu.

Täielik kood

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)

Allikas: habr.com