Пренесување податоци од колона со Apache Arrow

Преводот на статијата е подготвен специјално за студентите на курсот Инженер за податоци.

Во текот на изминатите неколку недели, ние Нонг Ли додадена на Апачи стрелка бинарен стрим формат, дополнувајќи го веќе постоечкиот формат на датотека со случаен пристап/IPC. Имаме Java и C++ имплементации и Python врзувања. Во оваа статија, ќе објаснам како функционира форматот и ќе покажам како можете да постигнете многу висока пропусност на податоци за DataFrame на панди.

Пренос на податоци од колона

Честото прашање што го добивам од корисниците на Arrow е високата цена за мигрирање на големи табеларни збирки на податоци од формат ориентиран кон ред или запис во формат на колона. За збирки на податоци од повеќе гигабајти, транспонирањето во меморија или на диск може да биде огромно.

За пренос на податоци, без разлика дали изворните податоци се ред или колона, една опција е да се испратат мали серии од редови, од кои секоја ќе содржи распоред на колона внатре.

Во 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 со 16 колони:

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 парчиња, кои на крајот ќе сочинуваат база на податоци од 1 GB:

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

Сето ова, се разбира, е добро, но можеби имате прашања. Колку брзо се случува? Како големината на делот влијае на перформансите на пандите за преземање DataFrame?

Изведба на стриминг

Како што се намалува големината на стриминг парчето, трошоците за реконструкција на соседна колонообразна DataFrame кај пандите се зголемуваат поради неефикасните шеми за пристап во кешот. Исто така, има некои трошоци за работа со C++ податочни структури и низи и нивните мемориски бафери.

За 1 MB како погоре, на мојот лаптоп (Quad-core Xeon E3-1505M) излегува:

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

Излегува дека ефективната пропусност е 7.75 Gb/s за враќање на DataFrame од 1 GB од 1024 парчиња од 1 MB. Што се случува ако користиме поголеми или помали парчиња? Еве ги резултатите што ги добивате:

Перформансите значително се намалуваат од 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