The Spring Cloud Data Flow implementation for Kubernetes uses the Spring Cloud Deployer Kubernetes library for orchestration. Before you begin setting up a Kubermetes cluster, refer to the compatibility matrix to learn more about deployer and server compatibility against Kubernetes release versions.

The Kubernetes Picking the Right Solution guide lets you choose among many options, so you can pick the one that you are most comfortable using.

All our testing is done with the Google Kubernetes Engine that is part of the Google Cloud Platform. That is also the target platform for this section. We have also successfully deployed with Minikube. We note where you need to adjust for deploying on Minikube.

The rest of this getting started guide assumes that you have a working Kubernetes cluster and a kubectl command line utility. See the docs for installation instructions: Installing and Setting up kubectl.

To deploy with kubectl

This section describes how to create streams without using Skipper. To do so:

You can create a stream with Spring Cloud Skipper. Skipper lets you discover applications and manage their lifecycles. For more about Skipper, see cloud.spring.io/spring-cloud-skipper/. For more about using Skipper with streams, see Deploying Streams by Using Skipper.

If you need to be able to connect from outside of the Kubernetes cluster to an application that you deploy (such as the http-source), you need to use either an external load balancer for the incoming connections or you need to use a NodePort configuration that exposes a proxy port on each Kubetnetes node. If your cluster does not support external load balancers (such as Minikube), you must use the NodePort approach. You can use deployment properties to configure the access. To specify that you want to have a load balancer with an external IP address created for your application’s service, use deployer.http.kubernetes.createLoadBalancer=true for the application. For the NodePort configuration, use deployer.http.kubernetes.createNodePort=<port>, where <port> is a number between 30000 and 32767.

The deployer uses ReplicationController by default. To use deployments instead, you can set the following option as part of the container env section in a deployment YAML file:

This is now the preferred setting and will be the default in future releases of the deployer.

The apps are deployed by default with the following Limits and Requests settings:

You might find that the 512Mi memory limit is too low. To increase it, you can provide a common spring.cloud.deployer.memory deployer property, as the following example shows (replace <app> with the name of the app for which you want to set the memory):

This property affects both the Requests and Limits memory value set for the container.

If you want to set the Requests and Limits values separately, you can use the deployer properties that are specific to the Kubernetes deployer. The following example shows how to set Limits to 1000m for CPU and 1024Mi for memory and Requests to 800m for CPU and 640Mi for memory:

Those values results in the following container settings being used:

You can also control the default values to which to set the cpu and memory requirements for the pods that are created as part of application deployments. You can declare the following as part of the container env section in a deployment YAML file:

The settings we have used so far only affect the settings for the container. They do not affect the memory setting for the JVM process in the container. If you would like to set JVM memory settings, you can provide an environment variable to do so. See the next section for details.

To influence the environment settings for a given application, you can take advantage of the spring.cloud.deployer.kubernetes.environmentVariables deployer property. For example, a common requirement in production settings is to influence the JVM memory arguments. You can achieve this by using the JAVA_TOOL_OPTIONS environment variable, as the following example shows:

This overrides the JVM memory setting for the desired <app> (replace <app> with the name of your application).

The liveness and readiness probes use paths called /health and /info respectively. They use a delay of 10 for both and a period of 60 and 10 respectively. You can change these defaults when you deploy the stream by using deployer properties.

The following example changes the liveness probe (replace <app> with the name of your application) by setting deployer properties:

The same can be declared as part of the container env section in a deployment YAML file:

Similarly, you can swap liveness for readiness to override the default readiness settings.

By default, port 8080 is used as the probe port. You can change the defaults for both liveness and readiness probe ports by using deployer properties, as the following example shows:

You can also set the port values in the container env section of a deployment YAML file:

You can access secured probe endpoints by using credentials stored in a Kubernetes secret. You can use an existing secret, provided the credentials are contained under the credentials key name of the secret’s data block. You can configure probe authentication on a per-application basis. When enabled, it is applied to both the liveness and readiness probe endpoints by using the same credentials and authentication type. Currently, only Basic authentication is supported.

To create a new secret:

You can use a SPRING_APPLICATION_JSON environment variable to set Data Flow server properties (including the configuration of maven repository settings) that are common across all of the Data Flow server implementations. These settings go at the server level in the container env section of a deployment YAML. The following example shows how to do so:

You can pull Docker images from a private registry on a per-application basis. First, you must create a secret in the cluster. Follow the Pull an Image from a Private Registry guide to create the secret.

Once you have created the secret, use the imagePullSecret property to set the secret to use, as the following example shows:

Replace <app> with the name of your application and mysecret with the name of the secret you created earlier.

You can also configure the image pull secret at the server level in the container env section of a deployment YAML, as the following example shows:

Replace mysecret with the name of the secret you created earlier.

You can add annotations to Kubernetes objects on a per-application basis. The supported object types are pod Deployment, Service and Job. Annotations are defined in a key:value format allowing for multiple annotations separated by a comma. For more information and use cases on annotations see Annotations.

The following example shows how you can configure applications to use annotations:

Replace <app> with the name of your application and the value of your annotations.

An Entry Point Style affects how application properties are passed to the container to be deployed. Currently, three styles are supported:

You can configure applications as follows:

Replace <app> with the name of your application and <Entry Point Style> with your desired Entry Point Style.

You can also configure the Entry Point Style at the server level in the container env section of a deployment YAML, as the following example shows:

Replace entryPointStye with the desired Entry Point Style.

You should choose an Entry Point Style of either exec or shell, to correspond to how the ENTRYPOINT syntax is defined in the container’s Dockerfile. For more information and uses cases on exec vs shell, see the ENTRYPOINT section of the Docker documentation.

Using the boot Entry Point Style corresponds to using the exec style ENTRYPOINT. Command line arguments from the deployment request are passed to the container, with the addition of application properties mapped into the SPRING_APPLICATION_JSON environment variable rather than command line arguments.

You can configure a custom service account for application deployments through properties. You can use an existing service account or create a new one. One way to create a service account is by using kubectl, as the following example shows:

Then you can configure individual applications as follows:

Replace <app> with the name of your application and myserviceaccountname with your service account name.

You can also configure the service account name at the server level in the container env section of a deployment YAML, as the following example shows:

Replace myserviceaccountname with the service account name to be applied to all deployments.

An image pull policy defines when a Docker image should be pulled to the local registry. Currently, three policies are supported:

The following example shows how you can individually configure applications:

Replace <app> with the name of your application and Always with your desired image pull policy.

You can configure an image pull policy at the server level in the container env section of a deployment YAML, as the following example shows:

Replace Always with your desired image pull policy.

Custom labels can be set on Deployment related objects. See Labels for more information on labels. Labels are specified in key:value format.

The following example shows how you can individually configure applications:

Replace <app> with the name of your application, myLabelName with your label name and myLabelValue with the value of your label.

Additionally, multiple labels can be applied, for example:

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 {spring-cloud-stream-docs}#_consumer_properties[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-docs}#_partitioning[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 {spring-cloud-stream-docs}#_persistent_publish_subscribe_support[persistent publish-subscribe semantics].

The {spring-cloud-stream-docs}#_binders[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 {spring-cloud-stream-docs}#_consumer_properties[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 {spring-cloud-stream-docs}#_rabbitmq_consumer_properties[consumer properties] and the autoBindDlq {spring-cloud-stream-docs}#_rabbit_producer_properties[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 {spring-cloud-stream-docs}#_kafka_consumer_properties[Consumer] and {spring-cloud-stream-docs}#_kafka_producer_properties[Producer] and Rabbit {spring-cloud-stream-docs}#_rabbitmq_consumer_properties[Consumer] and {spring-cloud-stream-docs}#_rabbit_producer_properties[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 pass configuration properties to the Data Flow Server by using Kubernetes ConfigMap and secrets.

The following example shows one possible configuration, which enables RabbitMQ, MySQL and Redis as well as basic security settings for the server:

The preceding example assumes that RabbitMQ is deployed with rabbitmq as the service name. For MySQL, it assumes that the service name is mysql. For Redis, it assumes that the service name is redis. Kubernetes publishes the host and port values of these services as environment variables that we can use when configuring the apps we deploy.

We prefer to provide the MySQL connection password in a Secrets file, as the following example shows:

The password is a base64-encoded value.

You can access the server logs by using the following command:

Stream applications are deployed with the stream name followed by the name of the application. For processors and sinks, an instance index is also appended.

To see all the pods that are deployed by the Spring Cloud Data Flow server, you can specify the role=spring-app label, as follows:

To see details for a specific application deployment you can use the following command:

To view the application logs, you can use the following command:

If you would like to tail a log you can use the following command:

Tasks are launched as bare pods without a replication controller. The pods remain after the tasks complete, which gives you an opportunity to review the logs.

To see all pods for a specific task, use the following command:

To review the task logs, use the following command:

You have two options to delete completed pods. You can delete them manually once they are no longer needed or you can use the Data Flow shell task execution cleanup command to remove the completed pod for a task execution.

To delete the task pod manually, use the following command:

To use the task execution cleanup command, you must first determine the ID for the task execution. To do so, use the task execution list command, as the following example (with output) shows:

Once you have the ID, you can issue the command to cleanup the execution artifacts (the completed pod), as the following example shows:

An Entry Point Style affects how application properties are passed to the task container to be deployed. Currently, three styles are supported:

You can configure the entry point style as follows:

Replace <Entry Point Style> with your desired Entry Point Style.

You can also configure the Entry Point Style at the server level in the container env section of a deployment YAML, as the following example shows:

Replace entryPointStyle with the desired Entry Point Style.

You should choose an Entry Point Style of either exec or shell, to correspond to how the ENTRYPOINT syntax is defined in the container’s Dockerfile. For more information and uses cases on exec vs shell, see the ENTRYPOINT section of the Docker documentation.

Using the boot Entry Point Style corresponds to using the exec style ENTRYPOINT. Command line arguments from the deployment request are passed to the container, with the addition of application properties mapped into the SPRING_APPLICATION_JSON environment variable rather than command line arguments.

To influence the environment settings for a given application, you can take advantage of the spring.cloud.scheduler.kubernetes.environmentVariables property. For example, a common requirement in production settings is to influence the JVM memory arguments. You can achieve this by using the JAVA_TOOL_OPTIONS environment variable, as the following example shows:

Additionally you can configure environment variables at the server level in the container env section of a deployment YAML, as the following example shows:

Replace myVar=myVal with your desired environment variables.

An image pull policy defines when a Docker image should be pulled to the local registry. Currently, three policies are supported:

The following example shows how you can individually configure containers:

Replace Always with your desired image pull policy.

You can configure an image pull policy at the server level in the container env section of a deployment YAML, as the following example shows:

Replace Always with your desired image pull policy.

Docker images that are private and require authentication can be pulled by configuring a Secret. First, you must create a Secret in the cluster. Follow the Pull an Image from a Private Registry guide to create the Secret.

Once you have created the secret, use the imagePullSecret property to set the secret to use, as the following example shows:

Replace mysecret with the name of the secret you created earlier.

You can also configure the image pull secret at the server level in the container env section of a deployment YAML, as the following example shows:

Replace mysecret with the name of the secret you created earlier.

By default the namespace used for scheduled tasks is default. This value can be set at the server level configuration in the container env section of a deployment YAML, as the following example shows:

You can configure a custom service account for scheduled tasks through properties. An existing service account can be used or a new one created. One way to create a service account is by using kubectl, as the following example shows:

Then you can configure the service account to use on a per-schedule basis as follows:

Replace myserviceaccountname with your service account name.

You can also configure the service account name at the server level in the container env section of a deployment YAML, as the following example shows:

Replace myserviceaccountname with the service account name to be applied to all deployments.

For more information on scheduling tasks see Scheduling Tasks.

To manually enable JDWP, first edit src/kubernetes/server/server-deployment.yaml and add an additional containerPort entry under spec.template.spec.containers.ports with a value of 5005. Additionally, add the JAVA_TOOL_OPTIONS environment variable under spec.template.spec.containers.env as the following example shows:

You can now start the Spring Cloud Data Flow Kubernetes Server. Once the server is up, you can verify the configuration changes on the scdf-server deployment, as the following example (with output) shows:

With the server started and JDWP enabled, you need to configure access to the port. In this example, we use the port-forward subcommand of kubectl. The following example (with output) shows how to expose a local port to your debug target by using port-forward:

You can now attach a debugger by pointing it to 127.0.0.1 as the host and 5005 as the port. The port-forward subcommand runs until stopped (by pressing CTRL+c, for example).

You can remove debugging support by reverting the changes to src/kubernetes/server/server-deployment.yaml. The reverted changes are picked up on the next deployment of the Spring Cloud Data Flow Kubernetes Server. Manually adding debug support to the configuration is useful when debugging should be enabled by default each time the server is deployed.

Rather than manually changing the server-deployment.yaml, Kubernetes objects can be “patched” in place. For convenience, patch files that provide the same configuration as the manual approach are included. To enable debugging by patching, use the following command:

Running the preceding command automatically adds the containerPort attribute and the JAVA_TOOL_OPTIONS environment variable. The following example (with output) shows how toverify changes to the scdf-server deployment:

To enable access to the debug port, rather than using the port-forward subcommand of kubectl, you can patch the scdf-server Kubernetes service object. You must first ensure that the scdf-server Kubernetes service object has the proper configuration. The following example (with output) shows how to do so:

If the output contains <unset>, you must patch the service to add a name for this port. The following example shows how to do so:

Now you can add the debug port, as the following example shows:

The following example (with output) shows how to verify the mapping:

The output shows that container port 5005 has been mapped to the NodePort of 31339. The following example (with output) shows how to get the IP address of the Minikube node:

With this information, you can create a debug connection by using a host of 192.168.99.100 and a port of 31339.

The following example shows how to disable JDWP:

The Kubernetes deployment object is rolled back to its state before being patched. The Kubernetes service object is then patched with a remove operation to remove port 5005 from the containerPorts list.

See Rolling Back a Deployment for more information on deployment rollbacks, including managing history and Updating API Objects in Place Using kubectl Patch.

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.

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. When using the Cloud Foundry platform Spring Cloud Data Flow will use the PCF Scheduler. When using Kubernetes, a CronJob will be used.

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:

The Bulk Import Applications page provides numerous options for defining and importing a set of applications all at once. For bulk import, the application definitions are expected to be expressed in a properties style, as follows:

<type>.<name> = <coordinates>

The following examples show a typical application definitions:

At the top of the bulk import page, a URI can be specified that points to a properties file stored elsewhere, it should contain properties formatted as shown in the previous example. Alternatively, by using the textbox labeled “Apps as Properties”, you can directly list each property string. Finally, if the properties are stored in a local file, the “Select Properties File” option opens a local file browser to select the file. After setting your definitions through one of these routes, click Import.

The following image shows the Bulk Import Applications page:

The Streams section of the Dashboard includes the Definitions tab that provides a listing of Stream definitions. There you have the option to deploy or undeploy those stream definitions. Additionally, you can remove the definition by clicking on Destroy. Each row includes an arrow on the left, which you can click to see a visual representation of the definition. Hovering over the boxes in the visual representation shows more details about the apps, including any options passed to them.

In the following screenshot, the timer stream has been expanded to show the visual representation:

If you click the details button, the view changes to show a visual representation of that stream and any related streams. In the preceding example, if you click details for the timer stream, the view changes to the following view, which clearly shows the relationship between the three streams (two of them are tapping into the timer stream):

The Streams section of the Dashboard includes the Create Stream tab, which makes available the Spring Flo designer: a canvas application that offers an interactive graphical interface for creating data pipelines.

In this tab, you can:

You should watch this screencast that highlights some of the "Flo for Spring Cloud Data Flow" capabilities. The Spring Flo wiki includes more detailed content on core Flo capabilities.

The following image shows the Flo designer in use:

The stream deploy page includes tabs that provide different ways to setup the deployment properties and deploy the stream. The following screenshots show the stream deploy page for foobar (time | log).

You can define deployments properties using:

You can switch between the both views, the form builder provides a more stronger validation of the inputs.

In chapter Fan-in and Fan-out you learned how we can support fan-in and fan-out use cases using named destinations. The UI provides dedicated support for named destinations as well:

In this example we have data from an HTTP Source and a JDBC Source that is being sent to the sharedData channel which represents a Fan-in use case. On the other end we have a Cassandra Sink and a File Sink subscribed to the sharedData channel which represents a Fan-out use case.

Creating Taps using the Dashboard is straightforward. Let’s say you have stream consisting of an HTTP Source and a File Sink and you would like to tap into the stream to also send data to a JDBC Sink. In order to create the tap stream simply connect the output connector of the HTTP Source to the JDBC Sink. The connection will be displayed as a dotted line, indicating that you created a tap stream.

The primary stream (HTTP Source to File Sink) will be automatically named, in case you did not provide a name for the stream, yet. When creating tap streams, the primary stream must always be explicitly named. In the picture above, the primary stream was named HTTP_INGEST.

Using the Dashboard, you can also switch the primary stream to become the secondary tap stream.

Simply hover over the existing primary stream, the line between HTTP Source and File Sink. Several control icons will appear, and by clicking on the icon labeled Switch to/from tap, you change the primary stream into a tap stream. Do the same for the tap stream and switch it to a primary stream.

Each app encapsulates a unit of work into a reusable component. Within the Data Flow runtime environment, apps let users create definitions for streams as well as tasks. Consequently, the Apps tab within the Tasks section lets users create task definitions.

The following image shows a typical list of task apps:

On this screen, you can perform the following actions:

This page lists the Data Flow task definitions and provides actions to launch or destroy those tasks. It also provides a shortcut operation to define one or more tasks with simple textual input, indicated by the Bulk Define Tasks button.

The following image shows the Definitions page:

The Executions tab shows the current running and completed tasks.

The following image shows the Executions tab:

After having launched a batch job, the Job Execution Details page will show information about the job.

The following image shows the Job Execution Details page:

The Job Execution Details page contains a list of the executed steps. You can further drill into the details of each step’s execution by clicking the magnifying glass icon.

The Step Execution Details page provides information about an individual step within a job.

The following image shows the Step Execution Details page:

On the top of the page, you can see a progress indicator the respective step, with the option to refresh the indicator. A link is provided to view the step execution history.

The Step Execution Details screen provides a complete list of all Step Execution Context key/value pairs.

On this screen, you can see a progress bar indicator in regards to the execution of the current step. Under the Step Execution History, you can also view various metrics associated with the selected step, such as duration, read counts, write counts, and others.

From the Task Definitions page a user can create or delete a schedule for a specific task definition.

On this screen you can perform the following actions:

Once the user clicks the clock icon on the Task Definition screen, Spring Cloud Data Flow will take the user to the Schedule Creation screen. On this screen a user can establish the schedule name, the cron expression as well as establish the properties and arguments to be used when the task is launched by this schedule.

On this screen you can perform the following actions: