ស្ទ្រីមទិន្នន័យជួរឈរជាមួយ 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 កំណាត់។ ដើម្បីចាប់ផ្តើម ចូរយើងបង្កើតស៊ុមទិន្នន័យ 1 MB ដំបូងដែលមាន 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 កំណាត់ ដែលនៅទីបំផុតនឹងបង្កើតសំណុំទិន្នន័យ 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

ទាំងអស់នេះពិតណាស់គឺល្អ ប៉ុន្តែអ្នកអាចមានសំណួរ។ តើរឿងនេះកើតឡើងលឿនប៉ុណ្ណា? តើទំហំកំណាត់ប៉ះពាល់ដល់ដំណើរការទាញយក 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 1GB ពី 1024 1MB កំណាត់។ តើមានអ្វីកើតឡើងប្រសិនបើយើងប្រើកំណាត់ធំជាងឬតូចជាង? ទាំងនេះគឺជាលទ្ធផល៖

ការសម្តែងធ្លាក់ចុះយ៉ាងខ្លាំងពី 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