From 748dd88e930952b07170c14ea346e0ed13e629f4 Mon Sep 17 00:00:00 2001 From: Hiroshi Hatake Date: Thu, 8 Oct 2026 15:23:24 +0900 Subject: [PATCH] pipeline: outputs: s3: document schema-directed Arrow and Parquet output - Note log_key is unsupported with format arrow and format parquet. - Also note it's unsupported with the deprecated compression arrow and compression parquet. - Update the format row to say Arrow or Parquet. - Replace "nullable" with plain-language description - Replace "NUL" with "null" in embedded bytes sentence Signed-off-by: Hiroshi Hatake Signed-off-by: Eric D. Schabell --- pipeline/outputs/s3.md | 103 +++++++++++++++++++++++++++++++++++++++-- 1 file changed, 100 insertions(+), 3 deletions(-) diff --git a/pipeline/outputs/s3.md b/pipeline/outputs/s3.md index 018255d30..c076fa529 100644 --- a/pipeline/outputs/s3.md +++ b/pipeline/outputs/s3.md @@ -16,7 +16,7 @@ The plugin can upload data to S3 using the [multipart upload API](https://docs.a The plugin lets you specify a maximum file size, and a timeout for uploads. A file will be created in S3 when the maximum size or the timeout is reached, whichever comes first. -Records are stored in files in S3 as newline delimited JSON. +By default, records are stored in files in S3 as newline-delimited JSON. See [AWS Credentials](../../administration/aws-credentials.md) for details about fetching AWS credentials. @@ -51,11 +51,11 @@ The [Prometheus success/retry/error metrics values](../../administration/monitor | `endpoint` | Custom endpoint for the S3 API. Endpoints can contain scheme and port. Set this parameter to send data to an S3-compatible object store other than Amazon S3, such as MinIO or Google Cloud Storage. Requests use path-style addressing (`endpoint/bucket/key`) and are signed with the configured `region`. | _none_ | | `external_id` | Specify an external ID for the STS API. Can be used with the `role_arn` parameter if your role requires an external ID. | _none_ | | `file_delivery_attempt_limit` | File delivery attempt limit. | `1` | -| `format` | Set the output format. Supported values: `json_lines`, `otlp_json`, `parquet`. When set to `parquet`, records are converted to Apache Parquet columnar format (requires Apache Arrow Parquet support at compile time). The `compression` option controls the page-level `codec` inside the Parquet file. When set to `otlp_json`, the `log_key` option isn't supported and only `logs` event chunks are converted. | `json_lines` | +| `format` | Set the output format. Supported values: `json_lines`, `otlp_json`, `arrow`, `parquet`. Arrow output requires Apache Arrow support at compile time and accepts only unset, `none`, or `zstd` compression. Parquet output also requires Apache Arrow Parquet support; the `compression` option controls the page-level `codec` inside the Parquet file. The `log_key` option isn't supported with Arrow or Parquet. Set `parquet.schema_file` for [schema-directed Arrow or Parquet conversion](#schema-directed-arrow-and-parquet-output). When set to `otlp_json`, the `log_key` option isn't supported and only `logs` event chunks are converted. | `json_lines` | | `host` | IP address or hostname of the target HTTP server. | `127.0.0.1` | | `json_date_format` | Specify the format of the date. Accepted values: `double`, `epoch`, `epoch_ms`, `iso8601` (2018-05-30T09:39:52.000681Z), `_java_sql_timestamp_` (2018-05-30 09:39:52.000681). | _none_ | | `json_date_key` | Specify the name of the date key in the output record. To disable the time key, set the value to `false`. | `date` | -| `log_key` | By default, the whole log record is sent to S3. When specifying a key name with this option, only the value of that key is sent to S3. For example, when using Docker you can specify `log_key log` and only the log message is sent to S3. The key name must match exactly. See [Key matching for `log_key`](#key-matching-for-log_key). | _none_ | +| `log_key` | By default, the whole log record is sent to S3. When specifying a key name with this option, only the value of that key is sent to S3. For example, when using Docker you can specify `log_key log` and only the log message is sent to S3. The key name must match exactly. Not supported with Arrow or Parquet output (`format arrow`, `format parquet`, or the deprecated `compression arrow` and `compression parquet`); the plugin fails to initialize if both are set. See [Key matching for `log_key`](#key-matching-for-log_key). | _none_ | | `log_level` | Specifies the log level for output plugin. If not set here, plugin uses global log level in `service` section. | `info` | | `log_suppress_interval` | Suppresses log messages from output plugin that appear similar within a specified time interval. `0` means no suppression. | `0` | | `match` | Set a tag pattern to match records that output should process. Exact matches or wildcards (for example `*`). | _none_ | @@ -74,6 +74,7 @@ The [Prometheus success/retry/error metrics values](../../administration/monitor | `net.tcp_keepalive_interval` | Interval between TCP keepalive probes when no response is received on a `keepidle` probe. | `-1` | | `net.tcp_keepalive_probes` | Number of unacknowledged probes to consider a connection dead. | `-1` | | `net.tcp_keepalive_time` | Interval between the last data packet sent and the first TCP keepalive probe. | `-1` | +| `parquet.schema_file` | Path to an Arrow integration-format schema object for direct MessagePack-to-Arrow conversion with `format arrow` or `format parquet`. Without this option, JSON conversion and schema inference remain unchanged. See [Schema-directed Arrow and Parquet output](#schema-directed-arrow-and-parquet-output). | _none_ | | `part_delivery_attempt_limit` | File part delivery attempt limit. | `1` | | `part_size` | Size of each part when uploading blob files. | `25M` | | `port` | TCP port of the target HTTP server. | `80` | @@ -221,6 +222,102 @@ Parquet format requires Apache Arrow Parquet support at compile time: The `AWS for Fluent Bit` version 3 container image includes these dependencies by default. +## Schema-directed Arrow and Parquet output + +Set `parquet.schema_file` with `format arrow` or `format parquet` to convert MessagePack records directly to Arrow using an explicit schema. Without this option, S3 keeps its existing JSON conversion and schema inference behavior. + +Arrow support must be enabled at build time with `-DFLB_ARROW=On` and the `arrow-glib` dependency. For `format parquet`, the build also requires Parquet support and `parquet-glib`. See [Build requirements](#build-requirements). + +{% tabs %} +{% tab title="fluent-bit.yaml" %} + +```yaml +pipeline: + outputs: + - name: s3 + match: '*' + bucket: example + region: us-east-1 + format: parquet + compression: zstd + use_put_object: on + parquet.schema_file: /etc/fluent-bit/schema.json +``` + +{% endtab %} +{% tab title="fluent-bit.conf" %} + +```text +[OUTPUT] + Name s3 + Match * + Bucket example + Region us-east-1 + Format parquet + Compression zstd + Use_Put_Object On + Parquet.Schema_File /etc/fluent-bit/schema.json +``` + +{% endtab %} +{% endtabs %} + +### Schema file + +The file contains an Arrow integration-format **schema object**, not a JSON Schema document or an Avro schema. For example, save the following as `/etc/fluent-bit/schema.json`: + +```json +{ + "fields": [ + { + "name": "message", + "nullable": true, + "type": {"name": "utf8"}, + "children": [] + }, + { + "name": "count", + "nullable": false, + "type": {"name": "int", "bitWidth": 64, "isSigned": true}, + "children": [] + } + ] +} +``` + +Supported types are: + +| Type name | Supported attributes | +| --- | --- | +| `bool` | None. | +| `utf8` | None. | +| `int` | `bitWidth`: `8`, `16`, `32`, or `64`; `isSigned`: `true` or `false`. | +| `floatingpoint` | `precision`: `SINGLE` or `DOUBLE`. | + +Field order is preserved. Field names must be unique and nonempty, and every field must specify `nullable`. The `children` attribute can be omitted or empty. Nested Arrow types, dictionaries, metadata, and other schema or type attributes are rejected. + +The schema is loaded once at startup. An invalid schema prevents startup. + +### Record conversion + +Columns select exact top-level record body keys. Extra keys are ignored. Missing or nil values become null only for columns that allow null values. Conversion fails for missing or nil values in columns that don't allow null values, wrong scalar types, invalid UTF-8, non-finite numbers, and integer overflow. + +Numeric strings aren't coerced. Integer-to-floating-point conversion follows the chosen floating-point precision and can round. Maps and arrays in a `utf8` column are serialized to JSON text. Scalar strings are appended directly, including embedded null bytes. + +Event timestamps and metadata aren't added as columns. The `json_date_key` and `json_date_format` options don't apply to this conversion path. Put a timestamp in the record body when it needs to be a column. + +### Buffering and schema changes + +S3 buffers the original MessagePack log events and builds the table at upload time. Schema buffers are stored under `store_dir/bucket/msgpack-`, where the hash covers the exact schema file bytes. This isolates them from JSON buffers written without a schema and buffers written with another schema. + +Restart with the same schema contents to resume pending uploads. Changing even schema whitespace selects a new directory. Pending buffers in the previous directory require the previous schema configuration to drain. + +Conversion failures follow the existing S3 retry and quarantine settings and never upload raw MessagePack as columnar data. + +### Integration tests + +The integration tests that inspect decoded column values require `pyarrow` in addition to the integration suite's standard Python dependencies. In the Fluent Bit source repository, run the S3 scenario with `-k 'schema or format'`, then repeat with `VALGRIND=1 VALGRIND_STRICT=1` on Linux or `LEAKS=1 LEAKS_STRICT=1` on macOS. + ## Permissions The plugin requires the following AWS IAM permissions: