ਅਪਾਚੇ ਐਰੋ ਨਾਲ ਕਾਲਮ ਡੇਟਾ ਨੂੰ ਸਟ੍ਰੀਮ ਕਰਨਾ

ਲੇਖ ਦਾ ਅਨੁਵਾਦ ਕੋਰਸ ਦੇ ਵਿਦਿਆਰਥੀਆਂ ਲਈ ਵਿਸ਼ੇਸ਼ ਤੌਰ 'ਤੇ ਤਿਆਰ ਕੀਤਾ ਗਿਆ ਸੀ ਡਾਟਾ ਇੰਜੀਨੀਅਰ.

ਪਿਛਲੇ ਕੁਝ ਹਫ਼ਤਿਆਂ ਵਿੱਚ ਸਾਡੇ ਕੋਲ ਹੈ ਨੋਂਗ ਲੀ ਵਿੱਚ ਸ਼ਾਮਲ ਕੀਤਾ ਗਿਆ ਅਪਾਚੇ ਤੀਰ ਬਾਈਨਰੀ ਸਟ੍ਰੀਮਿੰਗ ਫਾਰਮੈਟ, ਮੌਜੂਦਾ ਬੇਤਰਤੀਬੇ ਪਹੁੰਚ/IPC ਫਾਈਲ ਫਾਰਮੈਟ ਨੂੰ ਪੂਰਕ ਕਰਦਾ ਹੈ। ਸਾਡੇ ਕੋਲ Java ਅਤੇ C++ ਲਾਗੂਕਰਨ ਅਤੇ ਪਾਈਥਨ ਬਾਈਡਿੰਗ ਹਨ। ਇਸ ਲੇਖ ਵਿੱਚ, ਮੈਂ ਦੱਸਾਂਗਾ ਕਿ ਫਾਰਮੈਟ ਕਿਵੇਂ ਕੰਮ ਕਰਦਾ ਹੈ ਅਤੇ ਦਿਖਾਵਾਂਗਾ ਕਿ ਤੁਸੀਂ ਇੱਕ ਪਾਂਡਾ ਡੇਟਾਫ੍ਰੇਮ ਲਈ ਬਹੁਤ ਉੱਚ ਡੇਟਾ ਥ੍ਰਰੂਪੁਟ ਕਿਵੇਂ ਪ੍ਰਾਪਤ ਕਰ ਸਕਦੇ ਹੋ।

ਸਟ੍ਰੀਮਿੰਗ ਕਾਲਮ ਡੇਟਾ

ਇੱਕ ਆਮ ਸਵਾਲ ਜੋ ਮੈਂ ਤੀਰ ਉਪਭੋਗਤਾਵਾਂ ਤੋਂ ਪ੍ਰਾਪਤ ਕਰਦਾ ਹਾਂ, ਇੱਕ ਕਤਾਰ- ਜਾਂ ਰਿਕਾਰਡ-ਅਧਾਰਿਤ ਫਾਰਮੈਟ ਤੋਂ ਇੱਕ ਕਾਲਮ-ਅਧਾਰਿਤ ਫਾਰਮੈਟ ਵਿੱਚ ਸਾਰਣੀਬੱਧ ਡੇਟਾ ਦੇ ਵੱਡੇ ਸੈੱਟਾਂ ਨੂੰ ਮਾਈਗਰੇਟ ਕਰਨ ਦੀ ਉੱਚ ਕੀਮਤ ਹੈ। ਮਲਟੀ-ਗੀਗਾਬਾਈਟ ਡੇਟਾਸੈਟਾਂ ਲਈ, ਮੈਮੋਰੀ ਵਿੱਚ ਜਾਂ ਡਿਸਕ ਉੱਤੇ ਟ੍ਰਾਂਸਪੋਜ਼ ਕਰਨਾ ਇੱਕ ਬਹੁਤ ਵੱਡਾ ਕੰਮ ਹੋ ਸਕਦਾ ਹੈ।

ਡੇਟਾ ਨੂੰ ਸਟ੍ਰੀਮ ਕਰਨ ਲਈ, ਭਾਵੇਂ ਸਰੋਤ ਡੇਟਾ ਕਤਾਰ ਜਾਂ ਕਾਲਮ ਹੋਵੇ, ਇੱਕ ਵਿਕਲਪ ਕਤਾਰਾਂ ਦੇ ਛੋਟੇ ਬੈਚਾਂ ਨੂੰ ਭੇਜਣਾ ਹੈ, ਹਰੇਕ ਵਿੱਚ ਇੱਕ ਕਾਲਮ ਲੇਆਉਟ ਹੁੰਦਾ ਹੈ।

ਅਪਾਚੇ ਐਰੋ ਵਿੱਚ, ਟੇਬਲ ਦੇ ਹਿੱਸੇ ਨੂੰ ਦਰਸਾਉਣ ਵਾਲੇ ਇਨ-ਮੈਮੋਰੀ ਕਾਲਮ ਐਰੇ ਦੇ ਸੰਗ੍ਰਹਿ ਨੂੰ ਰਿਕਾਰਡ ਬੈਚ ਕਿਹਾ ਜਾਂਦਾ ਹੈ। ਇੱਕ ਲਾਜ਼ੀਕਲ ਟੇਬਲ ਦੇ ਇੱਕਲੇ ਡੇਟਾ ਢਾਂਚੇ ਨੂੰ ਦਰਸਾਉਣ ਲਈ, ਰਿਕਾਰਡਾਂ ਦੇ ਕਈ ਬੈਚ ਇਕੱਠੇ ਕੀਤੇ ਜਾ ਸਕਦੇ ਹਨ।

ਮੌਜੂਦਾ "ਰੈਂਡਮ ਐਕਸੈਸ" ਫਾਈਲ ਫਾਰਮੈਟ ਵਿੱਚ, ਅਸੀਂ ਫਾਈਲ ਦੇ ਅੰਤ ਵਿੱਚ ਟੇਬਲ ਸਕੀਮਾ ਅਤੇ ਬਲਾਕ ਸਥਾਨਾਂ ਵਾਲੇ ਮੈਟਾਡੇਟਾ ਨੂੰ ਰਿਕਾਰਡ ਕਰਦੇ ਹਾਂ, ਜਿਸ ਨਾਲ ਤੁਸੀਂ ਬਹੁਤ ਸਸਤੇ ਢੰਗ ਨਾਲ ਰਿਕਾਰਡਾਂ ਦੇ ਕਿਸੇ ਵੀ ਬੈਚ ਜਾਂ ਡੇਟਾ ਸੈੱਟ ਤੋਂ ਕੋਈ ਕਾਲਮ ਚੁਣ ਸਕਦੇ ਹੋ। ਸਟ੍ਰੀਮਿੰਗ ਫਾਰਮੈਟ ਵਿੱਚ, ਅਸੀਂ ਸੁਨੇਹਿਆਂ ਦੀ ਇੱਕ ਲੜੀ ਭੇਜਦੇ ਹਾਂ: ਇੱਕ ਰੂਪਰੇਖਾ, ਅਤੇ ਫਿਰ ਰਿਕਾਰਡਾਂ ਦੇ ਇੱਕ ਜਾਂ ਵੱਧ ਬੈਚ।

ਵੱਖ-ਵੱਖ ਫਾਰਮੈਟ ਕੁਝ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦਿੰਦੇ ਹਨ:

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 GB ਡੇਟਾ ਲਿਖਣਾ ਚਾਹੁੰਦੇ ਹਾਂ, ਜਿਸ ਵਿੱਚ 1024 MB ਹਰ ਇੱਕ ਦੇ ਹਿੱਸੇ ਸ਼ਾਮਲ ਹਨ। ਸ਼ੁਰੂ ਕਰਨ ਲਈ, ਆਓ 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 ਲਈ, ਜਿਵੇਂ ਕਿ ਉੱਪਰ, ਮੇਰੇ ਲੈਪਟਾਪ (ਕਵਾਡ-ਕੋਰ 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