Core Plugins and tasks DeduplicateItems
CertifiedDeduplicate a line-oriented file by key.
Core Plugins and tasks DeduplicateItems
Deduplicate a line-oriented file by key.
Reads the file twice: first to map each key (from expr) to its last occurrence offset, then to write only those last occurrences to a new file. Avoids loading the full file in memory.
Use for ordered “keep-last” semantics; expression can reference columns directly.
type: io.kestra.plugin.core.storage.DeduplicateItemsExamples
Remove duplicate customer emails from a CSV file.
id: deduplicate_items
namespace: company.team
tasks:
- id: generate_files
type: io.kestra.plugin.scripts.shell.Script
script: |
cat <<EOF > my_data.csv
order_id,customer_name,customer_email,product_id,price
1,Kelly Olsen,kelly@example.com,20,166.89
2,Miguel Moore,mccarthylee@example.net,14,171.63
3,Kelly Olsen,kelly@example.com,20,166.89
4,Jessica White,jessica@example.com,12,50.62
5,Jessica White,jessica@example.com,12,50.62
EOF
outputFiles:
- "my_data.csv"
- id: csv_to_ion
type: io.kestra.plugin.serdes.csv.CsvToIon
from: "{{ outputs.generate_files.outputFiles['my_data.csv'] }}"
- id: dedup
type: io.kestra.plugin.core.storage.DeduplicateItems
from: "{{ outputs.csv_to_ion.uri }}"
expr: "{{ customer_email }}"
Properties
expr *string
The Pebble expression to extract the deduplication key from each item
Headers from the file can be referenced directly e.g. {{ customer_email }}
from *string
The file to be deduplicated
Pebble expression referencing an Internal Storage URI e.g. {{ outputs.mytask.uri }}.
assets
Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset failure behavior
Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150Outputs
droppedItemsTotal integer
The total number of items that was dropped by the task
numKeys integer
The number of distinct keys observed by the task
processedItemsTotal integer
The total number of items that was processed by the task
uri string
uriThe deduplicated file URI
