Out of the box Spring Cloud Data Flow will use the H2 embedded database for storing state, Kafka for communication and no analytics. Customizations can be made to these components by editing the docker-compose.yml file as described below.

Spring Cloud Data Flow’s architectural style is different than other Stream and Batch processing platforms. For example in Apache Spark, Apache Flink, and Google Cloud Dataflow, applications run on a dedicated compute engine cluster. The nature of the compute engine gives these platforms a richer environment for performing complex calculations on the data as compared to Spring Cloud Data Flow, but it introduces the complexity of another execution environment that is often not needed when creating data-centric applications. That does not mean you cannot do real-time data computations when using Spring Cloud Data Flow. Refer to the section Analytics, which describes the integration of Redis to handle common counting-based use cases. Spring Cloud Stream also supports using Reactive APIs such as Project Reactor and RxJava which can be useful for creating functional style applications that contain time-sliding-window and moving-average functionality. Similarly, Spring Cloud Stream also supports the development of applications in that use the Kafka Streams API.

Apache Storm, Hortonworks DataFlow, and Spring Cloud Data Flow’s predecessor, Spring XD, use a dedicated application execution cluster, unique to each product, that determines where your code should run on the cluster and performs health checks to ensure that long-lived applications are restarted if they fail. Often, framework-specific interfaces are required in order to correctly “plug in” to the cluster’s execution framework.

As we discovered during the evolution of Spring XD, the rise of multiple container frameworks in 2015 made creating our own runtime a duplication of effort. There is no reason to build your own resource management mechanics when there are multiple runtime platforms that offer this functionality already. Taking these considerations into account is what made us shift to the current architecture, where we delegate the execution to popular runtimes, which you may already be using for other purposes. This is an advantage in that it reduces the cognitive distance for creating and managing data-centric applications as many of the same skills used for deploying other end-user/web applications are applicable.

The Data Flow Server uses an embedded servlet container and exposes REST endpoints for creating, deploying, undeploying, and destroying streams and tasks, querying runtime state, analytics, and the like. The Data Flow Server is implemented by using Spring’s MVC framework and the Spring HATEOAS library to create REST representations that follow the HATEOAS principle, as shown in the following image:

[NOTE] The Data Flow Server that deploys applications to the local machine is not intended to be used in production for streaming use cases but for the development and testing of stream based applications. The local Data Flow is intended to be used in production for batch use cases as a replacement for the Spring Batch Admin project. Both streaming and batch use cases are intended to be used in production when deploying to Cloud Foundry or Kuberenetes.

The Data Flow Server executable jars support basic HTTP, LDAP(S), File-based, and OAuth 2.0 authentication to access its endpoints. Refer to the security section for more information.

The Stream DSL describes linear sequences of data flowing through the system. For example, in the stream definition http | transformer | cassandra, each pipe symbol connects the application on the left to the one on the right. Named channels can be used for routing and to fan in/fan out data to multiple messaging destinations.

The concept of a tap can be used to ‘listen’ to the data that is flowing across any of the pipe symbols. "Taps" are just other streams that use an input any one of the "pipes" in a target stream and have an independent life cycle from the target stream.

For an application that consumes events, Spring Cloud Stream exposes a concurrency setting that controls the size of a thread pool used for dispatching incoming messages. See the Consumer properties documentation for more information.

A common pattern in stream processing is to partition the data as it moves from one application to the next. Partitioning is a critical concept in stateful processing, for either performance or consistency reasons, to ensure that all related data is processed together. For example, in a time-windowed average calculation example, it is important that all measurements from any given sensor are processed by the same application instance. Alternatively, you may want to cache some data related to the incoming events so that it can be enriched without making a remote procedure call to retrieve the related data.

Spring Cloud Data Flow supports partitioning by configuring Spring Cloud Stream’s output and input bindings. Spring Cloud Stream provides a common abstraction for implementing partitioned processing use cases in a uniform fashion across different types of middleware. Partitioning can thus be used whether the broker itself is naturally partitioned (for example, Kafka topics) or not (RabbitMQ). The following image shows how data could be partitioned into two buckets, such that each instance of the average processor application consumes a unique set of data.

To use a simple partitioning strategy in Spring Cloud Data Flow, you need only set the instance count for each application in the stream and a partitionKeyExpression producer property when deploying the stream. The partitionKeyExpression identifies what part of the message is used as the key to partition data in the underlying middleware. An ingest stream can be defined as http | averageprocessor | cassandra. (Note that the Cassandra sink is not shown in the diagram above.) Suppose the payload being sent to the HTTP source was in JSON format and had a field called sensorId. For example, consider the case of deploying the stream with the shell command stream deploy ingest --propertiesFile ingestStream.properties where the contents of the ingestStream.properties file are as follows:

The result is to deploy the stream such that all the input and output destinations are configured for data to flow through the applications but also ensure that a unique set of data is always delivered to each averageprocessor instance. In this case, the default algorithm is to evaluate payload.sensorId % partitionCount where the partitionCount is the application count in the case of RabbitMQ and the partition count of the topic in the case of Kafka.

Please refer to Passing Stream Partition Properties for additional strategies to partition streams during deployment and how they map onto the underlying Spring Cloud Stream Partitioning properties.

Also note that you cannot currently scale partitioned streams. Read Scaling at Runtime for more information.

Streams are composed of applications that use the Spring Cloud Stream library as the basis for communicating with the underlying messaging middleware product. Spring Cloud Stream also provides an opinionated configuration of middleware from several vendors, in particular providing persistent publish-subscribe semantics.

The Binder abstraction in Spring Cloud Stream is what connects the application to the middleware. There are several configuration properties of the binder that are portable across all binder implementations and some that are specific to the middleware.

For consumer applications, there is a retry policy for exceptions generated during message handling. The retry policy is configured by using the common consumer properties maxAttempts, backOffInitialInterval, backOffMaxInterval, and backOffMultiplier. The default values of these properties retry the callback method invocation 3 times and wait one second for the first retry. A backoff multiplier of 2 is used for the second and third attempts.

When the number of retry attempts has exceeded the maxAttempts value, the exception and the failed message become the payload of a message and are sent to the application’s error channel. By default, the default message handler for this error channel logs the message. You can change the default behavior in your application by creating your own message handler that subscribes to the error channel.

Spring Cloud Stream also supports a configuration option for both Kafka and RabbitMQ binder implementations that sends the failed message and stack trace to a dead letter queue. The dead letter queue is a destination and its nature depends on the messaging middleware (for example, in the case of Kafka, it is a dedicated topic). To enable this for RabbitMQ set the republishtoDlq and autoBindDlq consumer properties and the autoBindDlq producer property to true when deploying the stream. To always apply these producer and consumer properties when deploying streams, configure them as common application properties when starting the Data Flow server.

Additional messaging delivery guarantees are those provided by the underlying messaging middleware that is chosen for the application for both producing and consuming applications. Refer to the Kafka Consumer and Producer and Rabbit Consumer and Producer documentation for more details. You can find extensive declarative support for all the native QOS options.

Spring Cloud Stream is most closely integrated with Spring Integration’s imperative "one event at a time" programming model. This means you write code that handles a single event callback, as shown in the following example,

In this case, the String payload of a message coming on the input channel is handed to the log method. The @EnableBinding annotation is used to tie the input channel to the external middleware.

However, Spring Cloud Stream can support other programming styles, such as reactive APIs, where incoming and outgoing data is handled as continuous data flows and how each individual message should be handled is defined. With many reactive AOIs, you can also use operators that describe functional transformations from inbound to outbound data flows. Here is an example:

The target runtimes supported by Data Flow all have the ability to restart a long-lived application. Spring Cloud Data Flow sets up health probes are required by the runtime environment when deploying the application. You also have the ability to customize the health probes.

The collective state of all applications that make up the stream is used to determine the state of the stream. If an application fails, the state of the stream changes from ‘deployed’ to ‘partial’.

Each target runtime lets you control the amount of memory, disk, and CPU allocated to each application. These are passed as properties in the deployment manifest by using key names that are unique to each runtime. Refer to each platform’s server documentation for more information.

When deploying a stream, you can set the instance count for each individual application that makes up the stream. Once the stream is deployed, each target runtime lets you control the target number of instances for each individual application. Using the APIs, UIs, or command line tools for each runtime, you can scale up or down the number of instances as required.

Currently, scaling at runtime is not supported with the Kafka binder, as well as with partitioned streams, for which the suggested workaround is redeploying the stream with an updated number of instances. Both cases require a static consumer to be set up, based on information about the total instance count and current instance index.

You can control whether or not Data Flow bootstraps your database on startup. On most production environments chances are you will not have enough privileges to do so. If that’s the case you can disable it by setting the property spring.cloud.dataflow.rdbms.initialize.enable to false. The scripts that the server uses to bootstrap the database can be found under spring-cloud-dataflow-core/src/main/resources/ folder.

For new installations run the corresponding database script located under /schemas and /migrations.1.x.x, for upgrades from version 1.2.0 you only need to run the /migrations.1.x.x scripts.

To add a custom driver for the database (for example, Oracle), you should rebuild the Data Flow Server and add the dependency to the Maven pom.xml file. Since there is a Spring Cloud Data Flow Server for each target platform, you need to modify the appropriate maven pom.xml for each platform. There are tags in each GitHub repository for each server version.

To add a custom JDBC driver dependency for the local server implementation:

You can also provide default values when rebuilding the server by adding the necessary properties to the dataflow-server.yml file, as shown in the following example for PostgreSQL:

By default, the dashboard, management, and health endpoints use HTTP as a transport. You can switch to HTTPS by adding a certificate to your configuration in application.yml, as shown in the following example:

When using traditional authentication, Spring Cloud Data Flow is the sole authentication provider. In that case, Data Flow REST API users would use Basic Authentication to access the endpoints.

When using that option, users have a choice of three backing stores for authentication details:

OAuth 2.0 lets you integrate Spring Cloud Data Flow into Single Sign On (SSO) environments. The following OAuth2 Grant Types are used:

The REST endpoints can be accessed in two ways:

You can turn on OAuth2 authentication by adding the following to application.yml or by setting environment variables:

You can verify that basic authentication is working properly by using curl, as follows:

As a result, you should see a list of available REST endpoints.

Besides Basic Authentication, you can also provide an Access Token in order to access the REST Api. In order to make that happen, you would retrieve an OAuth2 Access Token from your OAuth2 provider first and then pass that Access Token to the REST Api using the Authorization Http header:

When enabling security, please also make sure that the Spring Boot HTTP Management Endpoints are secured as well. You can enable security for the management endpoints by adding the following to application.yml:

The stream applications deployed by Spring Cloud Data Flow can be based on Spring Boot 1.5 or Spring Boot 2.0. Both versions contains several features for monitoring your application in production. However, Spring Boot 1.x and 2.x as well as Spring Cloud Stream 1.x and 2.x differ in how monitoring is implemented. Since Spring Cloud Data Flow supports deploying 1.x and 2.x applications, we will cover both cases individually.

What is common across 1.x and 2.x applications is that Spring Cloud Stream apps can be configured to publish metrics to a messaging middleware destination. The Spring Cloud Data Flow Metrics Collector subscribes to this destination and aggregates metrics into a stream based view. The Metrics Collector 2.0 server supports collecting metrics from streams that contain only Boot 1.x or 2.x apps as well as streams that contain a mixture of Boot versions. The Data Flow UI queries the Metrics collector over HTTP to display messages rates next to each deployed application.

The following image shows the architecture when using Spring Cloud Stream’s metrics publisher, the Metrics Collector, and the Data Flow server:

By default, checksum values are not displayed for the shell dependency. If you need this feature enabled, set the spring.cloud.dataflow.version-info.dependency-fetch.enabled property to true.

There are reserved values (surrounded by curly braces) that you can insert into the URL that will make sure that the links are up to date:

For example, myrepository/org/springframework/cloud/spring-cloud-dataflow-shell/{version}/spring-cloud-dataflow-shell-{version}.jar produces myrepository/org/springframework/cloud/spring-cloud-dataflow-shell/1.2.3.RELEASE/spring-cloud-dataflow-shell-1.2.3.RELEASE.jar if you were using the 1.2.3.RELEASE version of the Spring Cloud Data Flow Shell

There is a Spring Shell-based client that talks to the Data Flow Server and is responsible for parsing the DSL. In turn, applications may have applications properties that rely on embedded languages, such as the Spring Expression Language.

The shell, Data Flow DSL parser, and SpEL have rules about how they handle quotes and how syntax escaping works. When combined together, confusion may arise. This section explains the rules that apply and provides examples of the most complicated situations you may encounter when all three components are involved.

A stream is defined by using a unix-inspired Pipeline syntax. The syntax uses vertical bars, also known as “pipes” to connect multiple commands. The command ls -l | grep key | less in Unix takes the output of the ls -l process and pipes it to the input of the grep key process. The output of grep in turn is sent to the input of the less process. Each | symbol connects the standard output of the command on the left to the standard input of the command on the right. Data flows through the pipeline from left to right.

In Data Flow, the Unix command is replaced by a Spring Cloud Stream application and each pipe symbol represents connecting the input and output of applications over messaging middleware, such as RabbitMQ or Apache Kafka.

Each Spring Cloud Stream application is registered under a simple name. The registration process specifies where the application can be obtained (for example, in a Maven Repository or a Docker registry). You can find out more information on how to register Spring Cloud Stream applications in this section. In Data Flow, we classify the Spring Cloud Stream applications as Sources, Processors, or Sinks.

As a simple example, consider the collection of data from an HTTP Source writing to a File Sink. Using the DSL, the stream description is:

http | file

A stream that involves some processing would be expressed as:

http | filter | transform | file

Stream definitions can be created by using the shell’s stream create command, as shown in the following example:

dataflow:> stream create --name httpIngest --definition "http | file"

The Stream DSL is passed in to the --definition command option.

The deployment of stream definitions is done through the shell’s stream deploy command.

dataflow:> stream deploy --name ticktock

The Getting Started section shows you how to start the server and how to start and use the Spring Cloud Data Flow shell.

Note that the shell calls the Data Flow Servers' REST API. For more information on making HTTP requests directly to the server, consult the REST API Guide.

The Stream Pipeline DSL described in the previous section automatically sets the input and output binding properties of each Spring Cloud Stream application. This can be done because there is only one input and/or output destination in a Spring Cloud Stream application that uses the provided binding interface of a Source, Processor, or Sink. However, a Spring Cloud Stream application can define a custom binding interface such as the one shown below

or as is common when creating a Kafka Streams application,

In these cases with multiple input and output bindings, Data Flow cannot make any assumptions about the flow of data from one application to another. Therefore the developer needs to set the binding properties to 'wire up' the application. The Stream Application DSL uses a comma, instead of the pipe symbol, to indicate that Data Flow should not configure the binding properties of the application. For example:

dataflow:> stream create --definition "orderGeneratorApp, baristaApp, hotDrinkDeliveryApp, coldDrinkDeliveryApp" --name myCafeStream

There are four applications in this stream. The baristaApp has two output destinations, hotDrinks and coldDrinks intended to be consumed by the hotDrinkDeliveryApp and coldDrinkDeliveryApp respectively. When deploying this stream, you need to set the binding properties so that the baristaApp sends hot drink messages to the hotDrinkDeliveryApp destination and cold drink messages to the coldDrinkDeliveryApp destination. For example

If you want to use consumer groups, you will need to set the Spring Cloud Stream application property spring.cloud.stream.bindings.<channelName>.producer.requiredGroups and spring.cloud.stream.bindings.<channelName>.group on the producer and consumer applications respectively.

Another common use case for the Stream Application DSL is to deploy a http gateway application that sends a synchronous request/reply message to a Kafka or RabbitMQ application. In this case both the http gateway application and the Kafka or RabbitMQ application can be a Spring Integration application that does not make use of the Spring Cloud Stream library.

It is also possible to deploy just a single application using the Stream application DSL.

Each application takes properties to customize its behavior. As an example, the http source module exposes a port setting that allows the data ingestion port to be changed from the default value.

dataflow:> stream create --definition "http --port=8090 | log" --name myhttpstream

This port property is actually the same as the standard Spring Boot server.port property. Data Flow adds the ability to use the shorthand form port instead of server.port. One may also specify the longhand version as well, as shown in the following example:

dataflow:> stream create --definition "http --server.port=8000 | log" --name myhttpstream

This shorthand behavior is discussed more in the section on Whitelisting application properties. If you have registered application property metadata you can use tab completion in the shell after typing -- to get a list of candidate property names.

The shell provides tab completion for application properties. The shell command app info --name <appName> --type <appType> provides additional documentation for all the supported properties.

You can register a Stream App with the App Registry by using the Spring Cloud Data Flow Shell app register command. You must provide a unique name, an application type, and a URI that can be resolved to the app artifact. For the type, specify source, processor, sink, or app. Here are a few examples for source, processor and sink:

When providing a URI with the maven scheme, the format should conform to the following:

For example, if you would like to register the snapshot versions of the http and log applications built with the RabbitMQ binder, you could do the following:

If you would like to register multiple apps at one time, you can store them in a properties file where the keys are formatted as <type>.<name> and the values are the URIs.

For example, if you would like to register the snapshot versions of the http and log applications built with the RabbitMQ binder, you could have the following in a properties file (for example, stream-apps.properties):

Then to import the apps in bulk, use the app import command and provide the location of the properties file with the --uri switch, as follows:

Registering an application using --type app is the same as registering a source, processor or sink. Applications of the type app are only allowed to be used in the Stream Application DSL, which uses a comma instead of the pipe symbol in the DSL, and instructs Data Flow not to configure the Spring Cloud Stream binding properties of the application. The application that is registered using --type app does not have to be a Spring Cloud Stream app, it can be any Spring Boot application. See the Stream Application DSL introduction for more information on using this application type.

The Spring Cloud Data Flow Server exposes a full RESTful API for managing the lifecycle of stream definitions, but the easiest way to use is it is through the Spring Cloud Data Flow shell. Start the shell as described in the Getting Started section.

New streams are created with the help of stream definitions. The definitions are built from a simple DSL. For example, consider what happens if we execute the following shell command:

This defines a stream named ticktock that is based off the DSL expression time | log. The DSL uses the "pipe" symbol (|), to connect a source to a sink.

This section describes how to deploy a Stream when the Spring Cloud Data Flow server is responsible for deploying the stream. The following section, Stream Lifecycle with Skipper, covers the new deployment and upgrade features when the Spring Cloud Data Flow server delegates to Skipper for stream deployment. The description of how deployment properties applies to both approaches of Stream deployment.

Give the ticktock stream definition:

dataflow:> stream create --definition "time | log" --name ticktock

To deploy the stream, use the following shell command:

dataflow:> stream deploy --name ticktock

The Data Flow Server resolves time and log to maven coordinates and uses those to launch the time and log applications of the stream, as shown in the following listing:

In the preceding example, the time source sends the current time as a message each second, and the log sink outputs it by using the logging framework. You can tail the stdout log (which has an <instance> suffix). The log files are located within the directory displayed in the Data Flow Server’s log output, as shown in the following listing:

You can also create and deploy the stream in one step by passing the --deploy flag when creating the stream, as follows:

However, it is not very common in real-world use cases to create and deploy the stream in one step. The reason is that when you use the stream deploy command, you can pass in properties that define how to map the applications onto the platform (for example, what is the memory size of the container to use, the number of each application to run, and whether to enable data partitioning features). Properties can also override application properties that were set when creating the stream. The next sections cover this feature in detail.

You can delete a stream by issuing the stream destroy command from the shell, as follows:

dataflow:> stream destroy --name ticktock

If the stream was deployed, it is undeployed before the stream definition is deleted.

Often you want to stop a stream but retain the name and definition for future use. In that case, you can undeploy the stream by name.

You can issue the deploy command at a later time to restart it.

dataflow:> stream deploy --name ticktock

Sometimes the one or more of the apps contained within a stream definition contain an invalid URI in its registration. This can caused by an invalid URI entered at app registration time or the app was removed from the repository from which it was to be drawn. To verify that all the apps contained in a stream are resolve-able, a user can use the validate command. For example:

In the example above the user validated their ticktock stream. As we see that both the source:time and sink:log are valid. Now let’s see what happens if we have a stream definition with a registered app with an invalid URI.

In this case Spring Cloud Data Flow states that the stream is invalid because source:bad-time has an invalid URI.

Skipper extends the Register a Stream App lifecycle with support of multi-versioned stream applications. This allows to upgrade or rollback those applications at runtime using the deployment properties.

Register a versioned stream application using the app register command. You must provide a unique name, application type, and a URI that can be resolved to the app artifact. For the type, specify "source", "processor", or "sink". The version is resolved from the URI. Here are a few examples:

The application URI should conform to one the following schema formats:

Multiple versions can be registered for the same applications (e.g. same name and type) but only one can be set as default. The default version is used for deploying Streams.

The first time an application is registered it will be marked as default. The default application version can be altered with the app default command:

The app list --id <type:name> command lists all versions for a given stream application.

The app unregister command has an optional --version parameter to specify the app version to unregister.

If a --version is not specified, the default version is unregistered.

The stream deploy necessitates default app versions to be set. The stream update and stream rollback commands though can use all (default and non-default) registered app versions.

This will create stream using the default mysource version (0.0.3). Then we can update the version to 0.0.2 like this:

An attempt to update the mysource to version 0.0.1 (not registered) will fail!

You create and deploy a stream by using Skipper in two steps:

The following example shows the two steps in action:

The stream info command shows useful information about the stream, including the deployment properties, as shown (with its output) in the following example:

There is an important optional command argument (called --platformName) to the stream deploy command. Skipper can be configured to deploy to multiple platforms. Skipper is pre-configured with a platform named default, which deploys applications to the local machine where Skipper is running. The default value of the command line argument --platformName is default. If you commonly deploy to one platform, when installing Skipper, you can override the configuration of the default platform. Otherwise, specify the platformName to one of the values returned by the stream platform-list command.

To update the stream, use the command stream update which takes as a command argument either --properties or --propertiesFile. You can pass in values to these command arguments in the same format as when deploy the stream with or without Skipper. There is an important new top level prefix available when using Skipper, which is version. If the Stream http | log was deployed, and the version of log which registered at the time of deployment was 1.1.0.RELEASE, the following command will update the Stream to use the 1.2.0.RELEASE of the log application. Before updating the stream with the specific version of the app, we need to make sure that the app is registered with that version.

To verify the deployment properties and the updated version, we can use stream info, as shown (with its output) in the following example:

When upgrading a stream, the --force option can be used to deploy new instances of currently deployed applications even if no applicaton or deployment properties have changed. This behavior is needed in the case when configuration information is obtained by the application itself at startup time, for example from Spring Cloud Config Server. You can specify which applications to force upgrade by using the option --app-names. If you do not specify any application names, all the applications will be force upgraded. You can specify --force and --app-names options together with --properties or --propertiesFile options.

Skipper keeps a history of the streams that were deployed. After updating a Stream, there will be a second version of the stream. You can query for the history of the versions using the command stream history --name <name-of-stream>.

Skipper keeps a “manifest” of the all the applications, their application properties, and their deployment properties after all values have been substituted. This represents the final state of what was deployed to the platform. You can view the manifest for any of the versions of a Stream by using the following command:

stream manifest --name <name-of-stream> --releaseVersion <optional-version>

If the --releaseVersion is not specified, the manifest for the last version is returned.

The following example shows the use of the manifest:

Using the command results in the following output:

The majority of the deployment and application properties were set by Data Flow to enable the applications to talk to each other and to send application metrics with identifying labels.

You can rollback to a previous version of the stream using the command stream rollback.

The optional --releaseVersion command argument adds the version of the stream. If not specified, the rollback goes to the previous stream version.

The application count is a dynamic property of the system. If, due to scaling at runtime, the application to be upgraded has 5 instances running, then 5 instances of the upgraded application are deployed.

Skipper has a simple 'red/black' upgrade strategy. It deploys the new version of the applications, using as many instances as the currently running version, and checks the /health endpoint of the application. If the health of the new application is good, then the previous application is undeployed. If the health of the new application is bad, then all new applications are undeployed and the upgrade is considered to be not successful.

The upgrade strategy is not a rolling upgrade, so if five applications of the application are running, then in a sunny-day scenario, five of the new applications are also running before the older version is undeployed.

Taps can be created at various producer endpoints in a stream. For a stream such as that defined in the following example, taps can be created at the output of http, step1 and step2:

stream create --definition "http | step1: transform --expression=payload.toUpperCase() | step2: transform --expression=payload+'!' | log" --name mainstream --deploy

To create a stream that acts as a 'tap' on another stream requires specifying the source destination name for the tap stream. The syntax for the source destination name is as follows:

:<streamName>.<label/appName>

To create a tap at the output of http in the preceding stream, the source destination name is mainstream.http To create a tap at the output of the first transform app in the stream above, the source destination name is mainstream.step1

The tap stream DSL resembles the following:

Note the colon (:) prefix before the destination names. The colon lets the parser recognize this as a destination name instead of an app name.

When a stream is made up of multiple apps with the same name, they must be qualified with labels: stream create --definition "http | firstLabel: transform --expression=payload.toUpperCase() | secondLabel: transform --expression=payload+'!' | log" --name myStreamWithLabels --deploy

Instead of referencing a source or sink application, you can use a named destination. A named destination corresponds to a specific destination name in the middleware broker (Rabbit, Kafka, and others). When using the | symbol, applications are connected to each other with messaging middleware destination names created by the Data Flow server. In keeping with the Unix analogy, one can redirect standard input and output using the less-than (<) and greater-than (>) characters. To specify the name of the destination, prefix it with a colon (:). For example, the following stream has the destination name in the source position:

dataflow:>stream create --definition ":myDestination > log" --name ingest_from_broker --deploy

This stream receives messages from the destination called myDestination, located at the broker, and connects it to the log app. You can also create additional streams that consume data from the same named destination.

The following stream has the destination name in the sink position:

dataflow:>stream create --definition "http > :myDestination" --name ingest_to_broker --deploy

It is also possible to connect two different destinations (source and sink positions) at the broker in a stream, as shown in the following example:

dataflow:>stream create --definition ":destination1 > :destination2" --name bridge_destinations --deploy

In the precding stream, both the destinations (destination1 and destination2) are located in the broker. The messages flow from the source destination to the sink destination over a bridge app that connects them.

By using named destinations, you can support fan-in and fan-out use cases. Fan-in use cases are when multiple sources all send data to the same named destination, as shown in the following example:

The preceding example directs the data payloads from the Amazon S3, FTP, and HTTP sources to the same named destination called data. Then an additional stream created with the following DSL expression would have all the data from those three sources sent to the file sink:

:data > file

The fan-out use case is when you determine the destination of a stream based on some information that is only known at runtime. In this case, the Router Application can be used to specify how to direct the incoming message to one of N named destinations.

A nice Video showing Fan-in and Fan-out behavior is also available.

The classes at the heart of the Java DSL are StreamBuilder, StreamDefinition, Stream, StreamApplication, and DataFlowTemplate. The entry point is a builder method on Stream that takes an instance of a DataFlowTemplate. To create an instance of a DataFlowTemplate, you need to provide a URI location of the Data Flow Server.

Spring Boot auto-configuration for StreamBuilder and DataFlowTemplate is also available. The properties in DataFlowClientProperties can be used to configure the connection to the Data Flow server. The common property to start using is spring.cloud.dataflow.client.uri

Consider the following example, using the definition style.

The create method returns an instance of a StreamDefinition representing a Stream that has been created but not deployed. This is called the definition style since it takes a single string for the stream definition, same as in the shell. If applications have not yet been registered in the Data Flow server, you can use the DataFlowOperations class to register them. With the StreamDefinition instance, you have methods available to deploy or destory the stream.

The Stream instance provides getStatus, destroy and undeploy methods to control and query the stream. If you are going to immediately deploy the stream, there is no need to create a separate local variable of the type StreamDefinition. You can just chain the calls together, as follows:

The deploy method is overloaded to take a java.util.Map of deployment properties.

The StreamApplication class is used in the 'fluent' Java DSL style and is discussed in the next section. The StreamBuilder class is returned from the method Stream.builder(dataFlowOperations). In larger applications, it is common to create a single instance of the StreamBuilder as a Spring @Bean and share it across the application.

The Java DSL offers two styles to create Streams.

To demonstrate both styles, we include a simple stream that uses both approaches. A complete sample for you to get started can be found in the Spring Cloud Data Flow Samples Repository.

The following example demonstrates the definition approach:

The following example demonstrates the fluent approach:

The waitAndDestroy method uses the getStatus method to poll for the stream’s status, as shown in the following example:

When using the definition style, the deployment properties are specified as a java.util.Map in the same manner as using the shell. The createDeploymentProperties method is defined as follows:

Is this case, application properties are also overridden at deployment time in addition to setting the deployer property count for the log application. When using the fluent style, the deployment properties are added by using the method addDeploymentProperty (for example, new StreamApplication("log").addDeploymentProperty("count", 2)), and you do not need to prefix the property with deployer.<app_name>.

The Stream applications can also be beans within your application that are injected in other classes to create Streams. There are many ways to structure Spring applications, but one way is to have an @Configuration class define the StreamBuilder and StreamApplications, as shown in the following example:

Then in another class you can @Autowire these classes and deploy a stream.

This style lets you share StreamApplications across multiple Streams.

Regardless of style you choose, the deploy(Map<String, String> deploymentProperties) method allows customization of how your streams will be deployed. We made it a easier to create a map with properties by using a builder style, as well as creating static methods for some properties so you don’t need to remember the name of such properties. If you take the previous example of createDeploymentProperties it could be rewritten as:

This utility class is meant to help with the creation of a Map and adds a few methods to assist with defining pre-defined properties.

As an example of a simple processing step, we can transform the payload of the HTTP posted data to upper case by using the following stream definition: http | transform --expression=payload.toUpperCase() | log

To create this stream enter the following command in the shell dataflow:> stream create --definition "http | transform --expression=payload.toUpperCase() | log" --name mystream --deploy

The following example uses a shell command to post some data:

dataflow:> http post --target localhost:1234 --data "hello"

The preceding example results in an upper-case 'HELLO' in the log, as follows:

2016-06-01 09:54:37.749 INFO 80083 --- [ kafka-binder-] log.sink : HELLO

To demonstrate the data partitioning functionality, the following listing deploys a stream with Kafka as the binder:

You should then see the following in the server logs:

When you review the words.log instance 0 logs, you should see the following:

When you review the words.log instance 1 logs, you shoul see the following:

This example has shown that payload splits that contain the same word are routed to the same application instance.

This example shows something a bit more complicated: swapping out the time source for something else. Another supported source type is http, which accepts data for ingestion over HTTP POSTs. Note that the http source accepts data on a different port from the Data Flow Server (default 8080). By default, the port is randomly assigned.

To create a stream using an http source but still using the same log sink, we would change the original command in the Simple Stream Processing example to the following:

`dataflow:> stream create --definition "http | log" --name myhttpstream --deploy

The preceding command produces the following output from the server:

Note that we do not see any other output this time until we actually post some data (by using a shell command). In order to see the randomly assigned port on which the http source is listening, run the following command:

dataflow:> runtime apps

You should see that the corresponding http source has a url property containing the host and port information on which it is listening. You are now ready to post to that url, as shown in the following example:

The stream then funnels the data from the http source to the output log implemented by the log sink, yielding output similar to the following:

We could also change the sink implementation. You could pipe the output to a file (file), to hadoop (hdfs), or to any of the other sink applications that are available. You can also define your own applications.

The history of the stream can be viewed by running the stream history command, as shown (with its output), in the following example:

The manifest is a YAML document that represents the final state of what was deployed to the platform. You can view the manifest for any stream version by using the stream manifest --name <name-of-stream> --releaseVersion <optional-version> command. If the --releaseVersion is not specified, the manifest for the last version is returned. The following listing shows a typical stream manifest command and its output:

The majority of the deployment and application properties were set by Data Flow in order to enable the applications to talk to each other and send application metrics with identifying labels.

If you compare this YAML document to the one for --releaseVersion=1 you will see the difference in the log application version.

Each application takes properties to customize its behavior. As an example, the timestamp task format setting establishes a output format that is different from the default value.

dataflow:> task create --definition "timestamp --format=\"yyyy\"" --name printTimeStamp

This timestamp property is actually the same as the timestamp.format property specified by the timestamp application. Data Flow adds the ability to use the shorthand form format instead of timestamp.format. One may also specify the longhand version as well, as shown in the following example:

dataflow:> task create --definition "timestamp --timestamp.format=\"yyyy\"" --name printTimeStamp

This shorthand behavior is discussed more in the section on Whitelisting application properties. If you have registered application property metadata you can use tab completion in the shell after typing -- to get a list of candidate property names.

The shell provides tab completion for application properties. The shell command app info --name <appName> --type <appType> provides additional documentation for all the supported properties.

While Spring Cloud Task does provide a number of out-of-the-box applications (at spring-cloud-task-app-starters), most task applications require custom development. To create a custom task application:

You can register a Task App with the App Registry by using the Spring Cloud Data Flow Shell app register command. You must provide a unique name and a URI that can be resolved to the app artifact. For the type, specify "task". The following listing shows three examples:

When providing a URI with the maven scheme, the format should conform to the following:

maven://<groupId>:<artifactId>[:<extension>[:<classifier>]]:<version>

If you would like to register multiple apps at one time, you can store them in a properties file where the keys are formatted as <type>.<name> and the values are the URIs. For example, the followinng listing would be a valid properties file:

Then you can use the app import command and provide the location of the properties file by using the --uri option, as follows:

app import --uri /tmp/task-apps.properties

For convenience, we have the static files with application-URIs (for both maven and docker) available for all the out-of-the-box Task app-starters. You can point to this file and import all the application-URIs in bulk. Otherwise, as explained earlier in this chapter, you can register them individually or have your own custom property file with only the required application-URIs in it. It is recommended, however, to have a “focused” list of desired application-URIs in a custom property file.

The following table lists the available static property files:

For example, if you would like to register all out-of-the-box task applications in bulk, you can do so with the following command:

dataflow:>app import --uri bit.ly/Clark-GA-task-applications-maven

You can also pass the --local option (which is TRUE by default) to indicate whether the properties file location should be resolved within the shell process itself. If the location should be resolved from the Data Flow Server process, specify --local false.

When using either app register or app import, if a task app is already registered with the provided name, it is not overridden by default. If you would like to override the pre-existing task app, then include the --force option.

You can create a task Definition from a task app by providing a definition name as well as properties that apply to the task execution. Creating a task definition can be done through the RESTful API or the shell. To create a task definition by using the shell, use the task create command to create the task definition, as shown in the following example:

A listing of the current task definitions can be obtained through the RESTful API or the shell. To get the task definition list by using the shell, use the task list command.

An adhoc task can be launched through the RESTful API or the shell. To launch an ad-hoc task through the shell, use the task launch command, as shown in the following example:

When a task is launched, any properties that need to be passed as command line arguments to the task application can be set when launching the task, as follows:

Additional properties meant for a TaskLauncher itself can be passed in by using a --properties option. The format of this option is a comma-separated string of properties prefixed with app.<task definition name>.<property>. Properties are passed to TaskLauncher as application properties. It is up to an implementation to choose how those are passed into an actual task application. If the property is prefixed with deployer instead of app, it is passed to TaskLauncher as a deployment property and its meaning may be TaskLauncher implementation specific.

dataflow:>task launch mytask --properties "deployer.timestamp.custom1=value1,app.timestamp.custom2=value2"

Spring Cloud Data Flow allows a user establish the maximum number of concurrently running tasks to prevent the saturation of IaaS/hardware resources. This limit can be configured by setting the spring.cloud.dataflow.task.maximum-concurrent-tasks property. By default it is set to 20. If the number of concurrently running tasks is equal or greater than the value set by spring.cloud.dataflow.task.maximum-concurrent-tasks the next task launch request will be declined and a warning message will be returned via the RESTful API, Shell or UI.

Once the task is launched, the state of the task is stored in a relational DB. The state includes:

A user can check the status of their task executions through the RESTful API or the shell. To display the latest task executions through the shell, use the task execution list command.

To get a list of task executions for just one task definition, add --name and the task definition name, for example task execution list --name foo. To retrieve full details for a task execution use the task execution status command with the id of the task execution, for example task execution status --id 549.

Destroying a Task Definition removes the definition from the definition repository. This can be done through the RESTful API or the shell. To destroy a task through the shell, use the task destroy command, as shown in the following example:

The task execution information for previously launched tasks for the definition remains in the task repository.

Sometimes the one or more of the apps contained within a task definition contain an invalid URI in its registration. This can be caused by an invalid URI entered at app registration time or the app was removed from the repository from which it was to be drawn. To verify that all the apps contained in a task are resolve-able, a user can use the validate command. For example:

In the example above the user validated their time-stamp task. As we see task:timestamp app is valid. Now let’s see what happens if we have a stream definition with a registered app with an invalid URI.

In this case Spring Cloud Data Flow states that the task is invalid because task:badtimestamp has an invalid URI.

Composed tasks are executed through a task application called the Composed Task Runner.

The lifecycle of a composed task has three parts:

Conditional execution is expressed by using a double ampersand symbol (&&). This lets each task in the sequence be launched only if the previous task successfully completed, as shown in the following example:

task create my-composed-task --definition "task1 && task2"

When the composed task called my-composed-task is launched, it launches the task called task1 and, if it completes successfully, then the task called task2 is launched. If task1 fails, then task2 does not launch.

You can also use the Spring Cloud Data Flow Dashboard to create your conditional execution, by using the designer to drag and drop applications that are required and connecting them together to create your directed graph, as shown in the following image:

The preceding diagram is a screen capture of the directed graph as it being created by using the Spring Cloud Data Flow Dashboard. You can see that are four components in the diagram that comprise a conditional execution:

The DSL supports fine- grained control over the transitions taken during the execution of the directed graph. Transitions are specified by providing a condition for equality based on the exit status of the previous task. A task transition is represented by the following symbol ->.

Splits allow multiple tasks within a composed task to be run in parallel. It is denoted by using angle brackets (<>) to group tasks and flows that are to be run in parallel. These tasks and flows are separated by the double pipe || symbol, as shown in the following example:

task create my-split-task --definition "<foo || bar || baz>"

The preceding example above launches tasks foo, bar and baz in parallel.

Using the Spring Cloud Data Flow Dashboard to create the same “split execution” would resemble the following image:

With the task DSL, a user may also execute multiple split groups in succession, as shown in the following example:

`task create my-split-task --definition "<foo || bar || baz> && <qux || quux>"'

In the preceding example, tasks foo, bar, and baz are launched in parallel. Once they all complete, then tasks qux and quux are launched in parallel. Once they complete, the composed task ends. However, if foo, bar, or baz fails, the split containing qux and quux does not launch.

Using the Spring Cloud Data Flow Dashboard to create the same “split with multiple groups” would resemble the following image:

Notice that there is a SYNC control node that is inserted by the designer when connecting two consecutive splits.

One way to launch a task with the task-launcher is to use the triggertask source. The triggertask source emits a message with a TaskLaunchRequest object that contains the required launch information. The triggertask can be added to the available sources by running the app register command, as follows (for the Rabbit Binder, in this case):

app register --type source --name triggertask --uri maven://org.springframework.cloud.stream.app:triggertask-source-rabbit:1.2.0.RELEASE

For example, to launch the timestamp task once every 60 seconds, the stream implementation would be as follows:

If you run runtime apps, you can find the log file for the task launcher sink. By using the tail command on that file, you can find the log file for the launched tasks. Setting of triggertask.environment-properties establishes the Data Flow Server’s H2 Database as the database where the task executions will be recorded. You can then see the list of task executions by using the shell command task execution list, as shown (with its output) in the following example:

Another way to start a task with the task-launcher would be to create a stream by using the Tasklaunchrequest-transform processor to translate a message payload to a TaskLaunchRequest.

The tasklaunchrequest-transform can be added to the available processors by executing the app register command, as follows (for the Rabbit Binder, in this case):

app register --type processor --name tasklaunchrequest-transform --uri maven://org.springframework.cloud.stream.app:tasklaunchrequest-transform-processor-rabbit:1.2.0.RELEASE

The following example shows the creation of a task that includes the tasklaunchrequest-transform:

stream create task-stream --definition "http --port=9000 | tasklaunchrequest-transform --uri=maven://org.springframework.cloud.task.app:timestamp-task:jar:1.2.0.RELEASE | task-launcher-local --maven.remote-repositories.repo1.url=http://repo.spring.io/libs-release"

A composed task can be launched with one of the task-launcher sinks as discussed here. Since we use the ComposedTaskRunner directly, we need to set up the task definitions it uses prior to the creation of the composed task launching stream. Suppose we wanted to create the following composed task definition: AAA && BBB. The first step would be to create the task definitions, as shown in the following example:

Now that the task definitions we need for composed task definition are ready, we need to create a stream that launches ComposedTaskRunner. So, in this case, we create a stream with

The stream should resemble the following:

In the preceding example, we see that the tasklaunchrequest-transform is establishing two primary components:

For now, we focus on the configuration that is required to launch the ComposedTaskRunner:

Spring Cloud Data Flow supports many database types out-of-the-box, so all the user typically has to do is declare the spring_datasource_* environment variables to establish what data store Spring Cloud Data Flow will need. So whatever database you decide to use for Spring Cloud Data Flow make sure that the your task also includes that database dependency in its pom.xml or gradle.build file. If the database dependency that is used by Spring Cloud Data Flow is not present in the Task Application, the task will fail and the task execution will not be recorded.

Spring Cloud Data Flow and your task application must access the same datastore instance. This is so that the task executions recorded by the task application can be read by Spring Cloud Data Flow to list them in the Shell and Dashboard views. Also the task app must have read & write privileges to the task data tables that are used by Spring Cloud Data Flow.

Given the understanding of Datasource dependency between Task apps and Spring Cloud Data Flow, let’s review how to apply them in various Task orchestration scenarios.

Spring Cloud Data Flow will schedule the execution of its tasks via a scheduling agent that is available on the cloud platform. For example: on the Cloud Foundry platform Spring Cloud Data Flow will use the PCF Scheduler.

By default the Spring Cloud Data Flow leaves the scheduling feature disabled. To enable the scheduling feature the following feature properties must be set to true:

The lifecycle of a schedule has 2 parts:

Now let’s create and test an application that prints the current time to the console.

To do so:

To finish our application, we need to modify the generated TimestampApplication code with the following contents so that it will launch a Task.

The @EnableTask annotation sets up TaskRepository which stores information about the task execution such as the start and end time of the task and the exit code.

In our demo, the TaskRepository uses an embedded H2 database to record the results of a task. This H2 embedded database is not a practical solution for a production environment, since the H2 database goes away once the task ends. However, for a quick getting-started experience, we will use this in our example as well as echoing to the logs what is being updated in that repository.

The CommandLineRunner is a Spring Boot interface that tells Boot to execute the code in the run method once. When our sample application runs, Spring Boot launches our TimestampCommandLineRunner and outputs our timestamp message to standard out.

Now let’s open the application.properties file in src/main/resources and configure two properties, the application name and logging. The application name is also used as the name of the task. The logging level is set to DEBUG so we can see more information on what is going on internally.

At this point, our application should work. Since this application is Spring Boot-based, we can run it from the command line by using ./mvnw spring-boot:run from the root of our application, as shown (with its output) in the following example:

The preceding output has three lines that of interest to us here:

Now that we have a working task, we can intentionally create an error, to show how a Spring Cloud Task handles errors. To do so:

Spring Cloud Task includes the ability to run additional processing both before and after the task. To add both features to our current sample application:

Nearly always, a real-world Spring Cloud Task needs to use a persistent (rather than an in-memory) database. In this example, we show how to add a MySQL database (MariaDB) to our Task. To do so: