FLUID 0.7.7 (preview) — Recursive Streaming Types

0.7.7 lets a column declare nested data (structs, arrays and maps, to any depth) and lets an embedded-logic build describe the Kafka records it reads and how they become rows, in place of a SQL statement.

• Schema: fluid-schema-0.7.7.json · Reference: specs/0.7.7/fluid-spec.html · Diff: diff-0.7.6-to-0.7.7.md
• Compatibility: additive over 0.7.6. scripts/check-compat.py --from 0.7.6 --to 0.7.7 reports no narrowing change, so a valid 0.7.6 contract stays valid when it declares "0.7.7".

What is new

Every addition below is optional. These are the property paths present in 0.7.7 and absent in 0.7.6.

fluidVersion accepts "0.7.7" in addition to "0.7.3" to "0.7.6". No enum gains a value.

One existing rule changes shape. Through 0.7.6, an embedded-logic build's properties always required sql. In 0.7.7 it requires sql unless the build has a stream and its parameters.flink declares management; the build is then described by the stream and the mapping alone. A mapping without a stream is refused. Every document that satisfied the old rule satisfies the new one.

Nested columns

exposes:
  - exposeId: orders
    kind: table
    binding:
      platform: local
      format: parquet
      location: { path: "./orders.parquet" }
    contract:
      schema:
        - { name: order_id, type: string, required: true }
        - name: customer
          type: struct
          fields:
            - { name: id, type: string }
            - { name: country, type: string }
        - name: tags
          type: array
          element: { type: string }
        - name: attributes
          type: map
          key: { type: string, required: true }
          value: { type: string }
        - name: items
          type: array
          element:
            type: struct
            fields:
              - { name: sku, type: string }
              - { name: price, type: "decimal(18,2)" }

A column of type: struct lists its child columns in fields, each an ordinary column, so a child can carry anything a column carries. A column of type: array types its element, and a column of type: map types its key and value. element, key and value are each a nested type: type is required, required is optional, and fields, element, key and value nest again. Nothing in the schema limits the depth; the reference implementation stops at 32 levels (non-normative). A nested type has no name; the column it belongs to has one.

fields must not be empty. The schema does not tie these members to type: it accepts fields on a column whose type is string. Whether a platform can hold a nested column, and which map keys it supports, is for the implementation (non-normative). The reference implementation accepts only a map key declared required: true and of a string type, and refuses a bare struct, array or map with no nested types on BigQuery and AWS Glue.

Streams and mappings

build:
  id: ingest-order-items
  pattern: embedded-logic
  engine: flink
  properties:
    stream:
      kind: kafka                  # the only value in this preview
      topic: orders.v1
      format: json                 # json | avro-confluent
      schema:                      # the record value's fields, as columns
        - { name: event_id, type: string, required: true }
        - name: customer
          type: struct
          fields:
            - { name: id, type: string }
        - name: items
          type: array
          element:
            type: struct
            fields:
              - { name: sku, type: string }
              - { name: price, type: "decimal(18,2)" }
      key:                         # optional
        format: raw                # raw | json | avro-confluent
        schema:
          - { name: order_key, type: string }
        prefix: key_               # optional; prefixes the key's column names
      metadata:                    # optional: Kafka record metadata, each read as a column
        - { name: kafka_offset, from: offset }
    mapping:
      explode: { path: items, as: item }   # one row per element of items
      columns:                             # only the columns read from somewhere else
        - { name: customer_id, from: customer.id }
        - { name: sku, from: item.sku }
        - { name: price, from: item.price, cast: "decimal(18,2)" }
    parameters:
      flink:
        management: operator       # with a stream, this is what makes sql optional

stream describes the records a build reads, apart from the engine that reads them. kind, topic, format and schema are required. topic names one Kafka topic, in Kafka's legal characters (letters, digits, ., _ and -, at most 249, and not . or ..): not a pattern and not a list. format: json is one JSON object per record. format: avro-confluent is Avro in the Confluent Schema Registry wire format, and the declared schema is the reader schema. The key takes the same schema as a list of columns. metadata[].from is one of topic, partition, offset, timestamp, timestamp-type, headers and leader-epoch. Brokers, consumer groups and registries are not part of the stream: they are parameters of the engine.

mapping says how the records become rows. An output column the mapping does not list, nested fields included, is read from the stream column or field at the same path and keeps its type, so a stream landed as it is needs no mapping. columns[] lists the rest. name is a dotted path of identifiers. from is a dotted path, with a quoted key for a map entry (attributes['channel']), and cast is any column type. explode produces one output row per element of an array, named by as in the paths that follow; a record whose array is null or empty produces no row. Path segments are identifiers, so items[0] is not a path.

The grammar is the schema's; whether the types of the paths and the output columns agree is not. The reference implementation checks every output column's type and nullability against the path it is read from before anything deploys (non-normative).

sql, or a stream

An embedded-logic build says what it does in one of two ways. With sql, the statement is the build, and a stream may sit beside it to describe what the statement reads. With a stream and a mapping, the build has no sql and the platform writes the query; the schema accepts that only when the build's parameters.flink declares management (any value; the example above uses operator). A build with neither is refused with a missing sql, the same error as in 0.7.6.

build:
  id: ingest-order-items
  pattern: embedded-logic
  engine: flink
  properties:
    language: flink_sql
    sql: |
      INSERT INTO order_items SELECT event_id FROM orders
    stream:
      kind: kafka
      topic: orders.v1
      format: json
      schema:
        - { name: event_id, type: string }

A complete contract

The reference implementation can deploy a managed Flink job for a build like this one: a Kubernetes cluster that runs the Flink Kubernetes Operator, writing an Iceberg REST catalog table. That is behaviour of one implementation (non-normative), and it needs operator-side allow-lists for the cluster context, namespace, image, service account and secrets the contract names. The contract below validates against the 0.7.7 JSON Schema (Draft 2020-12) and with data-product-forge 0.27.0 (fluid validate prints ✅ Valid FLUID contract (schema v0.7.7)). fluid validate refuses a managed Flink build that has no build id or no Iceberg output named in outputs, so the contract carries both.

fluidVersion: "0.7.7"
kind: DataProduct
id: commerce.order_items
name: Order items
metadata:
  owner: { team: data-platform, email: data-platform@example.com }
build:
  id: ingest-order-items
  pattern: embedded-logic
  engine: flink
  outputs: [order_items]
  execution:
    trigger:
      type: continuous
  properties:
    language: flink_sql
    stream:
      kind: kafka
      topic: orders.v1
      format: json
      schema:
        - { name: event_id, type: string, required: true }
        - name: customer
          type: struct
          fields:
            - { name: id, type: string }
        - name: items
          type: array
          element:
            type: struct
            fields:
              - { name: sku, type: string }
              - { name: price, type: "decimal(18,2)" }
      metadata:
        - { name: kafka_offset, from: offset }
    mapping:
      explode: { path: items, as: item }
      columns:
        - { name: customer_id, from: customer.id }
        - { name: sku, from: item.sku }
        - { name: price, from: item.price }
    parameters:
      flink:
        management: operator
        source:
          bootstrapServers: kafka:9092
          groupId: forge-commerce-order-items-v1
        deployment:
          name: commerce-order-items
          namespace: data-products
          context: development
          serviceAccount: flink
          image: forge-flink@sha256:0000000000000000000000000000000000000000000000000000000000000000
          checkpointUri: s3://flink-state/order-items/checkpoints
          savepointUri: s3://flink-state/order-items/savepoints
          haUri: s3://flink-state/order-items/ha
          secretRefs: [flink-storage]
exposes:
  - exposeId: order_items
    kind: table
    binding:
      platform: aws
      format: iceberg
      location:
        catalog: rest
        uri: http://iceberg-rest:8181
        warehouse: s3://warehouse/
        database: commerce
        table: order_items
    contract:
      schema:
        - { name: event_id, type: string, required: true }
        - { name: customer_id, type: string }
        - { name: sku, type: string }
        - { name: price, type: "decimal(18,2)" }
        - { name: kafka_offset, type: long, required: true }

parameters is an open object, so the flink settings above are not checked by the schema. The image digest is a placeholder.

References

• Schema Versions — status of every version, and how to choose fluidVersion.
• Changelog: 0.7.6 → 0.7.7.
• 0.7.6 — the stable version this preview is built on, including the continuous trigger used above.