სვეტის მონაცემების სტრიმინგი Apache Arrow-ით

სტატიის თარგმანი მომზადდა სპეციალურად კურსის სტუდენტებისთვის მონაცემთა ინჟინერი.

ბოლო რამდენიმე კვირის განმავლობაში ჩვენ ნონგ ლი დამატება აპაჩის ისარი ორობითი ნაკადის ფორმატი, რომელიც ავსებს უკვე არსებულ შემთხვევითი წვდომის/IPC ფაილის ფორმატს. ჩვენ გვაქვს Java და C++ იმპლემენტაციები და პითონის შეკვრა. ამ სტატიაში მე აგიხსნით, თუ როგორ მუშაობს ფორმატი და გაჩვენებთ, თუ როგორ შეგიძლიათ მიაღწიოთ მონაცემთა ძალიან მაღალ გამტარუნარიანობას პანდას 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 გბ მონაცემები, რომელიც შედგება 1 მბ ცალი თითო, სულ 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 მბ-ისთვის, როგორც ზემოთ, ჩემს ლეპტოპზე (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 გბ/წმ 1 GB DataFrame-ის აღსადგენად 1024 1 MB ბლოკიდან. რა მოხდება, თუ ვიყენებთ უფრო დიდ ან პატარა ნაწილებს? აქ არის შედეგები, რომლებსაც მიიღებთ:

შესრულება მნიშვნელოვნად იკლებს 256K-დან 64K ნაწილამდე. გამიკვირდა, რომ 1 მბ ნაწილაკები უფრო სწრაფად მუშავდებოდა, ვიდრე 16 მბ. ღირს უფრო საფუძვლიანი შესწავლა და იმის გაგება, არის თუ არა ეს ნორმალური განაწილება თუ სხვა რამ არის ჩართული.

ფორმატის მიმდინარე განხორციელებისას, მონაცემები პრინციპში არ არის შეკუმშული, ამიტომ ზომა მეხსიერებაში და "მავთულზე" დაახლოებით იგივეა. შეკუმშვა შეიძლება მომავალში გახდეს ვარიანტი.

სულ

სვეტის მონაცემების სტრიმინგი შეიძლება იყოს ეფექტური გზა დიდი მონაცემთა ნაკრების გადასატანად სვეტების ანალიტიკის ინსტრუმენტებზე, როგორიცაა პანდები მცირე ნაწილებში. მონაცემთა სერვისებს, რომლებიც იყენებენ მწკრივზე ორიენტირებულ მეხსიერებას, შეუძლიათ გადაიტანონ და გადაიტანონ მონაცემთა მცირე ნაწილი, რაც უფრო მოსახერხებელია თქვენი პროცესორის 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