id: change-feed-newest-wins-dedup-partition-pipeline
namespace: company.team
description: |
Takes a CSV or JSONL change feed (several versions of the same record, newest first), keeps the newest version of each
record, drops tombstones, splits the result into partitions processed in parallel by a worker subflow, and reassembles
a clean file and a rejects file. Every transform is a core storage task; there is no script in this flow.
labels:
pipeline: change-feed
style: no-code
inputs:
- id: source_file
type: FILE
required: false
description: |
The feed to process (CSV with a header, or JSONL with one object per line). Leave it empty to run on the built-in
sample feed, which the flow writes itself with `storage.Write`. Columns expected by the sample worker:
customer_id, updated_at, email, country, plan, status.
- id: source_format
type: SELECT
values: [ csv, jsonl ]
defaults: csv
description: Format of `source_file`. The built-in sample is CSV, so `jsonl`
needs an uploaded file.
- id: source_order
type: SELECT
values: [ newest_first, oldest_first ]
defaults: newest_first
description: |
Order of the versions in the feed. `DeduplicateItems` keeps the LAST occurrence of a key, so a newest-first feed
(an export sorted by `updated_at DESC`, or a log read from the tail) is reversed first. Choose `oldest_first` for
an append-only feed, where the last line already is the newest version.
- id: tombstone_status
type: STRING
defaults: deleted
description: Rows whose `status` equals this value after de-duplication are
removed. They mark deleted records.
- id: split_by
type: SELECT
values: [ rows, bytes ]
defaults: rows
description: How to partition the de-duplicated file for the worker subflow, by
row count or by chunk size.
- id: rows_per_partition
type: INT
defaults: 5
description: Rows per partition when `split_by` is `rows`. 5 fits the 14-row
sample; use thousands on a real feed.
- id: bytes_per_partition
type: STRING
defaults: 1KB
description: Chunk size when `split_by` is `bytes`, as `<number>KB`, `MB` or
`GB` (Split needs KB or more).
- id: max_reject_ratio
type: FLOAT
defaults: 0.5
description: |
Passed to every partition worker. A partition where more than this share of rows fails validation is refused
as a whole (it usually means shifted columns or a wrong delimiter in that part of the feed), its rows go to the
rejects file and the run fails unless `fail_on_partition_failure` is false. Between 0 and 1.
- id: partition_flow_id
type: STRING
defaults: change-feed-partition-worker
description: Id of the worker subflow in this namespace. The worker flow is
given in the blueprint description.
- id: fail_on_partition_failure
type: BOOL
defaults: true
description: true fails the run when any partition failed (after the clean and
rejects files are written). false ends the run in SUCCESS with the
failures in the labels and the dead-letter entry.
- id: kestra_api_url
type: STRING
defaults: http://localhost:8080
description: Base URL of this Kestra instance, only used to build execution
links in alerts when `kestra.url` is not configured.
variables:
dead_letter_key: change_feed_failed_partitions
failure_key: change_feed_last_failure
sample_feed: |
customer_id,updated_at,email,country,plan,status
C010,2026-10-20T09:30:00Z,olga@example.com,PL,pro,active
C003,2026-10-20T09:10:00Z,maria@example.com,ES,pro,active
C001,2026-10-20T08:50:00Z,ana.new@example.com,PT,team,active
C004,2026-10-20T08:40:00Z,karl@example.com,DE,free,deleted
C002,2026-10-20T08:30:00Z,jean-at-example.com,FR,free,active
C008,2026-10-20T08:20:00Z,sam@example.com,GB,team,active
C001,2026-10-20T08:00:00Z,ana@example.com,PT,free,active
C009,2026-10-20T07:45:00Z,yuki@example.com,JP,pro,active
C004,2026-10-19T18:00:00Z,karl@example.com,DE,free,active
C005,2026-10-19T17:00:00Z,li@example.com,CN,pro,active
C006,2026-10-19T16:00:00Z,nope,XXX,pro,active
C011,2026-10-19T15:30:00Z,noor@example.com,AE,free,active
C007,2026-10-19T15:00:00Z,pat@example.com,IE,free,active
C007,2026-10-19T14:30:00Z,pat@example.com,IE,free,deleted
C002,2026-10-19T14:00:00Z,jean@example.com,FR,free,active
C012,2026-10-19T13:00:00Z,tom@example.com,US,pro,active
C013,2026-10-19T12:00:00Z,eva@example.com,SE,team,active
C008,2026-10-19T11:00:00Z,sam.old@example.com,GB,free,active
C014,2026-10-19T10:00:00Z,raj@example.com,IN,pro,active
C015,2026-10-19T09:00:00Z,,BR,free,active
checks:
- when: "{{ inputs.source_format == 'csv' or inputs.source_file is defined }}"
message: source_format jsonl needs an uploaded source_file. The built-in sample
feed is CSV.
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ inputs.split_by != 'rows' or inputs.rows_per_partition >= 1 }}"
message: rows_per_partition must be at least 1 when split_by is rows.
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ inputs.split_by != 'bytes' or ([inputs.bytes_per_partition | trim |
upper] | jq('.[0] | test(\"^[0-9]+(KB|MB|GB)$\")') | first) }}"
message: bytes_per_partition must look like 512KB, 16MB or 1GB (Split refuses
sizes below 1KB).
style: ERROR
behavior: BLOCK_EXECUTION
- when: "{{ inputs.max_reject_ratio >= 0 and inputs.max_reject_ratio <= 1 }}"
message: max_reject_ratio is a share of the rows of a partition, between 0 and 1.
style: ERROR
behavior: BLOCK_EXECUTION
concurrency:
limit: 1
behavior: QUEUE
sla:
- id: max_runtime
type: MAX_DURATION
duration: PT2H
behavior: FAIL
labels:
change_feed_sla: breached
triggers:
- id: after_export
type: io.kestra.plugin.core.trigger.Flow
description: |
Starts one run per successful execution of the export flow `customer-change-export` in this namespace and takes
the feed from that flow's `export_file` output. Point `flowId` at your own export flow, or replace this trigger
by an S3, GCS or SFTP file trigger passing `trigger.uri` as `source_file`.
inputs:
source_file: "{{ trigger.outputs.export_file }}"
source_format: csv
dependsOn:
- namespace: company.team
flowId: customer-change-export
states: [ SUCCESS ]
tasks:
- id: write_sample_feed
type: io.kestra.plugin.core.storage.Write
description: Only when no file was uploaded. 20 versions of 15 records, newest
first, with two tombstones, three invalid rows and a record that was
deleted then re-created.
runIf: "{{ (inputs.source_file ?? '') == '' }}"
extension: .csv
content: "{{ vars.sample_feed }}"
- id: read_feed
type: io.kestra.plugin.core.flow.Switch
description: Converts the feed to ION, the row format every storage task below
reads. The record count comes from the converter.
value: "{{ inputs.source_format }}"
cases:
csv:
- id: csv_to_ion
type: io.kestra.plugin.serdes.csv.CsvToIon
description: "`onBadLines: ERROR` (the default) fails the whole run on a ragged
row; a feed that cannot be parsed must not be half-loaded."
from: "{{ inputs.source_file ?? outputs.write_sample_feed.uri }}"
header: true
jsonl:
- id: jsonl_to_ion
type: io.kestra.plugin.serdes.json.JsonToIon
from: "{{ inputs.source_file }}"
newLine: true
- id: feed_size
type: io.kestra.plugin.core.storage.Size
description: Bytes of the ION feed, for the summary and the labels (the input
file size is not exposed by a FILE input).
uri: "{{ outputs.csv_to_ion.uri ?? outputs.jsonl_to_ion.uri }}"
- id: refuse_empty_feed
type: io.kestra.plugin.core.execution.Assert
description: A header without rows is treated as an upstream failure, not as a
clean run with nothing to do.
conditions:
- "{{ (outputs.csv_to_ion.size ?? outputs.jsonl_to_ion.size) > 0 }}"
errorMessage: "The {{ inputs.source_format }} feed has no rows ({{
outputs.feed_size.size }} bytes of ION). Check the export that produced
it."
- id: feed_as_jsonl
type: io.kestra.plugin.serdes.json.IonToJson
description: |
`Reverse` works on text lines. On Kestra 2.0 the serdes and storage tasks write binary ION (one self-contained
record after another, no line breaks), and reversing that file byte-wise corrupts it. So the feed takes a short
detour through JSONL: one JSON object per line, reversed, read back into ION. Skipped for an oldest-first feed.
runIf: "{{ inputs.source_order == 'newest_first' }}"
from: "{{ outputs.csv_to_ion.uri ?? outputs.jsonl_to_ion.uri }}"
newLine: true
- id: reverse_feed
type: io.kestra.plugin.core.storage.Reverse
description: Last line first. After this the newest version of every record is
the last one, which is the one `DeduplicateItems` keeps.
runIf: "{{ inputs.source_order == 'newest_first' }}"
from: "{{ outputs.feed_as_jsonl.uri }}"
- id: reversed_feed_as_ion
type: io.kestra.plugin.serdes.json.JsonToIon
runIf: "{{ inputs.source_order == 'newest_first' }}"
from: "{{ outputs.reverse_feed.uri }}"
newLine: true
- id: dedupe_newest_wins
type: io.kestra.plugin.core.storage.DeduplicateItems
description: |
One row per `customer_id`, the last occurrence in file order, so the newest version after the reverse. Change
`expr` to your key (`{{ order_id }}`, or a composite such as `{{ account_id ~ '/' ~ sku }}`); it is evaluated
per row against the row's columns and cannot be taken from an input.
from: "{{ outputs.reversed_feed_as_ion.uri ?? outputs.csv_to_ion.uri ??
outputs.jsonl_to_ion.uri }}"
expr: "{{ customer_id }}"
- id: drop_tombstones
type: io.kestra.plugin.core.storage.FilterItems
description: |
Removes records whose NEWEST version is a delete. This filter has to run after the de-duplication: before it, an
older active version of a deleted record would survive, and a record deleted then re-created (C007 in the sample)
would be lost. The row context also sees `inputs`, which is how the status value is configurable.
from: "{{ outputs.dedupe_newest_wins.uri }}"
filterCondition: "{{ status == inputs.tombstone_status }}"
filterType: EXCLUDE
errorOrNullBehavior: INCLUDE
- id: split_rows
type: io.kestra.plugin.core.storage.Split
description: Row-count partitions. `Split` reads the records, so it works on the
binary ION file directly.
runIf: "{{ inputs.split_by == 'rows' }}"
from: "{{ outputs.drop_tombstones.uri }}"
rows: "{{ inputs.rows_per_partition }}"
- id: split_bytes
type: io.kestra.plugin.core.storage.Split
description: Size partitions, cut on record boundaries. Exactly one of `rows`
and `bytes` may be set, hence two tasks and a `runIf`.
runIf: "{{ inputs.split_by == 'bytes' }}"
from: "{{ outputs.drop_tombstones.uri }}"
bytes: "{{ inputs.bytes_per_partition | trim | upper }}"
- id: process_partitions
type: io.kestra.plugin.core.flow.Loop
description: |
One worker execution per partition, three at a time. `transmitFailed: false` keeps the loop going when a
partition is refused, and the Loop `outputs` collect one row per partition, failed ones included, so the
reassembly below knows which files to concatenate and which partitions to send to the rejects file.
values: "{{ outputs.split_rows.uris ?? outputs.split_bytes.uris }}"
concurrencyLimit: 3
transmitFailed: false
tasks:
- id: worker
type: io.kestra.plugin.core.flow.Subflow
description: The partition is passed as a `kestra://` URI into the worker's FILE
input. The labels make the child executions filterable by parent run
and partition.
namespace: "{{ flow.namespace }}"
flowId: "{{ inputs.partition_flow_id }}"
wait: true
transmitFailed: true
inputs:
partition: "{{ item.value }}"
partition_index: "{{ item.index }}"
max_reject_ratio: "{{ inputs.max_reject_ratio }}"
labels:
change_feed_run: "{{ execution.id }}"
partition: "{{ item.index }}"
outputs:
- id: row
type: JSON
value: |
{{ {
'index': item.index,
'partition_uri': item.value,
'state': outputs.worker.state ?? 'FAILED',
'execution_id': outputs.worker.executionId ?? null,
'clean_uri': outputs.worker.outputs.clean_uri ?? null,
'rejects_uri': outputs.worker.outputs.rejects_uri ?? null,
'rows_in': outputs.worker.outputs.rows_in ?? null,
'rows_rejected': outputs.worker.outputs.rows_rejected ?? null
} | toJson }}
- id: tally
type: io.kestra.plugin.core.output.OutputValues
description: |
Loop outputs arrive in completion order, so every list is sorted by partition index first: the clean file keeps
the order of the feed. Failed partitions contribute their whole input to the rejects file. `unique_by(.index)`
is a guard: on Kestra 2.0.5 an iteration whose Subflow task failed before a child execution existed (worker flow
not found) reported its outputs row twice.
values:
partitions_total: "{{ outputs.process_partitions.outputs | jq('[.[].outputs.row]
| unique_by(.index) | length') | first }}"
failed: "{{ outputs.process_partitions.outputs | jq('[.[].outputs.row] |
unique_by(.index) | map(select(.state != \"SUCCESS\"))') | first |
toJson }}"
failed_indexes: "{{ outputs.process_partitions.outputs | jq('[.[].outputs.row] |
unique_by(.index) | map(select(.state != \"SUCCESS\") | .index)') |
first | toJson }}"
clean_files: "{{ outputs.process_partitions.outputs | jq('[.[].outputs.row] |
unique_by(.index) | map(select(.state == \"SUCCESS\") | .clean_uri)') |
first | toJson }}"
reject_files: "{{ outputs.process_partitions.outputs | jq('[.[].outputs.row] |
unique_by(.index) | map(if .state == \"SUCCESS\" then .rejects_uri else
.partition_uri end)') | first | toJson }}"
- id: assemble_clean
type: io.kestra.plugin.core.flow.If
description: Concat needs at least one file; with every partition refused, the
clean file is written empty and the run is reported by
`report_failed_partitions` below.
condition: "{{ (fromJson(outputs.tally.values.clean_files) | length) > 0 }}"
then:
- id: concat_clean
type: io.kestra.plugin.core.storage.Concat
description: Binary ION records concatenate as they are, so the worker outputs
are joined without reading them.
files: "{{ outputs.tally.values.clean_files }}"
extension: .ion
- id: clean_csv
type: io.kestra.plugin.serdes.csv.IonToCsv
from: "{{ outputs.concat_clean.uri }}"
- id: clean_size
type: io.kestra.plugin.core.storage.Size
uri: "{{ outputs.clean_csv.uri }}"
else:
- id: empty_clean_file
type: io.kestra.plugin.core.storage.Write
description: Keeps the `clean_file` output valid when every partition was
refused, so a consumer of the outputs always finds a file.
content: ""
extension: .csv
- id: concat_rejects
type: io.kestra.plugin.core.storage.Concat
description: Row-level rejects of the successful partitions plus every row of
the refused partitions, in feed order. Always written, empty when nothing
was rejected; `rejects_csv.size` is the row count used in the labels and
the summary.
files: "{{ outputs.tally.values.reject_files }}"
extension: .ion
- id: rejects_csv
type: io.kestra.plugin.serdes.csv.IonToCsv
from: "{{ outputs.concat_rejects.uri }}"
- id: rejects_size
type: io.kestra.plugin.core.storage.Size
uri: "{{ outputs.rejects_csv.uri }}"
- id: label_run
type: io.kestra.plugin.core.execution.Labels
description: The Executions page can then be filtered on runs with rejects or
failed partitions without opening them.
labels:
feed_rows: "{{ outputs.csv_to_ion.size ?? outputs.jsonl_to_ion.size }}"
records: "{{ outputs.dedupe_newest_wins.numKeys }}"
rows_clean: "{{ outputs.clean_csv.size ?? 0 }}"
rows_rejected: "{{ outputs.rejects_csv.size }}"
partitions_failed: "{{ fromJson(outputs.tally.values.failed) | length }}"
- id: partition_failures
type: io.kestra.plugin.core.flow.If
description: |
Failed partitions are kept in KV for 7 days (the dead letter: partition file, worker execution id, rejects file),
so an operator can open the worker execution, fix the feed and re-run without digging through the run. A clean
run removes the entry.
condition: "{{ (fromJson(outputs.tally.values.failed) | length) > 0 }}"
then:
- id: remember_failed_partitions
type: io.kestra.plugin.core.kv.Set
key: "{{ vars.dead_letter_key }}"
kvType: JSON
ttl: P7D
value: |
{{ {
'execution_id': execution.id,
'source': inputs.source_file ?? 'built-in sample',
'rejects_csv': outputs.rejects_csv.uri,
'failed_partitions': fromJson(outputs.tally.values.failed)
} | toJson }}
- id: report_failed_partitions
type: io.kestra.plugin.core.execution.Fail
runIf: "{{ inputs.fail_on_partition_failure }}"
errorMessage: |
{{ fromJson(outputs.tally.values.failed) | length }} of {{ outputs.tally.values.partitions_total }} partitions were refused by worker `{{ inputs.partition_flow_id }}` (indexes {{ fromJson(outputs.tally.values.failed_indexes) | join(', ') }}). Their rows are in the rejects file; the clean file holds the other partitions.
{{ (fromJson(outputs.tally.values.failed) | length) == (outputs.tally.values.partitions_total | number) ? 'Every partition failed: check that the worker flow exists in ' ~ flow.namespace ~ ' and open one worker execution id from the KV entry ' ~ vars.dead_letter_key ~ '.' : 'Open the worker execution ids in the KV entry ' ~ vars.dead_letter_key ~ ' for the reject ratio of each refused partition.' }}
else:
- id: clear_dead_letter
type: io.kestra.plugin.core.kv.Delete
key: "{{ vars.dead_letter_key }}"
errorOnMissing: false
errors:
- id: failure_context
type: io.kestra.plugin.core.output.OutputValues
description: "`errorLogs()` is read once, here, before this branch writes its
own log lines. It only works in a worker task, so an `OutputValues`, not
`SetVariables`."
values:
failed_task: "{{ ((errorLogs() | length) > 0 and (errorLogs()[0].taskId ?? '')
!= '') ? errorLogs()[0].taskId : 'none' }}"
failure_message: "{{ (errorLogs() | length) > 0 ? errorLogs()[0].message : 'no
task error was logged, which is how an SLA breach ends the run' }}"
- id: log_failure
type: io.kestra.plugin.core.log.Log
level: ERROR
message: "Change feed run failed in task {{
outputs.failure_context.values.failed_task }}: {{
outputs.failure_context.values.failure_message }} {{ kestra.url ??
inputs.kestra_api_url }}/ui/main/executions/{{ flow.namespace }}/{{
flow.id }}/{{ execution.id }}"
- id: remember_failure
type: io.kestra.plugin.core.kv.Set
description: Where the last failed run and its feed are, for whoever picks it up next.
key: "{{ vars.failure_key }}"
kvType: JSON
ttl: P7D
value: "{{ {'execution_id': execution.id, 'failed_task':
outputs.failure_context.values.failed_task, 'message':
outputs.failure_context.values.failure_message, 'source':
inputs.source_file ?? 'built-in sample', 'format': inputs.source_format} |
toJson }}"
- id: alert_failure
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: allowFailure so a missing `SLACK_WEBHOOK_URL` does not hide the
original error; the same text is in the ERROR log above.
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: ":rotating_light: Change feed pipeline failed in `{{
outputs.failure_context.values.failed_task }}`: {{
outputs.failure_context.values.failure_message }} {{ kestra.url ??
inputs.kestra_api_url }}/ui/main/executions/{{ flow.namespace }}/{{
flow.id }}/{{ execution.id }}"
finally:
- id: summary
type: io.kestra.plugin.core.log.Log
description: One line per run whatever happened, with `n/a` for the steps that
did not run.
message: "Feed: {{ outputs.csv_to_ion.size ?? outputs.jsonl_to_ion.size ?? 'n/a'
}} rows ({{ outputs.feed_size.size ?? 'n/a' }} bytes). Records: {{
outputs.dedupe_newest_wins.numKeys ?? 'n/a' }} ({{
outputs.dedupe_newest_wins.droppedItemsTotal ?? 'n/a' }} older versions
dropped, {{ outputs.drop_tombstones.droppedItemsTotal ?? 'n/a' }}
tombstones). Partitions: {{ outputs.tally.values.partitions_total ?? 'n/a'
}}, failed {{ outputs.tally.values.failed_indexes ?? 'n/a' }}. Clean rows:
{{ outputs.clean_csv.size ?? 'n/a' }} ({{ outputs.clean_size.size ?? 'n/a'
}} bytes). Rejected rows: {{ outputs.rejects_csv.size ?? 'n/a' }} ({{
outputs.rejects_size.size ?? 'n/a' }} bytes)."
afterExecution:
- id: notify_outcome
type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
description: |
Runs once the final state is known. On 2.0.5 an SLA breach fails the run at once and skips `finally`, so this is
the only hook that reports it. Skipped for ordinary failures, which the `errors` branch already reported.
runIf: "{{ execution.state != 'FAILED' or (labels.change_feed_sla ?? '') ==
'breached' }}"
allowFailure: true
url: "{{ secret('SLACK_WEBHOOK_URL') }}"
messageText: "Change feed pipeline ended {{ execution.state }}{{
(labels.change_feed_sla ?? '') == 'breached' ? ' because the 2-hour SLA
was breached' : '' }}. Clean rows: {{ outputs.clean_csv.size ?? 'n/a' }},
rejected: {{ outputs.rejects_csv.size ?? 'n/a' }}, failed partitions: {{
outputs.tally.values.failed_indexes ?? 'n/a' }}. {{ kestra.url ??
inputs.kestra_api_url }}/ui/main/executions/{{ flow.namespace }}/{{
flow.id }}/{{ execution.id }}"
outputs:
- id: clean_file
type: FILE
value: "{{ outputs.clean_csv.uri ?? outputs.empty_clean_file.uri }}"
- id: rejects_file
type: FILE
value: "{{ outputs.rejects_csv.uri }}"
- id: summary
type: JSON
value: |
{{ {
'feed_rows': outputs.csv_to_ion.size ?? outputs.jsonl_to_ion.size,
'records': outputs.dedupe_newest_wins.numKeys,
'older_versions_dropped': outputs.dedupe_newest_wins.droppedItemsTotal,
'tombstones_dropped': outputs.drop_tombstones.droppedItemsTotal,
'partitions_total': outputs.tally.values.partitions_total | number,
'partitions_failed': fromJson(outputs.tally.values.failed) | length,
'rows_clean': outputs.clean_csv.size ?? 0,
'rows_rejected': outputs.rejects_csv.size
} | toJson }}
