Azure Produce

Azure Produce

Certified

Publish events to Azure Event Hubs

Reads records from a file or list, serializes bodies, batches by size/count, and sends to a single Event Hub. Defaults: bodySerializer=STRING, maxEventsPerBatch=1000. Optionally set partitionKey or maxBatchSizeInBytes.

yaml
type: io.kestra.plugin.azure.eventhubs.Produce

Publish a file as events into Azure EventHubs.

yaml
id: azure_eventhubs_send_events
namespace: company.team

inputs:
  - id: file
    type: FILE
    description: a CSV file with columns id, username, tweet, and timestamp

tasks:
  - id: read_csv_file
    type: io.kestra.plugin.serdes.csv.CsvToIon
    from: "{{ inputs.file }}"

  - id: transform_row_to_json
    type: io.kestra.plugin.graalvm.js.FileTransform
    from: "{{ outputs.read_csv_file.uri }}"
    script: |
      var result = {
        "body": {
          "username": row.username,
          "tweet": row.tweet
        }
      };
      row = result

  - id: send_to_eventhub
    type: io.kestra.plugin.azure.eventhubs.Produce
    from: "{{ outputs.transform_row_to_json.uri }}"
    eventHubName: my_eventhub
    namespace: my_event_hub_namespace
    connectionString: "{{ secret('EVENTHUBS_CONNECTION') }}"
    maxBatchSizeInBytes: 4096
    maxEventsPerBatch: 100
    bodySerializer: "JSON"
    bodyContentType: application/json
    eventProperties:
      source: kestra
Properties

Event Hub name

Target Event Hub entity

Structured data items, either as a map, a list of map, a URI, or a JSON string.

Structured data items can be defined in the following ways:

  • A single item as a map (a document).
  • A list of items as a list of maps (a list of documents).
  • A URI, supported schemes are kestra for internal storage files, file for host local files, and nsfile for namespace files.
  • A JSON String that will then be serialized either as a single item or a list of items.

Namespace

Event Hubs namespace when using AAD or shared key auth

Body content type

MIME type placed on each event for downstream consumers

DefaultSTRING
Possible Values
STRINGBINARYIONJSON

Body serializer

Serde used to serialize event bodies; defaults to STRING

Default{}

Serializer properties

Key/value options passed to the serializer

Default5

Client max retries

Max retry attempts for Event Hubs client operations; default 5

Default500

Retry delay (ms)

Delay between client retries in milliseconds; default 500

Event Hubs connection string

Namespace or Event Hub–level connection string; overrides key/sas fields when set

Custom endpoint address

Custom endpoint for Event Hubs (e.g., for private link); optional

Default{}

Event properties

Metadata properties applied to each event body

Max batch size (bytes)

Maximum batch payload size; optional

Default1000

Max events per batch

Cap on events per batch; defaults to 1000

Partition key

Routes all events in this run to the same partition using hash

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

SAS token

Precomputed SAS token for Event Hubs; optional alternative to shared key/connection string

Shared key

Access key paired with sharedKeyAccountName; ignored if connectionString is provided

Shared key account name

Event Hubs namespace name used with sharedKeyAccountAccessKey when no connection string is provided

Events sent

Batches sent

The total number of batches sent.

The total number of events sent.

More Plugins in this Category

Liquibase
Liquibase

Tasks that run Liquibase change management commands.

Data

2 tasks

OPA (Open Policy Agent)
OPA (Open Policy Agent)
Enterprise

Tasks to manage Open Policy Agent policies and decisions, including upload, evaluation, and partial evaluation.

Infrastructure

5 tasks

3 blueprints