Apache ArrowとJava: ライトニングスピードのビッグデータ転送

原文(投稿日：2020/05/23)へのリンク

その性質上、ビッグデータは大きすぎて単一のマシンに収まりません。データセットは複数のマシンに分割する必要があります。各パーティションは1つのプライマリマシンに割り当てられ、オプションでバックアップが割り当てられます。したがって、すべてのマシンが複数のパーティションを保持しています。ほとんどのビッグデータフレームワークは、パーティションをマシンに割り当てるためにランダムな戦略を使用しています。各計算ジョブが1つのパーティションを使用する場合、この戦略により、クラスター全体の計算負荷が適切に分散されます。ただし、ジョブで複数のパーティションが必要な場合は、他のマシンからパーティションをフェッチする必要がある可能性が高くなります。データを転送すると、常にパフォーマンスが低下します。

Apache Arrowは、データにクロス言語、クロスプラットフォーム、カラム型のインメモリデータ形式を提供します。データは各プラットフォームとプログラミング言語で同じバイト形式で表されるため、シリアル化の必要がなくなります。この共通フォーマットにより、ビッグデータシステムでのゼロコピーデータ転送が可能になり、データ転送のパフォーマンスへの影響を最小限に抑えることができます。

この記事の目的は、Apache Arrowを紹介し、Apache Arrow Javaライブラリーの基本概念を理解することです。この記事に付随するソースコードはここにあります。

通常、データ転送は以下で構成されます:

• ある形式でのデータのシリアル化
• シリアル化されたデータをネットワーク接続経由で送信
• 受信側でのデータの逆シリアル化

たとえば、Webアプリケーションのフロントエンドとバックエンド間の通信について考えてみましょう。一般に、JavaScript Object Notation（JSON）形式は、データのシリアル化に使用されます。少量のデータの場合、これで十分です。シリアライズとデシリアライズのオーバーヘッドはごくわずかであり、JSONは人間が読める形式であるため、デバッグが簡単になります。ただし、データ量が増えると、シリアライゼーションのコストが主要なパフォーマンス要因になる可能性があります。適切な注意を怠ると、システムはデータのシリアル化にほとんどの時間を費やすことになります。明らかに、私たちのCPUサイクルを使うのにもっと有益なことがあります。

このプロセスでは、ソフトウェアで制御する (逆) シリアル化という要素が1つあります。言うまでもなく、シリアル化フレームワークはたくさんあります。ProtoBuf、Thrift、MessagePack、その他多くのことを考えてみてください。それらの多くは、シリアル化のコストを最小化することを主要な目標として持っています。

シリアル化を最小限に抑えるための努力にもかかわらず、必然的にまだ (逆) シリアル化のステップがあります。コードが作用するオブジェクトは、ネットワークを通して送信されるバイトではありません。ネットワーク経由で受信されるバイトは、反対側のクランチのコードのオブジェクトではありません。結局、最速のシリアル化はシリアル化しないことです。

Apache Arrowは私のためにありますか？

概念的には、BallistaやDremioやビッグデータシステムの統合など、Apache Arrowは、ビッグデータシステムのバックボーンとして設計されています。 ユースケースがビッグデータシステムの領域にない場合、Apache Arrowのオーバーヘッドは問題に値しないでしょう。 ProtoBuf、FlatBuffers、Thrift、MessagePackなど、業界で広く採用されているシリアル化フレームワークが適しています。

Apache Arrowを使用したコーディングは、Javaオブジェクトが存在しないという意味で、プレーンな古いJavaオブジェクトを使用したコーディングとは大きく異なります。コードはずっと下のバッファで動作します。Apache Commons、Guavaなどの既存のユーティリティライブラリは使用できなくなりました。バイトバッファを使用するには、一部のアルゴリズムを再実装する必要がある場合があります。最後に重要なことですが、常にオブジェクトではなく列の観点から考える必要があります。

Apache Arrowの上にシステムを構築するには、Arrowバッファーの読み取り、書き込み、吸い込み、吐き出し、働くことが必要です。データオブジェクトのコレクション (つまり、ある種のデータベース) で機能するシステムを構築していて、列に適したものを計算し、これをクラスタで実行することを計画している場合、Arrowは間違いなく投資に値するものです。

Parquet (後述) との統合により、永続化が比較的簡単になります。クロスプラットフォーム、クロス言語の側面は、ポリグロットマイクロサービスアーキテクチャをサポートし、既存のビッグデータランドスケープとの容易な統合を可能にします。Arrow Flightと呼ばれる組み込みのRPCフレームワークにより、標準化された効率的な方法でデータセットを簡単に共有/提供できます。

ゼロコピーデータ転送

そもそもなぜシリアル化が必要なのですか？Javaアプリケーションでは、通常、オブジェクトとプリミティブ値を操作します。これらのオブジェクトは、コンピュータのRAMメモリー内のバイトに何らかの形でマップされます。JDKは、オブジェクトがコンピュータ上のバイトにどのようにマップされるかを理解しています。ただし、このマッピングは別のマシンでは異なる場合があります。たとえば、バイトオーダ (別名エンディアン) が考えられます。さらに、すべてのプログラミング言語が同じプリミティブ型のセットを持っているわけではなく、同様の型を同じように格納しているわけでもありません。

シリアル化は、オブジェクトが使用するメモリを一般的な形式に変換します。フォーマットには仕様があり、プログラミング言語とプラットフォームごとに、オブジェクトをシリアル化された形式に変換したり、戻すためのライブラリが用意されています。言い換えると、シリアル化とは、各プログラミング言語とプラットフォームの特有の方法を妨げることなく、データを共有することです。シリアル化により、プラットフォームとプログラミング言語の違いがすべて解消され、すべてのプログラマーが好きな方法で作業できるようになります。翻訳者のように、異なる言語を話す人々の間の言語の壁を取り除きます。

シリアル化は、ほとんどの状況で非常に役立ちます。ただし、大量のデータを転送する場合は、大きなボトルネックになります。したがって、それらの場合にシリアル化プロセスを排除できますか? これが本当は、Apache ArrowやFlatBuffersなどのゼロコピーシリアル化フレームワークの目標です。シリアル化手順を回避するために、オブジェクトではなく、シリアル化されたデータ自体を処理するものと考えることができます。ここでゼロコピーとは、アプリケーションで作業するバイトを変更せずにネットワーク経由で転送できることを指します。同様に、受信側では、アプリケーションは、逆シリアル化手順を実行せずに、そのままバイトで作業を開始できます。

ここでの大きな利点は、データが接続の両側でそのまま理解されるため、変換せずに、ある環境から別の環境にデータをそのまま転送できることです。

主な欠点は、プログラミングにおける特異性の喪失です。すべての操作はバイトバッファで実行されます。整数はなく、一連のバイトがあります。配列はなく、一連のバイトがあります。オブジェクトはなく、一連のバイトのコレクションがあります。もちろん、一般的な形式のデータを整数、配列、オブジェクトに変換することもできます。しかし、その場合は逆シリアル化を行うことになり、それはゼロコピーの目的に反することになります。Javaオブジェクトに転送されると、データを操作できるのはJavaだけになります。

これは実際にはどのように機能しますか? 2つのゼロコピーシリアル化フレームワークである、Apache ArrowとGoogleのFlatBuffersを簡単に見てみましょう。どちらもゼロコピーフレームワークですが、それぞれ異なるフレーバーであり、さまざまなユースケースに対応します。

FlatBuffersは当初、モバイルゲームをサポートするために開発されました。焦点は、最小限のオーバーヘッドで、サーバからクライアントへのデータの高速伝送にあります。単一のオブジェクトまたはオブジェクトのコレクションを送信できます。データは (ヒープ上で) ByteBuffersに格納され、FlatBuffers共通データレイアウトでフォーマットされます。FlatBuffersコンパイラは、データ仕様に基づいて、ByteBufferとの対話を簡素化するコードを生成します。データは、配列、オブジェクト、またはプリミティブであるかのように操作できます。舞台裏で、各アクセサメソッドは対応するバイトをフェッチし、そのバイトをJVMおよびコードの理解可能な構造に変換します。何らかの理由でバイトへのアクセスが必要な場合でも、アクセスできます。

Arrowは、メモリ内のリスト/配列/テーブルのレイアウト方法がFlatBufferと異なります。FlatBuffersはそのテーブルに行指向のフォーマットを使用するのに対し、Arrowは表形式のデータを格納するために列形式を使用します。そして、それがビッグデータセットに対する分析 (OLAP) クエリのすべての違いになります。

Arrowは、通常、単一のオブジェクトではなく、オブジェクトの大きなコレクションを転送しないビッグデータシステムを対象としています。一方、FlatBuffersは、シリアル化フレームワークとして販売 (および使用) されています。つまり、アプリケーションコードはJavaオブジェクトとプリミティブで動作し、データを送信するときにのみデータをFlatBuffersのメモリレイアウトに変換します。受信側が読み取り専用の場合、データをJavaオブジェクトに逆シリアル化する必要はありません。データはFlatBuffersのByteBuffersから直接読み取ることができます。

大きなデータセットでは、行数は通常、数千から数兆の範囲になります。このようなデータセットには、数列から数千の列がある場合があります。

このようなデータセット参照の典型的な分析クエリは、ひとにぎりの列です。たとえば、eコマーストランザクションのデータセットを想像してみてください。セールスマネージャは、特定の地域の売上高の概要をアイテムカテゴリ別にグループ化したいと考えていると想像できます。彼は個々の販売を見たくありません。平均販売価格で十分です。このようなクエリは、次の3つのステップで回答できます:

• リクエストされたリージョンのsalesのすべての行/オブジェクトIDを追跡しながら、リージョン列のすべての値をトラバース
• アイテムカテゴリ列の対応する値に基づいて、フィルタリングされたIDをグループ化
• 各グループの集計の計算

基本的に、クエリプロセッサは、常に1つの列をメモリに格納するだけで済みます。コレクションを列形式で格納することにより、単一のフィールド/列のすべての値に個別にアクセスできます。適切に設計されたフォーマットでは、これはレイアウトがCPUのSIMD命令に対して最適化されるように行われます。このような分析ワークロードの場合、Apache Arrowの列レイアウトは、FlatBuffersの行指向のレイアウトよりも適しています。

Apache Arrow

Apache Arrowの中核は、インメモリデータレイアウト形式です。形式に加えて、Apache Arrowは、Apache Arrow形式のデータを操作するための一連のライブラリ (C、C ++、C#、Go、Java、JavaScript、MATLAB、Python、R、Ruby、Rustを含む) を提供します。この記事の残りの部分では、Arrowの基本的な概念を理解し、Apache Arrowを使用してJavaアプリケーションを作成する方法について説明します。

基本的な概念

Vector Schema Root

一連の店舗の販売実績をモデル化しているとしましょう。通常、売上を表すオブジェクトに遭遇します。このようなオブジェクトには、次のようなさまざまなプロパティがあります

• ID
• 地域、都市、おそらく店舗の種類など、販売が行われた店舗に関する情報
• いくつかの顧客情報
• 販売した商品のID
• 販売された商品のカテゴリ (およびサブカテゴリ)
• 販売された商品の数
• その他…

Javaでは、売上はSaleクラスによってモデル化されます。クラスには、1つの売上のすべての情報が含まれます。すべての売上は、Saleオブジェクトのコレクションによって (メモリ内に) 表現されます。データベースの観点からは、Saleオブジェクトのコレクションは、行指向のリレーショナルデータベースに相当します。実際、このようなアプリケーションでは、通常、オブジェクトのコレクションが永続化のためにデータベースのリレーショナルテーブルにマップされます。

列指向のデータベースでは、オブジェクトのコレクションは列のコレクションに分解されます。すべてのIDは単一の列に格納されます。メモリには、すべてのIDが順番に格納されます。同様に、売上ごとにすべての店舗の都市を格納する列があります。概念的には、この列形式は、オブジェクトのコレクションを同じ長さの配列のセットに分解するものと考えることができます。オブジェクトのフィールドごとに1つの配列。

特定のオブジェクトを再構築するために、分解された配列は、特定のインデックスで各列/配列の値を選択することによって結合されます。たとえば、10番目の売上は、ID配列の10番目の値、店舗所在都市配列の10番目の値などを使用して再構成されます。

Apache Arrowは、列指向のリレーショナルデータベースのように機能します。Javaオブジェクトのコレクションは、列のコレクションに分解されます。これは、Arrowではベクトルと呼ばれます。ベクトルは、Arrowの列形式の基本単位です。

すべてのベクトルの元はFieldVectorです。Int4VectorやFloat8Vectorなど、プリミティブ型にはベクトル型があります。文字列には、VarCharVectorというベクトル型があります。任意のバイナリデータのベクトル型があります: VarBinaryVector。TimeStampVector、TimeStampSecVector、TimeStampTZVector、TimeMicroVectorなど、いくつかの型のベクトルが時間をモデル化するために存在します。

より複雑な構造を構成できます。StructVectorは、一連のベクトルを1つのフィールドにグループ化するために使用されます。たとえば、上記の販売を例に店舗情報について考えてみましょう。すべての店舗情報 (地域、都市、タイプ) を1つのStructVectorにグループ化できます。ListVectorを使用すると、要素の可変長リストを1つのフィールドに格納できます。MapVectorは、キーと値のマッピングを1つのベクトルに格納します。

データベースの類推を続けると、オブジェクトのコレクションはテーブルで表されます。テーブルの値を識別するために、テーブルにはスキーマ (名前から型へのマッピング) があります。行指向データベースでは、各行が名前を事前定義されたタイプの値にマップします。Javaでは、スキーマはクラス定義のメンバー変数のセットに対応します。列指向データベースにも同様にスキーマがあります。テーブルでは、スキーマの各名前が事前定義されたタイプの列にマップされます。

Apache Arrowの用語では、ベクトルのコレクションはVectorSchemaRootで表されます。VectorSchemaRootにはスキーマも含まれ、名前 (別名 フィールド) を列 (別名 ベクトル) にマッピングします。

バッファアロケータ

ベクトルに追加する値はどこに保存されますか? Arrowのベクトルはバッファに支えられています。通常、これは java.nio.ByteBuffer です。バッファは、バッファアロケータにプールされます。バッファアロケータに特定のサイズのバッファを作成するよう依頼するか、バッファアロケータにバッファの作成と自動拡張を任せて、新しい値を格納できます。バッファアロケータは、割り当てられたすべてのバッファを追跡します。

ベクトルは1つのアロケーターによって管理されます。 アロケータはベクトルを支えるバッファを所有していると言えます。ベクトルの所有権は、あるアロケータから別のアロケータに転送できます。

たとえば、データフローを実装しているとします。フローは一連の処理のステージで構成されます。各ステージは、データを次のステージに渡す前に、データに対していくつかの操作を実行します。各ステージには独自のバッファアロケータがあり、現在処理されているバッファを管理します。処理が完了すると、データは次のステージに渡されます。

言い換えると、ベクトルをサポートするバッファの所有権は、次のステージのバッファアロケータに転送されます。現在、そのバッファアロケータはメモリを管理し、不要になったときにメモリを解放する役割を担っています。

アロケータによって作成されるバッファはDirectByteBufferであるため、ヒープ以外の場所に格納されます。これは、データの使用が終了したら、メモリを解放する必要があることを意味します。これは、最初はJavaプログラマにとって奇妙に感じられます。しかし、それはApache Arrowでの作業の重要な部分です。ベクトルはAutoCloseableインターフェースをインプリメントするため、ベクトルの作成をtry-with-resourcesブロックでラップして、ベクトルを自動的に閉じること、つまりメモリを解放することをお勧めします。

例: 書き込み、読み込み、処理

この紹介を締めくくるために、Apache Arrowを使用したサンプルアプリケーションについて説明します。アイデアは、ディスク上のファイルから人の「データベース」を読み取り、データをフィルタリングして集計し、結果を出力することです。

Apache Arrowはメモリ内形式であることに注意してください。実際のアプリケーションでは、Parquetなどの永続的なストレージ用に最適化された他の (列) 形式が適しています。Parquetは、ディスクに書き込まれたデータに圧縮と中間サマリーを追加します。その結果、ディスクからのParquetファイルの読み取りと書き込みは、Apache Arrowファイルの読み取りと書き込みよりも高速になります。この例では、Arrowは純粋に教育目的で使用されています。

PersonクラスとAddressクラス (関連する部分のみを表示) があるとします:

public Person(String firstName, String lastName, int age, Address address) {
    this.firstName = firstName;
    this.lastName = lastName;
    this.age = age;

    this.address = address;
}

public Address(String street, int streetNumber, String city, int postalCode) {
    this.street = street;
    this.streetNumber = streetNumber;
    this.city = city;
    this.postalCode = postalCode;
}

2つのアプリケーションを作成します。最初のアプリケーションは、ランダムに生成された人々のコレクションを生成し、それらをArrow形式でディスクに書き込みます。次に、Arrow形式の「人のデータベース」をディスクからメモリに読み込むアプリケーションを作成します。すべての人から選択

• 「P」で始まる姓を持つ
• 年齢が18から35
• 「way」で終わる通りに住む

選択された人々について、都市ごとにグループ化された平均年齢を計算します。この例では、Apache Arrowを使用してインメモリデータ分析を実装する方法について、いくつかの視点を与える必要があります。

この例のコードは、このGitリポジトリにあります。

データの書き込み

データの書き込みを始める前に。Arrow形式はメモリ内データを対象としていることに注意してください。データのディスクストレージは最適化されていません。実際のアプリケーションでは、Parquetなどの形式を調べて、データを永続化するために、圧縮やその他のトリックをサポートして、列データのディスク上のストレージを高速化する必要があります。ここでは、議論の焦点を絞り簡潔にするために、データをArrow形式で書き出します。

Personオブジェクトの配列を指定して、people.arrowというファイルにデータを書き始めましょう。最初のステップは、Personオブジェクトの配列をArrow VectorSchemaRootに変換することです。Arrowを最大限に活用したい場合は、アプリケーション全体を記述してArrowベクトルを使用します。ただし、教育目的の場合は、ここで変換を行うと便利です。

private void vectorizePerson(int index, Person person, VectorSchemaRoot schemaRoot) {
    // Using setSafe: it increases the buffer capacity if needed
    ((VarCharVector) schemaRoot.getVector("firstName")).setSafe(index, person.getFirstName().getBytes());
    ((VarCharVector) schemaRoot.getVector("lastName")).setSafe(index, person.getLastName().getBytes());
    ((UInt4Vector) schemaRoot.getVector("age")).setSafe(index, person.getAge());

    List<FieldVector> childrenFromFields = schemaRoot.getVector("address").getChildrenFromFields();

    Address address = person.getAddress();
    ((VarCharVector) childrenFromFields.get(0)).setSafe(index, address.getStreet().getBytes());
    ((UInt4Vector) childrenFromFields.get(1)).setSafe(index, address.getStreetNumber());
    ((VarCharVector) childrenFromFields.get(2)).setSafe(index, address.getCity().getBytes());
    ((UInt4Vector) childrenFromFields.get(3)).setSafe(index, address.getPostalCode());
}

vectorizePersonでは、Personオブジェクトは、personスキーマを使用してschemaRootのベクトルにマップされます。setSafeメソッドは、バッキングバッファが次の値を保持するのに十分な大きさであることを保証します。バッキングバッファが十分に大きくない場合、バッファは拡張されます。

VectorSchemaRootは、スキーマとベクトルのコレクションのコンテナです。そのため、VectorSchemaRootクラスはスキーマレスデータベースと考えることができるため、スキーマは、オブジェクトのインスタンス化時にコンストラクタでスキーマが渡されたときにのみ認識されます。したがって、すべてのメソッド (getVectorなど) には、非常に一般的な戻り値の型 (この場合はFieldVector) があります。その結果、スキーマまたはデータセットの知識に基づいた多くのキャストが必要になります。

この例では、UInt4VectorsとUInt2Vectorを事前に割り当てることを選択できます (事前にバッチに何人の人がいるかがわかっているため) 。次に、setメソッドを使用して、バッファーサイズのチェックとバッファーを拡張するための再割り当てを回避できます。

vectorizePerson関数はChunkedWriterに渡すことができます。ChunkedWriterは、チャンクを処理し、Arrow形式のバイナリファイルに書き込むアブストラクトです。

void writeToArrowFile(Person[] people) throws IOException {
   new ChunkedWriter<>(CHUNK_SIZE, this::vectorizePerson).write(new File("people.arrow"), people);
}

The ChunkedWriter has a write method that looks like this:
public void write(File file, Person[] values) throws IOException {
   DictionaryProvider.MapDictionaryProvider dictProvider = new DictionaryProvider.MapDictionaryProvider();

   try (RootAllocator allocator = new RootAllocator();
        VectorSchemaRoot schemaRoot = VectorSchemaRoot.create(personSchema(), allocator);
        FileOutputStream fd = new FileOutputStream(file);
        ArrowFileWriter fileWriter = new ArrowFileWriter(schemaRoot, dictProvider, fd.getChannel())) {
       fileWriter.start();

       int index = 0;
       while (index < values.length) {
           schemaRoot.allocateNew();
           int chunkIndex = 0;
           while (chunkIndex < chunkSize && index + chunkIndex < values.length) {
               vectorizer.vectorize(values[index + chunkIndex], chunkIndex, schemaRoot);
               chunkIndex++;
           }
           schemaRoot.setRowCount(chunkIndex);
           fileWriter.writeBatch();

           index += chunkIndex;
           schemaRoot.clear();
       }
       fileWriter.end();
   }
}

これを分解してみましょう。まず、 (i) アロケータ、 (ii) schemaRoot、 (iii) dictProviderを作成します。 (i) メモリバッファを割り当てる、 (ii) ベクトルのコンテナ (バッファでサポートされる) にする、および (iii) 辞書圧縮を容易にする (これは今のところ無視できる) 必要があります。

次に、 (2) でArrowFileWriterが作成されます。VectorSchemaRootに基づいて、ディスクへの書き込みを処理します。このように、データセットをバッチで書き込むのは非常に簡単です。最後に重要なことですが、ライターを開始することを忘れないでください。

メソッドの残りの部分は、Person配列をチャンクでベクトルスキーマルートにベクトル化し、バッチごとに書き出すことです。

バッチで書くことの利点は何ですか? ある時点で、データはディスクから読み取られます。データが1つのバッチで書き込まれる場合、すべてのデータを一度に読み取ってメインメモリに格納する必要があります。バッチを記述することにより、リーダがデータを小さなチャンクで処理できるようになり、それによってメモリのフットプリントが制限されます。

ベクトルの値カウントまたはベクトルスキーマルートの行カウントを設定することを忘れないでください (含まれるすべてのベクトルの値カウントを間接的に設定します) 。カウントを設定しないと、ベクトルに値を格納した後でも、ベクトルは空で表示されます。

最後に、すべてのデータがベクターに格納されると、fileWriter.writeBatch() はそれらをディスクにコミットします。

メモリ管理に注意

3行目と4行目の schemaRoot.clear() と allocator.close() に注意してください。前者は、VectorSchemaRootに含まれるすべてのベクトルのすべてのデータをクリアし、行と値のカウントをゼロにリセットします。後者はアロケータを閉じます。割り当てられたバッファを解放するのを忘れた場合、この呼び出しはメモリリークがあることを通知します。

この設定では、アロケータのクローズ直後にプログラムが終了するため、クローズのほとんど不要です。ただし、実際の長期実行アプリケーションでは、メモリ管理が重要です。

メモリ管理の懸念は、Javaプログラマにとって異質なものと感じられます。ただし、この場合は、パフォーマンスの代償を払う必要があります。割り当てられたバッファについて十分に注意し、存続期間の終わりにそれらを解放します。

データの読み込み

Arrow形式のファイルからのデータの読み取りは、書き込みに似ています。アロケータ、ベクトルスキーマルート (スキーマなし、それはファイルの一部です) を設定し、ファイルを開いて、ArrowFileReaderに残りの処理を任せます。ファイルからスキーマを読み取るための初期化を忘れないでください。

バッチを読み取るには、fileReader.loadNextBatch() を呼び出します。次のバッチがまだ使用可能な場合は、ディスクから読み取られ、schemaRootのベクトルのバッファーにデータが入力され、処理の準備が整います。

次のコードスニペットは、Arrowファイルの読み取り方法を簡単に説明しています。whileループが実行されるたびに、バッチがVectorSchemaRootにロードされます。バッチの内容は、VectorSchemaRootによって記述されます。 (i) VectorSchemaRootのスキーマ、および (ii) 値の数は、エントリの数と同じです。

try (FileInputStream fd = new FileInputStream("people.arrow");
    ArrowFileReader fileReader = new ArrowFileReader(new SeekableReadChannel(fd.getChannel()), allocator)) {
   // Setup file reader
   fileReader.initialize();
   VectorSchemaRoot schemaRoot = fileReader.getVectorSchemaRoot();

   // Aggregate: Using ByteString as it is faster than creating a String from a byte[]
   while (fileReader.loadNextBatch()) {
      // Processing ...
   }
}

データ処理

最後に重要なことですが、フィルタリング、グループ化、および集計の手順により、データ分析ソフトウェアでArrowベクトルを操作する方法を理解できるはずです。これがArrowのベクトルを操作する方法であると偽って考えたくはありませんが、Apache Arrowを探索するための確固たる出発点になるはずです。実際のArrowコードのGandiva処理エンジンのソースコードをご覧ください。 Apache Arrowによるデータ処理は大きなトピックです。あなたは文字通りそれについての本を書くことができます。

サンプルコードはPersonのユースケースに固有のものであることに注意してください。たとえば、Arrowベクトルを使用してクエリプロセッサを構築する場合、ベクトルの名前と型は事前にわからないため、より一般的で理解しにくいコードになります。

Arrowは列形式であるため、1つの列だけを使用して、フィルタリング手順を個別に適用できます。

private IntArrayList filterOnAge(VectorSchemaRoot schemaRoot) {
    UInt4Vector age = (UInt4Vector) schemaRoot.getVector("age");
    IntArrayList ageSelectedIndexes = new IntArrayList();
    for (int i = 0; i < schemaRoot.getRowCount(); i++) {
        int currentAge = age.get(i);
        if (18 <= currentAge && currentAge <= 35) {
            ageSelectedIndexes.add(i);
        }
    }
    ageSelectedIndexes.trim();
    return ageSelectedIndexes;
}

このメソッドは、値が18から35の年齢ベクトルのロードされたチャンク内のすべてのインデックスを収集します。

各フィルタは、そのようなインデックスのソートされたリストを生成します。次のステップでは、これらのリストをインターセクト/マージして、選択したインデックスの単一のリストにします。このリストには、すべての基準を満たす行のすべてのインデックスが含まれています。

次のコードスニペットは、ベクトルと選択したIDのコレクションから、集計データ構造 (都市をカウントと合計にマッピングする) を簡単に入力する方法を示しています。

VarCharVector cityVector = (VarCharVector) ((StructVector) schemaRoot.getVector("address")).getChild("city");
UInt4Vector ageDataVector = (UInt4Vector) schemaRoot.getVector("age");

for (int selectedIndex : selectedIndexes) {
   String city = new String(cityVector.get(selectedIndex));
   perCityCount.put(city, perCityCount.getOrDefault(city, 0L) + 1);
   perCitySum.put(city, perCitySum.getOrDefault(city, 0L) + ageDataVector.get(selectedIndex));
}

集計データ構造が入力された後、都市ごとの平均年齢を出力するのは非常に簡単です:

for (String city : perCityCount.keySet()) {
    double average = (double) perCitySum.get(city) / perCityCount.get(city);
    LOGGER.info("City = {}; Average = {}", city, average);
}

結論

この記事では、列型、インメモリ、言語間データレイアウトフォーマットのApache Arrowを紹介しました。これは、ビッグデータシステムのビルディングブロックであり、クラスタ内のマシン間および異なるビッグデータシステム間の効率的なデータ転送に焦点を当てています。Apache Arrowを使用してJavaアプリケーションの開発を始めるために、Arrow形式でデータを読み書きする2つのサンプルアプリケーションを調べました。また、Apache Arrow Javaライブラリを使用してデータを処理する最初の経験を得ました。

Apache Arrowは列形式です。列指向のレイアウトは、通常、行指向のレイアウトよりも分析ワークロードに適しています。ただし、常にトレードオフがあります。特定のワークロードについては、行指向の形式がより良い結果をもたらす可能性があります。

VectorSchemaRoots、バッファ、およびメモリ管理は、慣用的なJavaコードのようには見えません。FlatBuffersなどの別のフレームワークから必要なすべてのパフォーマンスを得ることができる場合、あまり慣用的でない作業方法が、アプリケーションにApache Arrowを採用する決定に影響を与える可能性があります。

著者について

Joris Gillis氏はTrendMinerの研究開発者です。TrendMinerは、IIoT時系列データ用のセルフサービス分析ソフトウェアを作成しています。研究開発者として、スケーラブルな分析アルゴリズム、時系列データベース、および外部時系列データソースへの接続に取り組んでいます。