Core Plugins and tasks AssetShipper

Core Plugins and tasks AssetShipper

Certified

Forward asset lineage to one or more desired destinations.

The Asset Shipper task extracts asset lineage events from the Kestra backend and loads them to desired destinations including OpenLineage and file storage.

The task works incrementally in batches:

  1. Determines the starting timestamp using either:
    • The last successfully processed assert lineage event's timestamp (persisted in KV Store using the offsetKey)
    • Current time minus lookbackPeriod duration if no previous state exists
  2. Sends retrieved asset lineage events through configured assetExporters
  3. Stores the timestamp of the last processed asset lineage event to maintain state between executions
  4. Subsequent runs continue from the last stored timestamp

This incremental approach ensures reliable asset lineage event forwarding without gaps or duplicates

yaml
type: io.kestra.plugin.ee.assets.AssetShipper

Ship asset lineage events to OpenLineage

yaml
id: assetShipper
namespace: company.team

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@daily"

tasks:
  - id: logSync
    type: io.kestra.plugin.ee.assets.AssetShipper
    assetExporters:
      - id: file
        type: io.kestra.plugin.ee.openlineage.OpenLineageAssetExporter
        uri: http://localhost:3000DO
Properties
Min items1

List of asset exporters

The list of asset exporters to use for sending asset lineage

Definitions

This task is designed to be used when no external tool's asset shipper satisfies requirements. You can ship asset lineage events to the internal storage and use any other Kestra task to send it to a remote location.

Example

Ship asset lineage events to the internal storage

yaml
id: assetShipper
namespace: company.team

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@daily"

tasks:
  - id: logSync
    type: io.kestra.plugin.ee.assets.AssetShipper
    assetExporters:
      - id: file
        type: io.kestra.plugin.ee.assets.FileAssetExporter
        format: JSON
        maxLinesPerFile: 100
id*Requiredstring
Validation RegExp^[a-zA-Z0-9][a-zA-Z0-9_-]*
Min length1
type*Requiredobject
formatstring
DefaultION
Possible Values
IONJSON

Format of the exported files

This property defines the format of the exported files.

logFilePrefixstring
Defaultkestra-log-file

Prefix of the log files

This property sets the prefix of the log files name. The full file name will be logFilePrefix-localDateTime.json/ion.

maxLinesPerFileintegerstring

Maximum number of lines per file

This property specifies the maximum number of lines per log file.

Sends Kestra asset lineage events to an OpenLineage-compatible endpoint over HTTP. Uses the asset namespace and id as dataset namespace and name unless a per-asset-type mapping overrides them. Ensure the URI is reachable and accepts OpenLineage events; no authentication is configured here.

Example

Ship asset lineage events to OpenLineage

yaml
id: assetShipper
namespace: company.team

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@daily"

tasks:
  - id: logSync
    type: io.kestra.plugin.ee.assets.AssetShipper
    assetExporters:
      - id: file
        type: io.kestra.plugin.ee.openlineage.OpenLineageAssetExporter
        uri: http://localhost:3000

Ship asset lineage events to OpenLineage, specifying mappings for Table an Dataset asset types

yaml
id: assetShipper
namespace: company.team

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "@daily"

tasks:
  - id: logSync
    type: io.kestra.plugin.ee.assets.AssetShipper
    assetExporters:
      - id: file
        type: io.kestra.plugin.ee.openlineage.OpenLineageAssetExporter
        uri: http://localhost:3000
        mappings:
          io.kestra.plugin.ee.assets.Table:
            namespace: namespace
            name: name
          io.kestra.plugin.ee.assets.Dataset:
            namespace: namespace
            name: name
id*Requiredstring
Validation RegExp^[a-zA-Z0-9][a-zA-Z0-9_-]*
Min length1
type*Requiredobject
uri*Requiredstring

OpenLineage endpoint URI

HTTP destination for lineage events; rendered from the flow context

mappingsobject

Dataset field overrides by asset type

Optional map of asset class names to metadata keys for dataset namespace and name; defaults to asset namespace and id

Asset namespace to search

The asset namespace to use to filter asset lineage

SubTypestring

Asset types to search

The list of asset types to use to filter asset lineage

Delete asset usage after export

The asset shipper will delete the exported asset usage

Flow ID to search

The flow identifier to use to filter asset lineage

Flow namespace to search

The flow namespace to use to filter asset lineage

DefaultP1D

Starting duration before now

If no previous execution or state exists, the fetch start date is set to the current time minus this duration

Prefix of the KVStore key

The prefix of the KVStore key that contains the last execution's end fetched date

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

Task ID to search

The task identifier to use to filter asset lineage

Formatdate-time

The zoned date-time of the last fetched event, used as the starting date for the next execution

The outputs generated by each asset lineage event exporter

Definitions

The number of asset lineage fetched.

More Plugins in this Category

Compression
Compression

Tasks that compress and decompress files and archives for Kestra flows.

Core

6 tasks

4 blueprints

Cryptography
Cryptography

Tasks that secure files in Kestra pipelines using OpenPGP encryption, decryption, and signature validation.

CoreInfrastructure

2 tasks