| <?php |
| |
| /** |
| * 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 |
| * |
| * https://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. |
| */ |
| |
| namespace Apache\Avro\DataFile; |
| |
| use Apache\Avro\AvroIO; |
| use Apache\Avro\Datum\AvroIODatumReader; |
| use Apache\Avro\Datum\AvroIODatumWriter; |
| use Apache\Avro\IO\AvroFile; |
| use Apache\Avro\Schema\AvroSchema; |
| |
| /** |
| * @package Avro |
| */ |
| class AvroDataIO |
| { |
| /** |
| * @var int used in file header |
| */ |
| public const VERSION = 1; |
| |
| /** |
| * @var int count of bytes in synchronization marker |
| */ |
| public const SYNC_SIZE = 16; |
| |
| /** |
| * @var int count of items per block, arbitrarily set to 4000 * SYNC_SIZE |
| * @todo make this value configurable |
| */ |
| public const SYNC_INTERVAL = 64000; |
| |
| /** |
| * @var string map key for datafile metadata codec value |
| */ |
| public const METADATA_CODEC_ATTR = 'avro.codec'; |
| |
| /** |
| * @var string map key for datafile metadata schema value |
| */ |
| public const METADATA_SCHEMA_ATTR = 'avro.schema'; |
| /** |
| * @var string JSON for datafile metadata schema |
| */ |
| public const METADATA_SCHEMA_JSON = '{"type":"map","values":"bytes"}'; |
| |
| /** |
| * @var string codec value for NULL codec |
| */ |
| public const NULL_CODEC = 'null'; |
| |
| /** |
| * @var string codec value for deflate codec |
| */ |
| public const DEFLATE_CODEC = 'deflate'; |
| |
| public const SNAPPY_CODEC = 'snappy'; |
| |
| public const ZSTANDARD_CODEC = 'zstandard'; |
| |
| public const BZIP2_CODEC = 'bzip2'; |
| |
| /** |
| * @var array array of valid codec names |
| */ |
| private static $validCodecs = [ |
| self::NULL_CODEC, |
| self::DEFLATE_CODEC, |
| self::SNAPPY_CODEC, |
| self::ZSTANDARD_CODEC, |
| self::BZIP2_CODEC |
| ]; |
| |
| /** |
| * @var AvroSchema cached version of metadata schema object |
| */ |
| private static $metadataSchema; |
| |
| /** |
| * @returns int count of bytes in the initial "magic" segment of the |
| * Avro container file header |
| */ |
| public static function magicSize() |
| { |
| return strlen(self::magic()); |
| } |
| |
| /** |
| * @returns the initial "magic" segment of an Avro container file header. |
| */ |
| public static function magic() |
| { |
| return ('Obj' . pack('c', self::VERSION)); |
| } |
| |
| /** |
| * @returns AvroSchema object of Avro container file metadata. |
| */ |
| public static function metadataSchema() |
| { |
| if (is_null(self::$metadataSchema)) { |
| self::$metadataSchema = AvroSchema::parse(self::METADATA_SCHEMA_JSON); |
| } |
| return self::$metadataSchema; |
| } |
| |
| /** |
| * @param string $file_path file_path of file to open |
| * @param string $mode one of AvroFile::READ_MODE or AvroFile::WRITE_MODE |
| * @param string $schemaJson JSON of writer's schema |
| * @param string $codec compression codec |
| * @returns AvroDataIOWriter instance of AvroDataIOWriter |
| * |
| * @throws AvroDataIOException if $writers_schema is not provided |
| * or if an invalid $mode is given. |
| */ |
| public static function openFile( |
| $file_path, |
| $mode = AvroFile::READ_MODE, |
| $schemaJson = null, |
| $codec = self::NULL_CODEC |
| ) { |
| $schema = !is_null($schemaJson) |
| ? AvroSchema::parse($schemaJson) : null; |
| |
| $io = false; |
| switch ($mode) { |
| case AvroFile::WRITE_MODE: |
| if (is_null($schema)) { |
| throw new AvroDataIOException('Writing an Avro file requires a schema.'); |
| } |
| $file = new AvroFile($file_path, AvroFile::WRITE_MODE); |
| $io = self::openWriter($file, $schema, $codec); |
| break; |
| case AvroFile::READ_MODE: |
| $file = new AvroFile($file_path, AvroFile::READ_MODE); |
| $io = self::openReader($file, $schema); |
| break; |
| default: |
| throw new AvroDataIOException( |
| sprintf( |
| "Only modes '%s' and '%s' allowed. You gave '%s'.", |
| AvroFile::READ_MODE, |
| AvroFile::WRITE_MODE, |
| $mode |
| ) |
| ); |
| } |
| return $io; |
| } |
| |
| /** |
| * @param AvroIO $io |
| * @param AvroSchema $schema |
| * @param string $codec |
| * @returns AvroDataIOWriter |
| */ |
| protected static function openWriter($io, $schema, $codec = self::NULL_CODEC) |
| { |
| $writer = new AvroIODatumWriter($schema); |
| return new AvroDataIOWriter($io, $writer, $schema, $codec); |
| } |
| |
| /** |
| * @param AvroIO $io |
| * @param AvroSchema $schema |
| * @returns AvroDataIOReader |
| */ |
| protected static function openReader($io, $schema) |
| { |
| $reader = new AvroIODatumReader(null, $schema); |
| return new AvroDataIOReader($io, $reader); |
| } |
| |
| /** |
| * @param string $codec |
| * @returns boolean true if $codec is a valid codec value and false otherwise |
| */ |
| public static function isValidCodec($codec) |
| { |
| return in_array($codec, self::validCodecs()); |
| } |
| |
| /** |
| * @returns array array of valid codecs |
| */ |
| public static function validCodecs() |
| { |
| return self::$validCodecs; |
| } |
| } |