| .. Licensed to the Apache Software Foundation (ASF) under one | |
| .. or more contributor license agreements. See the NOTICE file | |
| .. distributed with this work for additional information | |
| .. regarding copyright ownership. The ASF licenses this file | |
| .. to you under the Apache License, Version 2.0 (the | |
| .. "License"); you may not use this file except in compliance | |
| .. with the License. You may obtain a copy of the License at | |
| .. http://www.apache.org/licenses/LICENSE-2.0 | |
| .. Unless required by applicable law or agreed to in writing, | |
| .. software distributed under the License is distributed on an | |
| .. "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | |
| .. KIND, either express or implied. See the License for the | |
| .. specific language governing permissions and limitations | |
| .. under the License. | |
| =========================== | |
| Reading/Writing IPC formats | |
| =========================== | |
| Arrow defines two types of binary formats for serializing record batches: | |
| * **Streaming format**: for sending an arbitrary number of record | |
| batches. The format must be processed from start to end, and does not support | |
| random access | |
| * **File or Random Access format**: for serializing a fixed number of record | |
| batches. It supports random access, and thus is very useful when used with | |
| memory maps | |
| Writing and Reading Streaming Format | |
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| First, let's populate a :class:`VectorSchemaRoot` with a small batch of records | |
| .. code-block:: Java | |
| BitVector bitVector = new BitVector("boolean", allocator); | |
| VarCharVector varCharVector = new VarCharVector("varchar", allocator); | |
| for (int i = 0; i < 10; i++) { | |
| bitVector.setSafe(i, i % 2 == 0 ? 0 : 1); | |
| varCharVector.setSafe(i, ("test" + i).getBytes(StandardCharsets.UTF_8)); | |
| } | |
| bitVector.setValueCount(10); | |
| varCharVector.setValueCount(10); | |
| List<Field> fields = Arrays.asList(bitVector.getField(), varCharVector.getField()); | |
| List<FieldVector> vectors = Arrays.asList(bitVector, varCharVector); | |
| VectorSchemaRoot root = new VectorSchemaRoot(fields, vectors); | |
| Now, we can begin writing a stream containing some number of these batches. For this we use :class:`ArrowStreamWriter` | |
| (DictionaryProvider used for any vectors that are dictionary encoded is optional and can be null)) | |
| .. code-block:: Java | |
| try ( | |
| ByteArrayOutputStream out = new ByteArrayOutputStream(); | |
| ArrowStreamWriter writer = new ArrowStreamWriter(root, /*DictionaryProvider=*/null, Channels.newChannel(out)); | |
| ) { | |
| // ... do write into the ArrowStreamWriter | |
| } | |
| Here we used an in-memory stream, but this could have been a socket or some other IO stream. Then we can do | |
| .. code-block:: Java | |
| writer.start(); | |
| // write the first batch | |
| writer.writeBatch(); | |
| // write another four batches. | |
| for (int i = 0; i < 4; i++) { | |
| // populate VectorSchemaRoot data and write the second batch | |
| BitVector childVector1 = (BitVector)root.getVector(0); | |
| VarCharVector childVector2 = (VarCharVector)root.getVector(1); | |
| childVector1.reset(); | |
| childVector2.reset(); | |
| // ... do some populate work here, could be different for each batch | |
| writer.writeBatch(); | |
| } | |
| writer.end(); | |
| Note that, since the :class:`VectorSchemaRoot` in the writer is a container that can hold batches, batches flow through | |
| :class:`VectorSchemaRoot` as part of a pipeline, so we need to populate data before ``writeBatch``, so that later batches | |
| could overwrite previous ones. | |
| Now the :class:`ByteArrayOutputStream` contains the complete stream which contains 5 record batches. | |
| We can read such a stream with :class:`ArrowStreamReader`. Note that the :class:`VectorSchemaRoot` within the reader | |
| will be loaded with new values on every call to :class:`loadNextBatch()` | |
| .. code-block:: Java | |
| try (ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(out.toByteArray()), allocator)) { | |
| // This will be loaded with new values on every call to loadNextBatch | |
| VectorSchemaRoot readRoot = reader.getVectorSchemaRoot(); | |
| Schema schema = readRoot.getSchema(); | |
| for (int i = 0; i < 5; i++) { | |
| reader.loadNextBatch(); | |
| // ... do something with readRoot | |
| } | |
| } | |
| Here we also give a simple example with dictionary encoded vectors | |
| .. code-block:: Java | |
| // create provider | |
| DictionaryProvider.MapDictionaryProvider provider = new DictionaryProvider.MapDictionaryProvider(); | |
| try ( | |
| final VarCharVector dictVector = new VarCharVector("dict", allocator); | |
| final VarCharVector vector = new VarCharVector("vector", allocator); | |
| ) { | |
| // create dictionary vector | |
| dictVector.allocateNewSafe(); | |
| dictVector.setSafe(0, "aa".getBytes()); | |
| dictVector.setSafe(1, "bb".getBytes()); | |
| dictVector.setSafe(2, "cc".getBytes()); | |
| dictVector.setValueCount(3); | |
| // create dictionary | |
| Dictionary dictionary = | |
| new Dictionary(dictVector, new DictionaryEncoding(1L, false, /*indexType=*/null)); | |
| provider.put(dictionary); | |
| // create original data vector | |
| vector.allocateNewSafe(); | |
| vector.setSafe(0, "bb".getBytes()); | |
| vector.setSafe(1, "bb".getBytes()); | |
| vector.setSafe(2, "cc".getBytes()); | |
| vector.setSafe(3, "aa".getBytes()); | |
| vector.setValueCount(4); | |
| // get the encoded vector | |
| IntVector encodedVector = (IntVector) DictionaryEncoder.encode(vector, dictionary); | |
| ByteArrayOutputStream out = new ByteArrayOutputStream(); | |
| // create VectorSchemaRoot | |
| List<Field> fields = Arrays.asList(encodedVector.getField()); | |
| List<FieldVector> vectors = Arrays.asList(encodedVector); | |
| try (VectorSchemaRoot root = new VectorSchemaRoot(fields, vectors)) { | |
| // write data | |
| ArrowStreamWriter writer = new ArrowStreamWriter(root, provider, Channels.newChannel(out)); | |
| writer.start(); | |
| writer.writeBatch(); | |
| writer.end(); | |
| } | |
| // read data | |
| try (ArrowStreamReader reader = new ArrowStreamReader(new ByteArrayInputStream(out.toByteArray()), allocator)) { | |
| reader.loadNextBatch(); | |
| VectorSchemaRoot readRoot = reader.getVectorSchemaRoot(); | |
| // get the encoded vector | |
| IntVector intVector = (IntVector) readRoot.getVector(0); | |
| // get dictionaries and decode the vector | |
| Map<Long, Dictionary> dictionaryMap = reader.getDictionaryVectors(); | |
| long dictionaryId = intVector.getField().getDictionary().getId(); | |
| try (VarCharVector varCharVector = | |
| (VarCharVector) DictionaryEncoder.decode(intVector, dictionaryMap.get(dictionaryId))) { | |
| // ... use decoded vector | |
| } | |
| } | |
| } | |
| Writing and Reading Random Access Files | |
| ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~ | |
| The :class:`ArrowFileWriter` has the same API as :class:`ArrowStreamWriter` | |
| .. code-block:: Java | |
| try ( | |
| ByteArrayOutputStream out = new ByteArrayOutputStream(); | |
| ArrowFileWriter writer = new ArrowFileWriter(root, /*DictionaryProvider=*/null, Channels.newChannel(out)); | |
| ) { | |
| writer.start(); | |
| // write the first batch | |
| writer.writeBatch(); | |
| // write another four batches. | |
| for (int i = 0; i < 4; i++) { | |
| // ... do populate work | |
| writer.writeBatch(); | |
| } | |
| writer.end(); | |
| } | |
| The difference between :class:`ArrowFileReader` and :class:`ArrowStreamReader` is that the input source | |
| must have a ``seek`` method for random access. Because we have access to the entire payload, we know the | |
| number of record batches in the file, and can read any at random | |
| .. code-block:: Java | |
| try (ArrowFileReader reader = new ArrowFileReader( | |
| new ByteArrayReadableSeekableByteChannel(out.toByteArray()), allocator)) { | |
| // read the 4-th batch | |
| ArrowBlock block = reader.getRecordBlocks().get(3); | |
| reader.loadRecordBatch(block); | |
| VectorSchemaRoot readBatch = reader.getVectorSchemaRoot(); | |
| } |