Serialization IonToAvro

Serialization IonToAvro
Certified
Convert an ION file to the Avro format

Converts an Amazon ION file to Avro format using a JSON Avro schema. If no schema is provided, one is inferred by scanning up to numberOfRowsToScan rows. Use onBadLines to control whether records that fail validation cause an error, log a warning, or are silently skipped.

yaml
type: io.kestra.plugin.serdes.avro.IonToAvro

Examples

Convert a CSV dataset to the Avro format via ION.

yaml
id: divvy_tripdata
namespace: company.team

variables:
  file_id: "{{ execution.startDate | dateAdd(-3, 'MONTHS') | date('yyyyMM') }}"

tasks:
  - id: get_zipfile
    type: io.kestra.plugin.core.http.Download
    uri: "https://divvy-tripdata.s3.amazonaws.com/{{ render(vars.file_id) }}-divvy-tripdata.zip"

  - id: unzip
    type: io.kestra.plugin.compress.ArchiveDecompress
    algorithm: ZIP
    from: "{{ outputs.get_zipfile.uri }}"

  - id: convert
    type: io.kestra.plugin.serdes.csv.CsvToIon
    from: "{{ outputs.unzip.files[render(vars.file_id) ~ '-divvy-tripdata.csv'] }}"

  - id: to_avro
    type: io.kestra.plugin.serdes.avro.IonToAvro
    from: "{{ outputs.convert.uri }}"
    datetimeFormat: "yyyy-MM-dd' 'HH:mm:ss"
    schema: |
      {
        "type": "record",
        "name": "Ride",
        "namespace": "com.example.bikeshare",
        "fields": [
          {"name": "ride_id", "type": "string"},
          {"name": "rideable_type", "type": "string"},
          {"name": "started_at", "type": {"type": "long", "logicalType": "timestamp-millis"}},
          {"name": "ended_at", "type": {"type": "long", "logicalType": "timestamp-millis"}},
          {"name": "start_station_name", "type": "string"},
          {"name": "start_station_id", "type": "string"},
          {"name": "end_station_name", "type": "string"},
          {"name": "end_station_id", "type": "string"},
          {"name": "start_lat", "type": "double"},
          {"name": "start_lng", "type": "double"},
          {
            "name": "end_lat",
            "type": ["null", "double"],
            "default": null
          },
          {
            "name": "end_lng",
            "type": ["null", "double"],
            "default": null
          },
          {"name": "member_casual", "type": "string"}
        ]
      }

Properties

from *string

Source file URI

Pebble expression referencing an Internal Storage URI e.g. {{ outputs.mytask.uri }}.

dateFormat string

Defaultyyyy-MM-dd[XXX]

Format to use when parsing date

datetimeFormat string

Defaultyyyy-MM-dd'T'HH:mm[:ss][.SSSSSS][XXX]

Format to use when parsing datetime

Default value is yyyy-MM-dd'T'HH: mm[: ss][.SSSSSS][XXX]

decimalSeparator string

Default.

Character to recognize as decimal point (e.g. use ‘,’ for European data)

Default value is '.'

falseValues array

SubTypestring

Default["f","false","disabled","0","off","no",""]

Values to consider as False

inferAllFields booleanstring

Defaultfalse

Try to infer all fields

If true, schema inference scans all rows (ignoring numberOfRowsToScan) and attempts to infer all field types using trueValues, falseValues, and nullValues. This prevents fields that are null in the first scanned rows from being typed as NULL. If false, only the first numberOfRowsToScan rows are scanned, and booleans/nulls are inferred only on fields declared in the schema as null or bool.

nullValues array

SubTypestring

Default["","#N/A","#N/A N/A","#NA","-1.#IND","-1.#QNAN","-NaN","1.#IND","1.#QNAN","NA","n/a","nan","null"]

Values to consider as null

numberOfRowsToScan integerstring

Default100

Number of rows that will be scanned while inferring. The more rows scanned, the more precise the output schema will be

Only use when the 'schema' property is empty. Ignored for schema inference when inferAllFields is true — in that case, all rows are scanned.

onBadLines string

DefaultERROR

Possible Values

ERRORWARNSKIP

How to handle bad records (e.g., null values in non-nullable fields or type mismatches)

Can be ERROR, WARN, or SKIP.

pluginDefaultsRef string

Reference (ref) of the pluginDefaults to apply to this task.

schema string

The avro schema associated with the data

If empty, the task will try to infer the schema from the current data; use the 'numberOfRowsToScan' property if needed

strictSchema booleanstring

Defaultfalse

Whether to consider a field present in the data but not declared in the schema as an error

Default value is false

timeFormat string

DefaultHH:mm[:ss][.SSSSSS][XXX]

Format to use when parsing time

timeZoneId string

DefaultEtc/UTC

Timezone to use when no timezone can be parsed on the source

If null, the timezone defaults to UTC. Default value is the system timezone

trueValues array

SubTypestring

Default["t","true","enabled","1","on","yes"]

Values to consider as True

Outputs

size integer

Default0

The number of records converted

uri string

Formaturi

URI of a temporary result file

Metrics

records counter

Number of records converted

Core Plugins and tasks

Tasks that provide Kestra's built-in orchestration, I/O, and observability capabilities.

Core

Clickhouse

Tasks that connect to ClickHouse via JDBC to run queries and loads.

Data

Serialization IonToAvroCertifiedConvert an ION file to the Avro format

Serialization IonToAvro

More Plugins in this Category

Core Plugins and tasks

Clickhouse

Serialization IonToAvro
Certified
Convert an ION file to the Avro format