Dataflow triggers

Dataflow triggers overview

Dataflow triggers are instructions for the event framework to kick off tasks in response to events that occur in the flow. Or you might use a dataflow trigger to stop a flow after the JDBC Query Consumer source processes all available data.

The event framework consist of the following components:

event generation: The event framework generates flow-related events and stage-related events. The framework generates flow events only when the flow starts and stops. The framework generates stage events when specific stage-related actions take place. The action that generates an event differs from stage to stage and is related to how the stage processes data.; For example, the Amazon S3 target generates events after it completes writing data to an object and after it processes a whole file.; Events produce event records. Flow-related event records are passed immediately to the specified event consumer. Stage-related event records are passed through the flow in an event stream.
task execution: To trigger a task, you need an executor. Executor stages perform tasks in Data Collector or external systems. Each time an executor receives an event, it performs the specified task.; For example, the Pipeline Finisher executor stops a flow upon receiving an event, transitioning the flow to a Finished state.
event storage: To store event information, pass the event to a target. The target writes the event records to the target system, just like any other data.; For example, you might store event records to keep an audit trail of the files that the flow source reads.

Flow event generation

The event framework generates flow events in Data Collector flows at specific points in the flow lifecycle. You can configure the flow properties to pass each event to an executor or to another flow for more complex processing.

The event framework generates the following flow-related events:

Flow Start: The flow start event is generated as the flow initializes, immediately after it starts and before individual stages are initialized. This can allow time for an executor to perform a task before stages initialize.; Most executors wait for confirmation that a task completes. As a result, the flow waits for the executor to complete the task before continuing with stage initialization. For example, if you configure the JDBC Query executor to truncate a table before the flow begins, the flow waits until the task is complete before processing any data.; If the executor fails to process the event, for example if a JDBC Query executor fails to execute the specified query or if the query fails, then the initialization phase fails and the flow does not start. Instead the flow transitions to a failure state.
Flow Stop: The flow stop event is generated as the flow stops, either manually, programmatically, or due to a failure. The stop event is generated after all stages have completed processing and cleaning up temporary resources, such as removing temporary files. This allows an executor to perform a task after flow processing is complete, before the flow fully stops.; Similar to start event consumers, the behavior of the executor that consumes the event determines whether the flow waits for the executor task to complete before allowing the flow to stop. Also, if the processing of the flow stop event fails for any reason, the flow transitions to a failed state even though the data processing was successful.

Flow events differ from stage events as follows:

Virtual processing - Unlike stage events, flow events are not processed by stages that you configure in the canvas. They are passed to an event consumer that you configure in the flow properties.
The event consumer does not display in the flow’s canvas. As a result, flow events are also not visualized in data preview.
Single-use events - You can configure only one event consumer for each event type within the flow properties: one for the Start event and one for the Stop event.
When necessary, you can pass flow events to another flow. In the event consuming flow, you can include as many stages as you need for more complex processing.

Using flow events

You can configure flow events in standalone flows. When you configure a flow event, you can configure it to be consumed by an executor or another flow.

Pass an event to an executor when the executor can perform all of the tasks that you need. You can configure one executor for each event type.

Pass an event to another flow when you need to perform more complex tasks in the consuming flow, such as passing the event to multiple executors or to an executor and target for storage.

Pass to an executor

You can configure a flow to pass each event type to an executor stage. This allows you to trigger a task when the flow starts or stops. You configure the behavior for each event type separately. And you can discard any event that you do not want to use.

Note: If the specified executor fails to process the event, for example if a Shell executor fails to execute a script, the flow transitions to a failure state.

To pass a flow event to the executor, perform the following steps:

In the flow properties, select the executor that you want to consume the event.
In the flow properties, configure the executor to perform the task.

Example

Say you want to send an email when the flow starts. First, you configure the flow to use the Email executor for the flow start event. Since you don't need the Stop event, you can simply use the default discard option:

Flow properties page configured with the Email executor for the Start Event and the default Discard as the Stop Event

Then, also in the flow properties, you configure the Email executor. You can configure a condition for the email to be sent. If you omit the condition, the executor sends the email each time it receives an event:

Flow properties Start Event tab showing email configuration properties

Stage event generation

You can configure certain stages to generate events. Event generation differs from stage to stage, based on the way the stage processes data. For details about each the event generation for each stage, see "Event Generation" in the stage documentation.

The following table lists event-generating stages and when they can generate events:

Stage	Generates events when the stage...
Amazon S3 source	Starts processing an object. Completes processing an object. Completes processing all available objects and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
Azure Blob Storage source	Starts processing an object. Completes processing an object. Completes processing all available objects and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
Azure Data Lake Storage Gen2 source	Starts processing an object. Completes processing an object. Completes processing all available objects and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
Azure Data Lake Storage Gen2 (Legacy) source	Starts processing a file. Completes processing a file. Completes processing all available files and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
Directory source	Starts processing a file. Completes processing a file. Completes processing all available files and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
File Tail source	Starts processing a file. Completes processing a file. For more information, see Event Generation in the source documentation.
Google BigQuery source	Successfully completes a query. For more information, see Event Generation in the source documentation.
Google Cloud Storage source	Completes processing all available objects and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
Groovy Scripting source	Runs a script that generates events. For more information, see Event Generation in the source documentation.
JavaScript Scripting source	Runs a script that generates events. For more information, see Event Generation in the source documentation.
JDBC Multitable Consumer source	Completes processing the data returned by the queries for all tables. For more information, see Event Generation in the source documentation.
JDBC Query Consumer source	Completes processing all data returned by a query. Successfully completes a query. Fails to complete a query. For more information, see Event Generation in the source documentation.
Jython Scripting source	Runs a script that generates events. For more information, see Event Generation in the source documentation.
MongoDB Atlas source	Completes processing all data returned by a query. For more information, see Event Generation in the source documentation.
Oracle Bulkload source	Completes processing data in a table. For more information, see Event Generation in the source documentation.
Oracle CDC source	Flow starts LogMiner session starts Changes occur to monitored tables For more information, see Event Generation in the source documentation.
Oracle CDC Client source	Reads DDL statements in the redo log. For more information, see Event Generation in the source documentation.
Oracle Multitable Consumer source	Completes processing the data returned by the queries for all tables. For more information, see Event Generation in the source documentation.
Oracle XStream source	Receives DDL events Cleans up XStream components For more information, see Event Generation in the source documentation.
Salesforce source	Completes processing all data returned by a query. For more information, see Event Generation in the source documentation.
Salesforce Bulk API 2.0 source	Completes processing all data returned by a query. For more information, see Event Generation in the source documentation.
SAP HANA Query Consumer source	Completes processing all data returned by a query. Successfully completes a query. Fails to complete a query. For more information, see Event Generation in the source documentation.
SFTP/FTP/FTPS Client source	Starts processing a file. Completes processing a file. Completes processing all available files and the configured batch wait time has elapsed. For more information, see Event Generation in the source documentation.
SQL Server CDC Client source	Completes processing the data in the associated CDC tables. When you enable checking for schema changes, the source generates an event each time it detects a schema change. For more information, see Event Generation in the source documentation.
SQL Server Change Tracking source	Completes processing the data in all specified change tracking tables. For more information, see Event Generation in the source documentation.
Web Client source	Finishes processing all available data. For more information, see Event generation in the source documentation.
Windowing Aggregator processor	Performs aggregations, based on the configured window type and time window. For more information, see Event Generation in the processor documentation.
Groovy Evaluator processor	Runs a script that generates events. For more information, see Event Generation in the processor documentation.
JavaScript Evaluator processor	Runs a script that generates events. For more information, see Event Generation in the processor documentation.
Jython Evaluator processor	Runs a script that generates events. For more information, see Event Generation in the processor documentation.
TensorFlow Evaluator processor	Evaluates the entire batch at once. For more information, see Event Generation in the processor documentation.
Web Client processor	Finishes processing all available data. For more information, see Event Generation in the target documentation.
Windowing Aggregator processor	Performs aggregations. For more information, see Event Generation in the processor documentation.
Amazon S3 target	Completes writing to an object. Completes streaming a whole file. For more information, see Event Generation in the target documentation.
Azure Blob Storage target	Completes writing to a blob. For more information, see Event Generation in the target documentation.
Azure Data Lake Storage Gen2 target	Closes a file. Completes streaming a whole file. For more information, see Event Generation in the target documentation.
Google Cloud Storage target	Completes writing to an object. Completes streaming a whole file. For more information, see Event Generation in the target documentation.
Local FS target	Closes a file. Completes streaming a whole file. For more information, see Event Generation in the target documentation.
SFTP/FTP/FTPS Client target	Closes a file. Completes streaming a whole file. For more information, see Event Generation in the target documentation.
Snowflake File Uploader target	Completes streaming a whole file. For more information, see Event Generation in the target documentation.
Web Client target	Finishes processing all available data. For more information, see Event Generation in the target documentation.
ADLS Gen2 File Metadata executor	Changes file metadata, such as the file name, location, or permissions. Creates an empty file. Removes a file or directory. For more information, see Event Generation in the executor documentation.
Amazon S3 executor	Creates a new Amazon S3 object. Copies an object another location. Adds tags to an existing object. For more information, see Event Generation in the executor documentation.
Databricks Job Launcher executor	Starts a Databricks job. For more information, see Event Generation in the executor documentation.
Databricks Query executor	Determines that the submitted query completed successfully. Determines that the submitted query failed to complete. For more information, see Event Generation in the executor documentation.
Google BigQuery executor	Determines that the submitted query completed successfully. Determines that the submitted query failed to complete. For more information, see Event Generation in the executor documentation.
Google Cloud Storage executor	Creates a new Google Cloud Storage object. Copies an object to another location. Moves an object to another location. Adds metadata to an existing object. For more information, see Event Generation in the executor documentation.
JDBC Query executor	Determines that the submitted query completed successfully. Determines that the submitted query failed to complete. For more information, see Event Generation in the executor documentation.
Snowflake executor	Determines that the submitted query completed successfully. Determines that the submitted query failed to complete. For more information, see Event Generation in the executor documentation.

Using stage events

You can use stage-related events in any way that suits your needs. When configuring the event stream for stage events, you can add additional stages to the stream. For example, you might use a Stream Selector to route different types of events to different executors. But you cannot merge the event stream with a data stream.

There are two general types of event streams that you might create:

Task execution streams that route events to an executor to perform a task.
Event storage streams that route events to a target to store event information.

You can, of course, configure an event stream that performs both tasks by routing event records to both an executor and a target. You can also configure event streams to route data to multiple executors and targets, as needed.

Task execution streams

A task execution stream routes event records from the event-generating stage to an executor stage. The executor performs a task each time it receives an event record.

For example, you have a flow that reads from Kafka and writes files to HDFS:

Flow canvas showing a flow with a Kafka Consumer source and Hadoop FS target

When Hadoop FS closes a file, you would like the file moved to a different directory and the file permissions changed to read-only.

Leaving the rest of the flow as is, you can enable event handling in the Hadoop FS target, connect it to the HDFS File Metadata executor, and configure the HDFS File Metadata executor to files and change permissions. The resulting flow looks like this:

Flow canvas showing a flow with a Hadoop FS target connected to a HDFS File Metadata executor

If you needed to set permissions differently based on the file name or location, you could use a Stream Selector to route the event records accordingly, then use two HDFS File Metadata executors to alter file permissions, as follows:

Flow canvas showing a flow with a Hadoop FS target connected to tow HDFS File Metadata executors

Event storage streams

An event storage stream routes event records from the event-generating stage to a target. The target writes the event record to a target system.

Event records include information about the event in record header attributes and record fields. You can add processors to the event stream to enrich the event record before writing it to the target.

For example, you have a flow that uses the Directory source to process weblogs:

Flow canvas showing a flow with a Directory source

Directory generates event records each time it starts and completes reading a file, and the event record includes a field with the file path of the file. For auditing purposes, you'd like to write this information to a database table.

Leaving the rest of the flow as is, you can enable event handling for the Directory source and simply connect it to the JDBC Producer as follows:

Flow canvas showing a flow with a Directory source connected to a JDBC Producer destination to store events

But you want to know when events occur. The Directory event record stores the event creation time in the sdc.event.creation_timestamp record header attribute. So you can use an Expression Evaluator with the following expression to add the creation date and time to the record:

${record:attribute('sdc.event.creation_timestamp')}

And if you have multiple flows writing events to the same location, you can use the following expression to include the flow name in the event record as well:

${flow:name()}

The Expression Evaluator and the final flow looks like this:

Flow canvas showing a flow with a Directory source connected to an Expression Evaluator configured to add the creation time and date to a record

Executors

Executors perform tasks when they receive event records.

You can use the following executor stages for event handling:

ADLS Gen2 File Metadata executor: Changes file metadata, creates an empty file, or removes a file or directory in Azure Data Lake Storage Gen2 upon receiving an event.; When changing file metadata, the executor can rename and move files in addition to specifying the owner and group, and updating permissions and ACLs for files. When creating an empty file, the executor can specify the owner and group, and set permissions and ACLs for the file. When removing files and directories, the executor performs the task recursively.; You can use the executor in any logical way, such as moving a file after an Azure Data Lake Storage Gen2 target closes it.
Amazon S3 executor: Creates new Amazon S3 objects for the specified content, copies objects within a bucket, or adds tags to existing Amazon S3 objects upon receiving an event.; You can use the executor in any logical way, such as writing information from an event record to a new S3 object, or copying or tagging objects after they are written by the Amazon S3 target.
Databricks Job Launcher executor: Starts a Databricks job for each event.; You can use the executor in any logical way, such as running Databricks jobs after the Amazon S3 target closes files.
Databricks Query executor: Runs a Spark SQL query on Databricks upon receiving an event.; You can use the executor in any logical way.
Email executor: Sends a custom email to the configured recipients upon receiving an event. You can optionally configure a condition that determines when to send the email.; You can use the executor in any logical way, such as sending an email each time the Azure Data Lake Storage target completes streaming a whole file.; For a solution that describes how to use the Email executor, see Sending email during flow processing.
Google BigQuery executor: Runs one or more SQL queries on Google BigQuery upon receiving an event.; You can use the executor in any logical way, such as running a SQL query each time a Google BigQuery source successfully completes a query.
Google Cloud Storage executor: Creates new Google Cloud Storage objects for the specified content, copies or moves objects within a project, or adds metadata to existing objects upon receiving an event.; You can use the executor in any logical way, such as moving objects after they are read by the Google Cloud Storage source or adding metadata to objects after they are written by the Google Cloud Storage target.
Pipeline Finisher executor: Stops the flow when it receives an event, transitioning the flow to a Finished state. Allows the flow to complete all expected processing before stopping.; You can use the Pipeline Finisher executor in any logical way, such as stopping a flow upon receiving a no-more-data event from the JDBC Query Consumer source. This enables you to achieve "batch" processing - stopping the flow when all available data is processed rather than leaving the flow to sit idle indefinitely.; For example, you might use the Pipeline Finisher executor with the JDBC Multitable Consumer to stop the flow when it processes all queried data in the specified tables.; For a solution that describes how to use the Pipeline Finisher executor, see Stopping a flow after processing all available data.
JDBC Query executor: Connects to a database using JDBC and runs one or more specified SQL queries.; Use to run SQL queries on a database after an event occurs.
SFTP/FTP/FTPS Client executor: Connects to an SFTP, FTP, or FTPS server and moves or removes a file upon receiving an event.; You can use the executor in any logical way, such as moving a file after the SFTP/FTP/FTPS Client source finishes reading it.
Shell executor: Executes a user-defined shell script for each event.; You can use the executor in any logical way.
Snowflake executor: Loads whole files staged by the Snowflake File Uploader target to Snowflake tables.; For details on how to use the executor, see Snowflake File Uploader and executor flows.

Logical pairings

You can use events in any way that works for your needs. The following tables outline some logical pairings of event generation with executors and targets.

Flow events

Flow Event Type	Event Consumer
Flow Start	Any single executor, except Pipeline Finisher. Another flow for additional processing.
Flow Stop	Any single executor, except Pipeline Finisher. Another flow for additional processing.

Source events

Event Generating Source	Event Consumer
Amazon S3	Pipeline Finisher executor to stop the flow after processing queried data from all objects. Email executor to send email when the source completes processing available objects. Any target for event storage.
Azure Blob Storage	Pipeline Finisher executor to stop the flow after processing queried data from all objects. Email executor to send email when the source completes processing available objects. Any target for event storage.
Azure Data Lake Storage Gen2	Pipeline Finisher executor to stop the flow after processing queried data from all objects. Email executor to send email after the source completes processing available objects. Any target for event storage.
Azure Data Lake Storage Gen2 (Legacy)	Pipeline Finisher executor to stop the flow after processing all available files. Email executor to send email after closing a file or streaming a whole file. Any target for event storage.
Directory	Email executor to send email each time the source starts or completes processing a file. Pipeline Finisher executor to stop the flow after processing all available files. Any target for event storage.
File Tail	Email executor to send email each time the source starts or completes processing a file. Any target for event storage.
Google BigQuery	Email executor to send email each time the source successfully completes a query. Google BigQuery executor to run SQL queries each time the source successfully completes a query. Any target for event storage.
Google Cloud Storage	Google Cloud Storage executor to perform a task after reading an object. Pipeline Finisher executor to stop the flow after processing queried data from all objects. Email executor to send email when the source completes processing available objects. Any target for event storage.
JDBC Multitable Consumer	Pipeline Finisher executor to stop the flow after processing queried data from all tables. Email executor to send email when the source completes processing all data returned by queries. Any target for event storage.
JDBC Query Consumer	Route the no-more-data event to the Pipeline Finisher executor to stop the flow after processing queried data. Email executor to send email each time the source successfully completes a query, fails to complete a query, or completes processing all available data. Any target for event storage.
MongoDB Atlas	Route the no-more-data event to the Pipeline Finisher executor to stop the flow after processing queried data. Email executor to send email when the source completes processing available data. Any target for event storage.
Oracle Bulkload	Email executor to send email after it completes reading data in a table. Any target for event storage.
Oracle CDC	Email executor to send email for each event. Any target for event storage.
Oracle CDC Client	Email executor to send email each time it reads DDL statements in the redo logs. Any target for event storage.
Oracle Multitable Consumer	Pipeline Finisher executor to stop the flow after processing queried data from all tables. Email executor to send email when the source completes processing all data returned by queries. Any target for event storage.
Oracle XStream	Email executor to send email for each event. Any target for event storage.
Salesforce	Pipeline Finisher executor to stop the flow after processing queried data. Email executor to send email when the source completes processing all data returned by a query. Any target for event storage.
Salesforce Bulk API 2.0	Pipeline Finisher executor to stop the flow after processing queried data. Email executor to send email when the source completes processing all data returned by a query. Any target for event storage.
SAP HANA Query Consumer	Route the no-more-data event to the Pipeline Finisher executor to stop the flow after processing queried data. Email executor to send email each time the source successfully completes a query, fails to complete a query, or completes processing all available data. Any target for event storage.
SFTP/FTP/FTPS Client	Email executor to send email each time the source starts or completes processing a file. Pipeline Finisher executor to stop the flow after processing all available files. SFTP/FTP/FTPS Client executor to move or remove a processed file. Any target for event storage.
SQL Server Change Tracking	Pipeline Finisher executor to stop the flow after processing available data.
Web Client	Pipeline Finisher executor to stop the flow after processing all available data. Email executor to send email when the source completes processing all data returned by queries. Any target for event storage.

Processor events

Event Generating Processor	Event Consumer
Groovy Evaluator	Any logical executor. Any target for event storage.
JavaScript Evaluator	Any logical executor. Any target for event storage.
Jython Evaluator	Any logical executor. Any target for event storage.
TensorFlow Evaluator	Any target for event storage.
Web Client	Email executor to send email after processing all available data. Any target for event storage.
Windowing Aggregator	Email executor for notification when results pass specified thresholds. Any target for event storage.

Target events

Event Generating Target	Event Consumer
Amazon S3	Amazon S3 executor to create or copy objects or to add tags to closed objects. Databricks Job Launcher executor to run a Databricks application after closing an object or whole file. Email executor to send email after closing an object or whole file. Any target for event storage.
Azure Blob Storage	Email executor to send email after closing a file or streaming a whole file. Any target for event storage.
Azure Data Lake Storage Gen2	ADLS Gen2 executor to change file metadata, create an empty file, or remove a file or directory after closing a file. Email executor to send email after closing a file or streaming a whole file. Any target for event storage.
Google Cloud Storage	Databricks Job Launcher executor to run a Databricks job after closing an object or whole file. Google Cloud Storage executor to perform a task after writing an object or whole file. Email executor to send email after closing an object or whole file. Any target for event storage.
Local FS	Email executor to send email after the target closes a file or streams a whole file. Any target for event storage.
SFTP/FTP/FTPS Client	Email executor to send email each time the target closes a file or streams a whole file. SFTP/FTP/FTPS Client executor to move a closed file. Any target for event storage.
Snowflake File Uploader	Snowflake executor to run queries to load data to Snowflake tables each time the target streams a whole file. Email executor to send email each time the target streams a whole file. Any target for event storage.

Executor events

Event Generating Executor	Event Consumer
ADLS Gen2 File Metadata executor	Email executor to send email each time the executor changes file metadata. Any target for event storage.
Amazon S3	Email executor to send email each time the executor changes object metadata. Any target for event storage.
Databricks Job Launcher executor	Email executor to send email each time the Databricks Job Launcher executor starts a Databricks job. Any target for event storage.
Databricks Query executor	Email executor to send email when a query succeeds or fails. Any target for event storage.
Google BigQuery executor	Email executor to send email when a query succeeds or fails. Any target for event storage.
Google Cloud Storage executor	Email executor to send email when a query succeeds or fails. Any target for event storage.
JDBC Query executor	Email executor to send email when a query succeeds or fails. Any target for event storage.
Snowflake executor	Email executor to send email when a query succeeds or fails. Any target for event storage.

Event records

Event records are records that are created when a stage or flow event occurs.

Most event records pass general event information in record headers, such as when the event occurred. They also can include event-specific details in record fields, like the name and location of the output file that was closed.

Event records generated by the File Tail source are the exception - they include all event information in record fields.

Event record header attributes

In addition to the standard record header attributes, most event records include record header attributes for event information such as the event type and when the event occurred.

As with any record header attribute, you can use the Expression Evaluator and the record:attribute function to include record header attribute information as a field in the record. For example, when storing event records, you most likely want to include the time of the event in the event record, using the following expression in an Expression Evaluator:

${record:attribute('sdc.event.creation_timestamp')}

Note that all record header attributes are String values. For more information about working with record header attributes, see Record header attributes.

Most events include the following event record header attributes. The exception, File Tail, writes all of the event information to record fields.

Event Record Header Attribute	Description
sdc.event.type	Event type. Defined by the stage that generates the event. For information about the event types available for an event generating stage, see the stage documentation.
sdc.event.version	Integer that indicates the version of the event record type.
sdc.event.creation_timestamp	Epoch timestamp when the stage created the event.

Note: Stage-generated event records differ from stage to stage. For a description of stage events, see "Event Record" in the documentation for the event-generating stage. For a description of flow events, see Flow event records.

Summary

Here are the key points about dataflow triggers and the event framework:

You can use the event framework in any flow where the logic suits your needs.
The event framework generates flow-related events and stage-related events.
You can use flow events in standalone flows.
Flow events are generated when the flow starts and stops. For details, see Flow event generation.
You can configure each flow event type to pass to a single executor or to another flow for more complex processing.
Stage events are generated based on the processing logic of the stage. For a list of event-generating stages, see Stage event generation.
Events generate event records to pass relevant information regarding the event, such as the path to the file that was closed.
Stage-generated event records differ from stage to stage. For a description of stage events, see "Event Record" in the documentation for the event-generating stage. For a description of flow events, see Flow event records.
In the simplest use case, you can route stage event records to a target to save event information.
You can route stage event records to an executor stage so it can perform a task upon receiving an event.
For a list of logical event generation and executor pairings, see Logical pairings.
You can add processors to event streams for stage events or to consuming flows for flow events.
For example, you might add an Expression Evaluator to add the event generation time to an event record before writing it to a target. Or, you might use a Stream Selector to route different types of event records to different executors.
When working with stage events, you cannot merge event streams with data streams.
You can use the Dev Data Generator and To Event development stages to generate events for flow development and testing. For more information about the development stages, see Development stages.

For examples of how you might use the event framework, see the case studies earlier in this chapter.