Apache Arrow로 열 데이터 스트리밍

이 기사의 번역은 코스의 학생들을 위해 특별히 준비되었습니다. 데이터 엔지니어.

지난 몇 주 동안 우리는 농 리 에 추가 아파치 애로우 이미 존재하는 랜덤 액세스/IPC 파일 형식을 보완하는 이진 스트림 형식. Java 및 C++ 구현과 Python 바인딩이 있습니다. 이 기사에서는 형식이 작동하는 방식을 설명하고 pandas 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개의 청크에 대해 각각 1MB 청크로 구성된 1024GB의 데이터를 쓰고 싶다고 가정합니다. 먼저 1개의 열이 있는 첫 번째 16MB 데이터 프레임을 생성해 보겠습니다.

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

이 데이터는 메모리에 있기 때문에 Zero-copy 작업을 통해 Arrow 레코드의 읽기 배치를 얻습니다. 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

물론 이 모든 것이 좋지만 질문이 있을 수 있습니다. 얼마나 빨리 발생합니까? 청크 크기는 pandas DataFrame 검색 성능에 어떤 영향을 줍니까?

스트리밍 성능

스트리밍 청크 크기가 감소함에 따라 비효율적인 캐시 액세스 체계로 인해 pandas에서 연속적인 열 형식 DataFrame을 재구성하는 비용이 증가합니다. 또한 C++ 데이터 구조와 배열 및 해당 메모리 버퍼로 작업하는 데 약간의 오버헤드가 있습니다.

위와 같이 1MB의 경우 내 노트북(쿼드 코어 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 DataFrame을 복원하기 위한 유효 처리량은 1Gb/s인 것으로 나타났습니다. 크거나 작은 청크를 사용하면 어떻게 될까요? 얻은 결과는 다음과 같습니다.

성능은 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)

출처 : habr.com