C++과 Apache Arrow를 활용한 컬럼 기반 데이터 처리 성능 향상

   sudo apt update
   sudo apt install -y libarrow-dev

   conda install -c conda-forge arrow-cpp

   git clone https://github.com/apache/arrow.git
   cd arrow/cpp
   mkdir build && cd build
   cmake .. -DARROW_BUILD_STATIC=ON -DARROW_BUILD_SHARED=ON
   make -j$(nproc)
   sudo make install

cmake_minimum_required(VERSION 3.14)
project(ArrowExample)

find_package(Arrow REQUIRED)

add_executable(arrow_example main.cpp)
target_link_libraries(arrow_example PRIVATE Arrow::arrow)

#include <arrow/api.h>
#include <iostream>

int main() {
    // Int32 타입의 컬럼을 생성하기 위한 Builder
    arrow::Int32Builder builder;

    // 데이터 추가
    builder.Append(10);
    builder.Append(20);
    builder.Append(30);

    // Array로 변환
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // 데이터 출력
    auto int_array = std::static_pointer_cast<arrow::Int32Array>(array);
    for (int i = 0; i < int_array->length(); i++) {
        std::cout << "Value at index " << i << ": " << int_array->Value(i) << std::endl;
    }

    return 0;
}

Value at index 0: 10  
Value at index 1: 20  
Value at index 2: 30  

#include <arrow/api.h>
#include <iostream>

int main() {
    // Double 타입의 컬럼 데이터 생성
    arrow::DoubleBuilder builder;
    builder.Append(3.5);
    builder.Append(4.8);
    builder.Append(2.1);
    builder.Append(7.9);

    // Array 변환
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // 연산 수행 (합계 계산)
    auto double_array = std::static_pointer_cast<arrow::DoubleArray>(array);
    double sum = 0.0;
    for (int i = 0; i < double_array->length(); i++) {
        sum += double_array->Value(i);
    }

    std::cout << "Sum: " << sum << std::endl;

    return 0;
}

Sum: 18.3

#include <arrow/api.h>
#include <arrow/io/file.h>
#include <arrow/csv/api.h>
#include <iostream>

int main() {
    // CSV 파일 읽기 위한 파일 입력 스트림 생성
    std::shared_ptr<arrow::io::ReadableFile> input_file;
    arrow::io::ReadableFile::Open("data.csv", arrow::default_memory_pool(), &input_file);

    // CSV 읽기 옵션 설정
    auto read_options = arrow::csv::ReadOptions::Defaults();
    auto parse_options = arrow::csv::ParseOptions::Defaults();
    auto convert_options = arrow::csv::ConvertOptions::Defaults();

    // CSV 파일을 Arrow 테이블로 변환
    auto maybe_reader = arrow::csv::TableReader::Make(arrow::default_memory_pool(), input_file, read_options, parse_options, convert_options);
    std::shared_ptr<arrow::Table> table;
    maybe_reader->Read().Value(&table);

    // 변환된 데이터 출력
    std::cout << "CSV Loaded as Arrow Table with " << table->num_rows() << " rows and " << table->num_columns() << " columns." << std::endl;

    return 0;
}

#include <arrow/api.h>
#include <arrow/io/memory.h>
#include <arrow/json/api.h>
#include <iostream>

int main() {
    // JSON 데이터 예제
    std::string json_data = R"([
        {"id": 1, "name": "Alice", "age": 30},
        {"id": 2, "name": "Bob", "age": 25},
        {"id": 3, "name": "Charlie", "age": 35}
    ])";

    // JSON 데이터 읽기 위한 메모리 버퍼 생성
    auto buffer = std::make_shared<arrow::Buffer>(json_data);
    auto input = std::make_shared<arrow::io::BufferReader>(buffer);

    // JSON 변환 옵션 설정
    auto read_options = arrow::json::ReadOptions::Defaults();
    auto parse_options = arrow::json::ParseOptions::Defaults();

    // JSON을 Arrow 테이블로 변환
    auto maybe_reader = arrow::json::TableReader::Make(arrow::default_memory_pool(), input, read_options, parse_options);
    std::shared_ptr<arrow::Table> table;
    maybe_reader->Read().Value(&table);

    // 변환된 데이터 출력
    std::cout << "JSON Loaded as Arrow Table with " << table->num_rows() << " rows and " << table->num_columns() << " columns." << std::endl;

    return 0;
}

#include <arrow/api.h>
#include <arrow/io/file.h>
#include <parquet/arrow/writer.h>
#include <iostream>

int main() {
    // 간단한 Int32 컬럼 데이터 생성
    arrow::Int32Builder builder;
    builder.Append(10);
    builder.Append(20);
    builder.Append(30);
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // Arrow 테이블 생성
    auto schema = arrow::schema({arrow::field("values", arrow::int32())});
    auto table = arrow::Table::Make(schema, {array});

    // Parquet 파일 저장
    std::shared_ptr<arrow::io::FileOutputStream> outfile;
    arrow::io::FileOutputStream::Open("output.parquet", &outfile);
    parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), outfile, 1);

    std::cout << "Arrow Table saved as Parquet file." << std::endl;

    return 0;
}

#include <arrow/api.h>
#include <iostream>

int main() {
    // Int32 타입의 Arrow Array 생성
    arrow::Int32Builder builder;
    builder.Append(10);
    builder.Append(20);
    builder.Append(30);
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // 버퍼를 직접 접근하여 데이터 확인 (Zero-copy 방식)
    auto int_array = std::static_pointer_cast<arrow::Int32Array>(array);
    auto data_buffer = int_array->values();

    // 버퍼의 포인터를 사용하여 데이터 접근
    const int32_t* raw_data = reinterpret_cast<const int32_t*>(data_buffer->data());

    std::cout << "Accessing data without copy:" << std::endl;
    for (int i = 0; i < int_array->length(); i++) {
        std::cout << "Value at index " << i << ": " << raw_data[i] << std::endl;
    }

    return 0;
}

Accessing data without copy:
Value at index 0: 10
Value at index 1: 20
Value at index 2: 30

#include <arrow/io/memory.h>
#include <arrow/io/file.h>
#include <iostream>

int main() {
    // 파일을 메모리 매핑하여 읽기
    std::shared_ptr<arrow::io::MemoryMappedFile> mmap_file;
    arrow::io::MemoryMappedFile::Open("data.arrow", arrow::io::FileMode::READ, &mmap_file);

    // 파일 크기 확인
    int64_t size = mmap_file->size();
    std::shared_ptr<arrow::Buffer> buffer;
    mmap_file->Read(size, &buffer);

    std::cout << "Memory-mapped file size: " << size << " bytes" << std::endl;
    std::cout << "First 10 bytes of data: " << buffer->ToString().substr(0, 10) << std::endl;

    return 0;
}

Memory-mapped file size: 10240 bytes
First 10 bytes of data: [binary data...]

#include <arrow/api.h>
#include <iostream>
#include <thread>
#include <vector>

// 데이터 처리를 수행하는 함수
void ProcessColumn(std::shared_ptr<arrow::Int32Array> column, int start, int end, int thread_id) {
    int sum = 0;
    for (int i = start; i < end; ++i) {
        sum += column->Value(i);
    }
    std::cout << "Thread " << thread_id << ": Sum = " << sum << std::endl;
}

int main() {
    // 데이터 생성 (컬럼 기반)
    arrow::Int32Builder builder;
    for (int i = 1; i <= 1000; i++) {
        builder.Append(i);
    }
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    auto int_array = std::static_pointer_cast<arrow::Int32Array>(array);
    int num_threads = 4;  // 4개의 스레드 사용
    int chunk_size = int_array->length() / num_threads;

    std::vector<std::thread> threads;

    // 스레드를 생성하여 데이터 처리
    for (int i = 0; i < num_threads; i++) {
        int start = i * chunk_size;
        int end = (i == num_threads - 1) ? int_array->length() : (i + 1) * chunk_size;
        threads.emplace_back(ProcessColumn, int_array, start, end, i);
    }

    // 모든 스레드 종료 대기
    for (auto& th : threads) {
        th.join();
    }

    return 0;
}

Thread 0: Sum = 125250
Thread 1: Sum = 375250
Thread 2: Sum = 625250
Thread 3: Sum = 875250

#include <arrow/api.h>
#include <iostream>
#include <omp.h>

int main() {
    // 컬럼 데이터 생성
    arrow::Int32Builder builder;
    for (int i = 1; i <= 1000000; i++) {
        builder.Append(i);
    }
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    auto int_array = std::static_pointer_cast<arrow::Int32Array>(array);
    int sum = 0;

    // OpenMP를 이용한 병렬 처리
    #pragma omp parallel for reduction(+:sum)
    for (int i = 0; i < int_array->length(); i++) {
        sum += int_array->Value(i);
    }

    std::cout << "Total sum using OpenMP: " << sum << std::endl;

    return 0;
}

Total sum using OpenMP: 500000500000

#include <arrow/api.h>
#include <arrow/python/pyarrow.h>
#include <pybind11/pybind11.h>

namespace py = pybind11;

py::object ConvertArrowToPandas(std::shared_ptr<arrow::Table> table) {
    // Apache Arrow 테이블을 Pandas DataFrame으로 변환
    auto maybe_py_table = arrow::py::wrap_table(table);
    return maybe_py_table;
}

import pyarrow as pa
import pyarrow.parquet as pq
import pandas as pd

# Parquet 파일을 Arrow 테이블로 변환
table = pq.read_table("data.parquet")

# Arrow 테이블을 Pandas DataFrame으로 변환 (Zero-copy)
df = table.to_pandas()
print(df.head())

import pyspark.sql.functions as F
from pyspark.sql import SparkSession
import pyarrow as pa

# Spark 세션 생성
spark = SparkSession.builder.appName("ArrowSpark").getOrCreate()

# Arrow 기반 데이터프레임 활성화
spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")

# 데이터 로드
df = spark.read.csv("data.csv", header=True)

# Pandas UDF를 사용한 고속 연산
@F.pandas_udf("double")
def add_one(series: pd.Series) -> pd.Series:
    return series + 1

df = df.withColumn("new_col", add_one(F.col("numeric_col")))
df.show()

#include <arrow/api.h>
#include <arrow/io/file.h>
#include <parquet/arrow/writer.h>

int main() {
    // Int32 컬럼 생성
    arrow::Int32Builder builder;
    builder.Append(100);
    builder.Append(200);
    builder.Append(300);
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // Arrow 테이블 생성
    auto schema = arrow::schema({arrow::field("values", arrow::int32())});
    auto table = arrow::Table::Make(schema, {array});

    // Parquet 파일 저장
    std::shared_ptr<arrow::io::FileOutputStream> outfile;
    arrow::io::FileOutputStream::Open("output.parquet", &outfile);
    parquet::arrow::WriteTable(*table, arrow::default_memory_pool(), outfile, 1);

    std::cout << "Arrow Table saved as Parquet file." << std::endl;

    return 0;
}

#include <arrow/api.h>
#include <benchmark/benchmark.h>

static void ArrowColumnProcessing(benchmark::State& state) {
    // Int32 컬럼 데이터 생성
    arrow::Int32Builder builder;
    for (int i = 0; i < state.range(0); i++) {
        builder.Append(i);
    }
    std::shared_ptr<arrow::Array> array;
    builder.Finish(&array);

    // 컬럼 데이터 처리 (합계 계산)
    for (auto _ : state) {
        auto int_array = std::static_pointer_cast<arrow::Int32Array>(array);
        int sum = 0;
        for (int i = 0; i < int_array->length(); i++) {
            sum += int_array->Value(i);
        }
        benchmark::DoNotOptimize(sum);
    }
}

BENCHMARK(ArrowColumnProcessing)->Range(1<<10, 1<<20);
BENCHMARK_MAIN();

#include <arrow/memory_pool.h>
#include <iostream>

int main() {
    // 기본 메모리 풀 가져오기
    arrow::MemoryPool* pool = arrow::default_memory_pool();

    // 메모리 사용량 출력
    std::cout << "Memory allocated: " << pool->bytes_allocated() << " bytes" << std::endl;

    return 0;
}

#include <arrow/csv/api.h>
#include <arrow/io/file.h>
#include <parquet/arrow/reader.h>
#include <parquet/arrow/writer.h>
#include <chrono>
#include <iostream>

void TestParquetReadPerformance() {
    auto start = std::chrono::high_resolution_clock::now();

    std::shared_ptr<arrow::io::ReadableFile> input_file;
    arrow::io::ReadableFile::Open("data.parquet", arrow::default_memory_pool(), &input_file);

    std::shared_ptr<arrow::Table> table;
    parquet::arrow::FileReaderBuilder builder;
    builder.Open(input_file);
    std::shared_ptr<parquet::arrow::FileReader> reader;
    builder.Build(&reader);
    reader->ReadTable(&table);

    auto end = std::chrono::high_resolution_clock::now();
    std::chrono::duration<double> elapsed = end - start;
    std::cout << "Parquet Read Time: " << elapsed.count() << " seconds" << std::endl;
}

int main() {
    TestParquetReadPerformance();
    return 0;
}