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

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

All our testing is done using the Google Kubernetes Engine that is part of the Google Cloud Platform. That is a also the target platform for this section. We have also successfully deployed using Minikube and we will 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.

This is an optional step. Deploy Skipper if you want the added features of upgrading and rolling back Streams since Data Flow delegates to Skipper for those features.

The Deployment resource for Skipper is shown below:

The resource for the Skipper service is shown below:

Run the following commands to start Skipper as the companion server for Spring Cloud Data Flow:

You can use the command kubectl get all -l app=skipper to verify that the deployment, pod and service resources are running. Use the command kubectl delete all -l app=skipper to clean up afterwards.

Use the kubectl get svc scdf-server command to locate the EXTERNAL_IP address assigned to scdf-server, we will use that later to connect from the shell.

So the URL you need to use is in this case is: 130.211.203.247

If you are using Minikube then you don’t have an external load balancer and the EXTERNAL-IP will show as <pending>. You need to use the NodePort assigned for the skipper service. Use this command to look up the URL to use:

Refer to the section Streams deployed using Skipper for more information.

If you need to be able to connect to from outside of the Kubernetes cluster to an app that you deploy, like the http-source, then you need to use either an external load balancer for the incoming connections or you need to use a NodePort configuration that will expose a proxy port on each Kubetnetes Node. If your cluster doesn’t support external load balancers, like the Minikube, then you must use the NodePort approach. You can use deployment properties for configuring the access. Use deployer.http.kubernetes.createLoadBalancer=true for the app to specify that you want to have a LoadBalancer with an external IP address created for your app’s service. For the NodePort configuration use deployer.http.kubernetes.createNodePort=<port> where <port> should be a number between 30000 and 32767.

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

You might find that the 512Mi memory limit is too low and to increase it you can provide a common spring.cloud.deployer.memory deployer property like this (replace <app> with the name of the app you would like to set this for):

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

If you would like to set the Requests and Limits values separately you would have to use the deployer properties that are specific to the Kubernetes deployer. To set the Limits to 1000m for cpu, 1024Mi for memory and Requests to 800m for cpu, 640Mi for memory you can use the following properties:

That should result in the following container settings being used:

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 for this, see the next section for details.

To influence the environment settings for a given app, 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. This can be achieved by using the JAVA_TOOL_OPTIONS environment variable:

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

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

Here is an example changing the liveness probe (just replace <app> with the name of your app):

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

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:

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.

Configuration properties can be passed to the Data Flow Server using Kubernetes ConfigMap and Secrets.

An example configuration could look like the following where we configure Rabbit MQ, MySQL and Redis as well as basic security settings for the server:

We assume here that Rabbit MQ is deployed using rabbitmq as the service name. For MySQL we assume the service name is mysql and for Redis we assume it is redis. Kubernetes will publish these services' host and port values 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:

The password is provided as a base64 encoded value.

The deployer uses Replication Controllers 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.

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

You can modify the settings used for the liveness and readiness probes. This might be necessary if your cluster is slower and the apps need more time to start up. Here is an example of setting the delay and period for the liveness probe:

See KubernetesDeployerProperties for more of the supported options.

Data Flow Server properties that are common across all of the Data Flow Server implementations including the configuration of maven repository settings can be set in a similar manner although the latter might be easier to set using a SPRING_APPLICATION_JSON environment variable like:

You can access the server logs by using the following command (just supply the name of pod for the server):

The stream apps are deployed with the stream name followed by the name of the app and for processors and sinks there is also an instance index appended.

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

To see details for a specific app deployment you can use (just supply the name of pod for the app):

For the application logs use:

If you would like to tail a log you can use:

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

To see all pods for a specific task use this command while providing the task name:

To review the task logs use:

You have two options to delete completed pods. You can delete them manually once they are no longer needed.

To delete the task pod use:

You can also use the Data Flow shell command task execution cleanup command to remove the completed pod for a task execution.

First we need to determine the ID for the task execution:

Next we issue the command to cleanup the execution artifacts (the completed pod):

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.

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 <appType>:<appName> 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, or sink. Here are a few examples:

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:

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

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:

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 <appType>:<appName> 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:

dataflow:>task launch mytask --arguments "--server.port=8080,--custom=value"

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"

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.

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.

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:

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.

Spring Cloud Data Flow is built upon several Spring projects, but ultimately the dataflow-server is a Spring Boot app, so the logging techniques that apply to any Spring Boot application are applicable here as well.

While troubleshooting, following are the two primary areas where enabling the DEBUG logs could be useful.

When you use the Data Flow Template, the only needed Data Flow dependency is the Spring Cloud Data Flow Rest Client, as shown in the following Maven snippet:

With that dependency, you get the DataFlowTemplate class as well as all the dependencies needed to make calls to a Spring Cloud Data Flow server.

When instantiating the DataFlowTemplate, you also pass in a RestTemplate. Please be aware that the needed RestTemplate requires some additional configuration to be valid in the context of the DataFlowTemplate. When declaring a RestTemplate as a bean, the following configuration suffices:

Now you can instantiate the DataFlowTemplate with the following code:

Depending on your requirements, you can now make calls to the server. For instance, if you want to get a list of currently available applications you can run the following code:

The following table describes the changes in terminology from Spring XD to Spring Cloud Data Flow:

If you have custom Spring XD modules, you need refactor them to use Spring Cloud Stream and Spring Cloud Task annotations. As part of that work, you need to update dependencies and build them as normal Spring Boot applications.

Terminology wise, in Spring Cloud Data Flow, the message bus implementation is commonly referred to as binders.

A Task, by definition, is any application that does not run forever, and they end at some point. Tasks include Spring Batch jobs. Task applications can be used for on-demand use cases, such as database migration, machine learning, scheduled operations, and others. With Spring Cloud Task, you can build Spring Batch jobs as microservice applications.

The following table shows the changes to shell and DSL commands:

The following table shows the changes to the REST API:

The Admin-UI is now named Dashboard. The URI for accessing the Dashboard is changed from localhost:9393/admin-ui to localhost:9393/dashboard.

Spring Cloud Data Flow comes with a significantly simplified architecture. In fact, when compared with Spring XD, you need fewer peripherals to use Spring Cloud Data Flow.

To support centralized and consistent management of an application’s configuration properties, Spring Cloud Config client libraries have been included in the Spring Cloud Data Flow server as well as the Spring Cloud Stream applications provided by the Spring Cloud Stream App Starters. You can also pass common application properties to all streams when the Data Flow Server starts.

Spring Cloud Data Flow is a Spring Boot application. Depending on the platform of your choice, you can download the respective release uber jar and deploy or push it to the runtime platform (Cloud Foundry, Apache Yarn, Kubernetes, or Apache Mesos). For example, if you run Spring Cloud Data Flow on Cloud Foundry, you can download the Cloud Foundry server implementation and do a cf push, as explained in the Cloud Foundry Reference Guide.

The hdfs-sink application builds upon Spring Hadoop 2.4.0 release, so this application is compatible with the following Hadoop distributions:

Spring Cloud Data Flow can be deployed and used with Apche YARN in two different ways:

The remainder of this appendix reviews some use cases to show the differences between Spring XD and Spring Cloud Data Flow.

There is a full profile that generates documentation. You can build only the documentation by using the following command:

$ ./mvnw clean package -DskipTests -P full -pl spring-cloud-dataflow-server-kubernetes-docs -am

If you do not have an IDE preference, we recommend that you use Spring Tools Suite or Eclipse when working with the code. We use the m2eclipse Eclipse plugin for Maven support. Other IDEs and tools generally also work without issue.

Before we accept a non-trivial patch or pull request, we need you to sign the contributor’s agreement. Signing the contributor’s agreement does not grant anyone commit rights to the main repository, but it does mean that we can accept your contributions, and you will get an author credit if we do. Active contributors might be asked to join the core team and be given the ability to merge pull requests.

None of the following guidelines is essential for a pull request, but they all help your fellow developers understand and work with your code. They can also be added after the original pull request but before a merge.