JVM 上の高性能データ形式ライブラリ パッケージである Apache Arrow の紹介とアーキテクチャ (Gkatziouras)

<依存関係>
    <グループ ID>org.apache.arrow</グループ ID>
    <artifactId>矢印メモリネットティ</artifactId>
    <バージョン>${arrow.version}</バージョン>
</依存関係>
<依存関係>
    <グループ ID>org.apache.arrow</グループ ID>
    <artifactId>矢印ベクトル</artifactId>
    <バージョン>${arrow.version}</バージョン>
</依存関係>

パッケージ com.gkatzioura.ar​​row;

java.io.IOException をインポートします。

java.util.List をインポートします。

org.apache.arrow.vector.types.pojo.ArrowType をインポートします。

org.apache.arrow.vector.types.pojo.Field をインポートします。

org.apache.arrow.vector.types.pojo.FieldType をインポートします。

org.apache.arrow.vector.types.pojo.Schema をインポートします。

パブリッククラスSchemaFactory {

パブリック静的スキーマ DEFAULT_SCHEMA = createDefault();

パブリック静的スキーマcreateDefault() {

var strField = new Field("col1", FieldType.nullable(new ArrowType.Utf8()), null);

var intField = new Field("col2", FieldType.nullable(new ArrowType.Int(32, true)), null);

新しいスキーマ(List.of(strField, intField))を返します。

}

パブリック静的スキーマ schemaWithChildren() {

var amount = new Field("amount", FieldType.nullable(new ArrowType.Decimal(19,4,128)), null);

var currency = new Field("currency",FieldType.nullable(new ArrowType.Utf8()), null);

var itemField = new Field("item", FieldType.nullable(new ArrowType.Utf8()), List.of(amount,currency));

新しいスキーマ(List.of(itemField))を返します。

}

パブリック静的SchemafromJson(String jsonString) {

試す {

Schema.fromJSON(jsonString) を返します。

} キャッチ (IOException e) {

新しい ArrowExampleException(e) をスローします。

}

}

}

{
  「フィールド」: [ {
    "名前" : "col1",
    「null可能」: true、
    "タイプ" ： {
      "名前" : "utf8"
    },
    "子供たち" ： [ ]
  }, {
    "名前" : "col2",
    「null可能」: true、
    "タイプ" ： {
      "名前" : "int",
      「ビット幅」: 32,
      "isSigned" : 真
    },
    "子供たち" ： [ ]
  } ]
}

パブリック静的スキーマ schemaWithChildren() {
    var amount = new Field("amount", FieldType.nullable(new ArrowType.Decimal(19,4,128)), null);
    var currency = new Field("currency",FieldType.nullable(new ArrowType.Utf8()), null);
    var itemField = new Field("item", FieldType.nullable(new ArrowType.Utf8()), List.of(amount,currency));
 
    新しいスキーマ(List.of(itemField))を返します。
}

パッケージ com.gkatzioura.ar​​row;
 
lombok.Builder をインポートします。
lombok.Data をインポートします。
 
@データ
@ビルダー
パブリッククラスDefaultArrowEntry {
 
    プライベート文字列 col1;
    プライベート整数col2;
 
}

パッケージ com.gkatzioura.ar​​row;
 
java.io.Closeable をインポートします。
java.io.IOException をインポートします。
java.nio.channels.WritableByteChannel をインポートします。
java.util.List をインポートします。
 
org.apache.arrow.memory.RootAllocator をインポートします。
org.apache.arrow.vector.IntVector をインポートします。
org.apache.arrow.vector.VarCharVector をインポートします。
org.apache.arrow.vector.VectorSchemaRoot をインポートします。
org.apache.arrow.vector.dictionary.DictionaryProvider をインポートします。
org.apache.arrow.vector.ipc.ArrowStreamWriter をインポートします。
org.apache.arrow.vector.util.Text をインポートします。
 
static com.gkatzioura.ar​​row.SchemaFactory.DEFAULT_SCHEMA をインポートします。
 
パブリッククラスDefaultEntriesWriterはCloseableを実装します{
 
    プライベート最終 RootAllocator rootAllocator;
    private final VectorSchemaRoot vectorSchemaRoot; //ベクトルアロケータの作成:
 
    パブリックDefaultEntriesWriter() {
        rootAllocator = 新しい RootAllocator();
        ルートアロケータを作成します。
    }
 
    パブリック void write(List<DefaultArrowEntry> defaultArrowEntries、int batchSize、WritableByteChannel out) {
        バッチサイズ <= 0 の場合
            バッチサイズ = defaultArrowEntries.size();
        }
 
        DictionaryProvider.MapDictionaryProvider dictProvider = 新しい DictionaryProvider.MapDictionaryProvider();
        試してください(ArrowStreamWriterライター = 新しいArrowStreamWriter(vectorSchemaRoot, dictProvider, out)) {
            ライターを起動します。
 
            VarCharVector の childVector1 = (VarCharVector) vectorSchemaRoot.getVector(0);
            IntVector の childVector2 = (IntVector) vectorSchemaRoot.getVector(1);
            子ベクター1.リセット();
            子ベクター2.リセット();
 
            ブール値 exactBatches = defaultArrowEntries.size()%batchSize == 0;
            int バッチカウンタ = 0;
 
            for(int i=0; i < defaultArrowEntries.size(); i++) {
                childVector1.setSafe(batchCounter、新しい Text(defaultArrowEntries.get(i).getCol1()));
                childVector2.setSafe(batchCounter、defaultArrowEntries.get(i).getCol2());
 
                バッチカウンタ++;
 
                バッチカウンタ == バッチサイズの場合
                    バッチサイズをセットします。
                    ライター.writeBatch();
                    バッチカウンタ = 0;
                }
            }
 
            if(!exactBatches) {
                vectorSchemaRoot.setRowCount(バッチカウンタ);
                ライター.writeBatch();
            }
 
            ライター.end();
        } キャッチ (IOException e) {
            新しい ArrowExampleException(e) をスローします。
        }
    }
 
    @オーバーライド
    パブリック void close() は IOException をスローします {
        ベクトルスキーマルートを閉じます。
        ルートアロケータを閉じます。
    }
 
}

パブリックDefaultEntriesToBytesConverter() {
    rootAllocator = 新しい RootAllocator();
    ルートアロケータを作成します。
}

ArrowStreamWriter ライター = 新しい ArrowStreamWriter(vectorSchemaRoot、dictProvider、Channels.newChannel(out));
ライターを起動します。

VarCharVector の childVector1 = (VarCharVector) vectorSchemaRoot.getVector(0);
IntVector の childVector2 = (IntVector) vectorSchemaRoot.getVector(1);
子ベクター1.リセット();
子ベクター2.リセット();

childVector1.setSafe(i, 新しい Text(defaultArrowEntries.get(i).getCol1()));
childVector2.setSafe(i, defaultArrowEntries.get(i).getCol2());

バッチサイズをセットします。
ライター.writeBatch();

@オーバーライド
パブリック void close() は IOException をスローします {
    ベクトルスキーマルートを閉じます。
    ルートアロケータを閉じます。
}