Asset Triggers, Tasks, Functions, and Exporters

This reference lists the triggers, tasks, Pebble function, and exporters that read or act on assets.

For how to attach assets to a task, see the asset declaration reference. For the full property schemas, follow the plugin links in each section.

Summary

EventTrigger and AssetShipper run inside Kestra. The tasks and the FreshnessTrigger from the io.kestra.plugin.kestra.ee package call the Kestra API, so they accept the connection properties described at the end of this page.

Metadata queries

EventTrigger, FreshnessTrigger, List, and PurgeAssets accept a metadataQuery list. Each entry compares one top-level metadata key with a value:

An asset matches only when it satisfies every entry. Use each field only once per operator in the same list.

metadataQuery:
  - field: environment
    type: EQUAL_TO
    value: prod
  - field: model_layer
    type: NOT_EQUAL_TO
    value: staging

Asset type filters

Filters such as assetType and types compare the full type string exactly. To match a built-in type, use its full class name, for example io.kestra.plugin.ee.assets.Table. A shortened value such as TABLE matches only assets whose type is literally TABLE.

EventTrigger

io.kestra.plugin.ee.assets.EventTrigger starts an execution each time a matching asset event occurs.

Kestra emits each event in these situations:

The execution exposes the asset that matched as trigger.asset:

This flow starts whenever a staging Table asset in company.data is created or updated:

id: refresh_marts_on_staging_change
namespace: company.data

tasks:
  - id: log_change
    type: io.kestra.plugin.core.log.Log
    message: "{{ trigger.asset.id }} was {{ trigger.asset.event }} at {{ trigger.asset.eventTime }}"

triggers:
  - id: staging_table_changed
    type: io.kestra.plugin.ee.assets.EventTrigger
    namespace: company.data
    assetType: io.kestra.plugin.ee.assets.Table
    events:
      - CREATED
      - UPDATED
    metadataQuery:
      - field: model_layer
        type: EQUAL_TO
        value: staging

For a complete walkthrough, see Run a flow when a table changes.

FreshnessTrigger

io.kestra.plugin.kestra.ee.assets.FreshnessTrigger polls for assets whose last update is older than a threshold.

Staleness is based on the asset’s updated timestamp. Any write refreshes it: a task that declares the asset as an output, the API, or the UI. Lineage and usage records don’t count.

When at least one asset is stale, the trigger starts one execution for all stale assets found in that check. It starts a new execution at every interval for as long as the assets stay stale.

The execution exposes the stale assets as the list trigger.assets. Each item contains:

This flow checks every 30 minutes for production mart tables that haven’t been written in 6 hours and logs each one:

id: stale_marts_monitor
namespace: company.monitoring

tasks:
  - id: for_each_stale_asset
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ trigger.assets }}"
    tasks:
      - id: log_stale
        type: io.kestra.plugin.core.log.Log
        message: "{{ fromJson(item.value).id }} last updated {{ fromJson(item.value).lastUpdated }} ({{ fromJson(item.value).staleDuration }} seconds ago)"

triggers:
  - id: stale_marts
    type: io.kestra.plugin.kestra.ee.assets.FreshnessTrigger
    assetType: io.kestra.plugin.ee.assets.Table
    maxStaleness: PT6H
    interval: PT30M
    metadataQuery:
      - field: environment
        type: EQUAL_TO
        value: prod
      - field: model_layer
        type: EQUAL_TO
        value: mart

Loop passes each stale asset to its child tasks as a JSON string, so read its fields with fromJson(item.value).

For a complete walkthrough, see Detect and refresh stale data.

Set

io.kestra.plugin.kestra.ee.assets.Set creates an asset.

Set fails if an asset with the same ID already exists. To create or update an asset as part of the work that produces it, declare it in the producing task’s assets.outputs instead. That path updates existing assets and records lineage, which Set doesn’t.

id: register_customers_table
namespace: company.data

tasks:
  - id: register
    type: io.kestra.plugin.kestra.ee.assets.Set
    namespace: company.data
    assetId: analytics.customers_by_country
    assetType: io.kestra.plugin.ee.assets.Table
    displayName: Customers by country
    assetDescription: Customer counts aggregated by country.
    metadata:
      owner: data-team
      environment: prod

List

io.kestra.plugin.kestra.ee.assets.List queries assets.

Every fetch type also returns the number of results as outputs.<task_id>.size.

id: list_data_team_tables
namespace: company.data

tasks:
  - id: list_tables
    type: io.kestra.plugin.kestra.ee.assets.List
    namespace: company.data
    types:
      - io.kestra.plugin.ee.assets.Table
    metadataQuery:
      - field: owner
        type: EQUAL_TO
        value: data-team
    fetchType: FETCH

  - id: log_ids
    type: io.kestra.plugin.core.log.Log
    message: "{{ outputs.list_tables.assets | jq('.[].id') }}"

Delete

io.kestra.plugin.kestra.ee.assets.Delete deletes one asset by its required assetId. Deleting an asset emits a DELETED event and removes the asset from the dependencies graph.

- id: delete_asset
  type: io.kestra.plugin.kestra.ee.assets.Delete
  assetId: analytics.customers_by_country

PurgeAssets

io.kestra.plugin.kestra.ee.assets.PurgeAssets hard-deletes assets, asset usage records, and lineage events older than a cutoff date. Locked assets are skipped.

The purge reference lists every property and the filter combinations each record type supports. For retention strategy, see Purging data.

Acquire and Release

io.kestra.plugin.kestra.ee.locks.Acquire locks an asset for the current execution. While the lock is held, writes from other executions fail. The execution that holds the lock can still write to the asset, and each write extends the lock.

If another owner already holds the lock, Acquire fails with HTTP 423. Add a retry to wait for the lock.

io.kestra.plugin.kestra.ee.locks.Release releases the lock on its required assetId if the current execution holds it. If the lock has expired or belongs to another owner, Release does nothing, so it’s safe to call unconditionally.

Acquire needs the LOCK permission on the ASSET resource, and Release needs UNLOCK.

For locking from the UI and how locks affect each write path, see Lock an asset. For a complete walkthrough, see Prevent concurrent changes to a resource.

assets() Pebble function

The assets() function returns the assets that match its arguments. Use it anywhere an expression is allowed, for example to fill a SELECT or MULTISELECT input with live asset IDs.

assets(id=null, type=null, namespace=null, metadata=null)

When you pass namespace, Kestra also checks allowed namespaces: if any namespace in the tenant restricts its allowed namespaces without listing the namespace you pass, the function fails with All namespaces are not allowed. Namespaces allow all others by default, so this only matters once you restrict them. To avoid the check, filter by type or metadata instead.

The function returns a list. Each item has id, namespace, type, displayName, description, metadata, tenantId, created, updated, and deleted. Use the jq filter to extract fields.

This flow lets the user pick from the production Table assets in company.data:

id: pick_tables
namespace: company.data

inputs:
  - id: tables
    type: MULTISELECT
    expression: '{{ assets(type="io.kestra.plugin.ee.assets.Table", namespace="company.data", metadata={"environment": "prod"}) | jq(".[].id") }}'

tasks:
  - id: for_each_table
    type: io.kestra.plugin.core.flow.Loop
    values: "{{ inputs.tables }}"
    tasks:
      - id: log_table
        type: io.kestra.plugin.core.log.Log
        message: "Selected {{ item.value }}"

This expression reads one metadata value from a known asset:

message: "{{ assets(id='reports.weekly_csv') | jq('.[0].metadata.path') | first }}"

AssetShipper

io.kestra.plugin.ee.assets.AssetShipper exports asset lineage events to one or more destinations. Each lineage event describes one task run: the flow and task, the execution, the run state, and the input and output assets.

The task runs incrementally. On its first run, it exports events from the last lookbackPeriod. It then stores the timestamp of the last exported event in the KV Store and continues from there on the next run. Schedule it with a Schedule trigger.

Outputs: size (number of events exported), endFetchedDate, and outputs (one entry per exporter).

FileAssetExporter

io.kestra.plugin.ee.assets.FileAssetExporter writes events to files in internal storage. Use it when no exporter targets your destination, then send the files on with another task.

id: export_lineage_to_file
namespace: company.data

tasks:
  - id: export
    type: io.kestra.plugin.ee.assets.AssetShipper
    assetExporters:
      - id: file
        type: io.kestra.plugin.ee.assets.FileAssetExporter
        format: JSON

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

OpenLineageAssetExporter

io.kestra.plugin.ee.openlineage.OpenLineageAssetExporter sends each event as an OpenLineage run event over HTTP. The Kestra flow and task become the OpenLineage job, and input and output assets become datasets.

Every asset of a mapped type must have the mapped metadata keys. For a complete walkthrough, see Send asset lineage to OpenLineage.

Dashboards

The io.kestra.plugin.ee.dashboard.data.Assets data source charts your current asset inventory in a custom dashboard. See Assets data source for fields and examples.

Connection properties for API-based tasks

Set, List, Delete, PurgeAssets, Acquire, Release, and FreshnessTrigger call the Kestra API. By default, they connect to the current instance with the authentication configured for SDK tasks, and they act on the current tenant. Override these defaults when needed:

If kestra.url is an address that only works from outside the worker, such as a host port mapped to a Docker container, set kestraUrl to an address the worker can reach. Otherwise the task fails with a connection error.

The credentials need the matching ASSET permissions. See the ASSET permissions reference.