അപ്പാച്ചെ ആരോ ഉപയോഗിച്ച് കോളം ഡാറ്റ സ്ട്രീം ചെയ്യുന്നു

ലേഖനത്തിന്റെ വിവർത്തനം കോഴ്‌സിലെ വിദ്യാർത്ഥികൾക്കായി പ്രത്യേകം തയ്യാറാക്കിയതാണ് "ഡാറ്റ എഞ്ചിനീയർ".

കഴിഞ്ഞ കുറച്ച് ആഴ്‌ചകളായി ഞങ്ങൾക്കുണ്ട് നോങ് ലി ചേർത്തു അപ്പാച്ചെ അമ്പടയാളം ബൈനറി സ്ട്രീമിംഗ് ഫോർമാറ്റ്, നിലവിലുള്ള റാൻഡം ആക്സസ്/IPC ഫയൽ ഫോർമാറ്റ് പൂർത്തീകരിക്കുന്നു. ഞങ്ങൾക്ക് Java, C++ എന്നിവയും പൈത്തൺ ബൈൻഡിംഗുകളും ഉണ്ട്. ഈ ലേഖനത്തിൽ, ഫോർമാറ്റ് എങ്ങനെ പ്രവർത്തിക്കുന്നുവെന്ന് ഞാൻ വിശദീകരിക്കുകയും ഒരു പാണ്ടസ് ഡാറ്റാഫ്രെയിമിനായി നിങ്ങൾക്ക് എങ്ങനെ ഉയർന്ന ഡാറ്റ ത്രൂപുട്ട് നേടാമെന്ന് കാണിക്കുകയും ചെയ്യും.

സ്ട്രീമിംഗ് കോളം ഡാറ്റ

ആരോ ഉപയോക്താക്കളിൽ നിന്ന് എനിക്ക് ലഭിക്കുന്ന ഒരു സാധാരണ ചോദ്യം, ഒരു വരി അല്ലെങ്കിൽ റെക്കോർഡ്-ഓറിയന്റഡ് ഫോർമാറ്റിൽ നിന്ന് ഒരു കോളം-ഓറിയന്റഡ് ഫോർമാറ്റിലേക്ക് വലിയ സെറ്റ് ടാബ്ലർ ഡാറ്റ മൈഗ്രേറ്റ് ചെയ്യുന്നതിനുള്ള ഉയർന്ന വിലയാണ്. മൾട്ടി-ജിഗാബൈറ്റ് ഡാറ്റാസെറ്റുകൾക്ക്, മെമ്മറിയിലോ ഡിസ്കിലോ ട്രാൻസ്പോസ് ചെയ്യുന്നത് ഒരു വലിയ ജോലിയാണ്.

ഡാറ്റ സ്ട്രീം ചെയ്യുന്നതിന്, ഉറവിട ഡാറ്റ വരിയോ നിരയോ ആണെങ്കിലും, ഒരു ഓപ്‌ഷൻ, വരികളുടെ ചെറിയ ബാച്ചുകൾ അയയ്ക്കുക എന്നതാണ്, ഓരോന്നിനും ഉള്ളിൽ ഒരു കോളം ലേഔട്ട് അടങ്ങിയിരിക്കുന്നു.

അപ്പാച്ചെ ആരോയിൽ, ഒരു ടേബിൾ ചങ്കിനെ പ്രതിനിധീകരിക്കുന്ന ഇൻ-മെമ്മറി കോളം അറേകളുടെ ശേഖരത്തെ റെക്കോർഡ് ബാച്ച് എന്ന് വിളിക്കുന്നു. ഒരു ലോജിക്കൽ ടേബിളിന്റെ ഒരൊറ്റ ഡാറ്റാ ഘടനയെ പ്രതിനിധീകരിക്കുന്നതിന്, നിരവധി ബാച്ചുകളുടെ റെക്കോർഡുകൾ ശേഖരിക്കാൻ കഴിയും.

നിലവിലുള്ള "റാൻഡം ആക്‌സസ്" ഫയൽ ഫോർമാറ്റിൽ, ടേബിൾ സ്‌കീമയും ഫയലിന്റെ അവസാനം ബ്ലോക്ക് ലേഔട്ടും അടങ്ങുന്ന മെറ്റാഡാറ്റ ഞങ്ങൾ റെക്കോർഡ് ചെയ്യുന്നു, ഒരു ഡാറ്റാ സെറ്റിൽ നിന്ന് ഏത് ബാച്ച് റെക്കോർഡുകളോ കോളമോ വളരെ കുറഞ്ഞ വിലയ്ക്ക് തിരഞ്ഞെടുക്കാൻ നിങ്ങളെ അനുവദിക്കുന്നു. ഒരു സ്ട്രീമിംഗ് ഫോർമാറ്റിൽ, ഞങ്ങൾ സന്ദേശങ്ങളുടെ ഒരു പരമ്പര അയയ്‌ക്കുന്നു: ഒരു ഔട്ട്‌ലൈൻ, തുടർന്ന് ഒന്നോ അതിലധികമോ ബാച്ചുകളുടെ റെക്കോർഡുകൾ.

വ്യത്യസ്ത ഫോർമാറ്റുകൾ ഇതുപോലെ കാണപ്പെടുന്നു:

പൈആരോയിലെ സ്ട്രീമിംഗ് ഡാറ്റ: ആപ്ലിക്കേഷൻ

ഇത് എങ്ങനെ പ്രവർത്തിക്കുന്നുവെന്ന് നിങ്ങളെ കാണിക്കാൻ, ഒരൊറ്റ സ്ട്രീം ചങ്കിനെ പ്രതിനിധീകരിക്കുന്ന ഒരു ഉദാഹരണ ഡാറ്റാസെറ്റ് ഞാൻ സൃഷ്ടിക്കും:

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 നിരകളുള്ള ആദ്യത്തെ 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

ഈ ഡാറ്റ മെമ്മറിയിലുള്ളതിനാൽ, ആരോ റെക്കോർഡുകളുടെ ബാച്ചുകൾ റീഡിംഗ് സീറോ-കോപ്പി ഓപ്പറേഷനാണ്. ഞാൻ സ്ട്രീം റീഡർ തുറക്കുന്നു, ഡാറ്റ വായിക്കുന്നു 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-ന്, മുകളിൽ പറഞ്ഞതുപോലെ, എന്റെ ലാപ്‌ടോപ്പിൽ (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 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