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 complexity of another execution environment that is often not needed when creating data centric applications. That doesn’t mean you cannot do real time data computations when using Spring Cloud Data Flow. Refer to the analytics section which describes the integration of Redis to handle common counting based use-cases as well as the RxJava integration for functional API driven analytics use-cases, such as time-sliding-window and moving-average among others.

Similarly, 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 execute on the cluster and perform health checks to ensure that long lived applications are restarted if they fail. Often, framework specific interfaces are required to be used 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 efforts. 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, runtimes that 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.

Spring Cloud Stream is most closely integrated with Spring Integration’s imperative "event at a time" programming model. This means you write code that handles a single event callback. For 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 what is used to tie together the input channel to the external middleware.

However, Spring Cloud Stream can support other programming styles. The use of reactive APIs where incoming and outgoing data is handled as continuous data flows and it defines how each individual message should be handled. You can also use operators that describe functional transformations from inbound to outbound data flows. The upcoming versions will support Apache Kafka’s KStream API in the programming model.

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 out data to multiple messaging destinations.

Taps can be used to ‘listen in’ to the data that if flowing across any of the pipe symbols. Taps can be used as sources for new streams with an in independent life cycle.

For an application that will consume 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 (e.g., Kafka topics) or not (e.g., 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 only need 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 will be 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 isn’t shown in the diagram above). Suppose the payload being sent to the http source was in JSON format and had a field called sensorId. Deploying the stream with the shell command stream deploy ingest --propertiesFile ingestStream.properties where the contents of the file ingestStream.properties are

will 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 can’t currently scale partitioned streams. Read the section 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 using the {spring-cloud-stream-docs}#_consumer_properties[common consumer properties] maxAttempts, backOffInitialInterval, backOffMaxInterval, and backOffMultiplier. The default values of these properties will 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 will become the payload of a message and be 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 will send 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 (e.g in the case of Kafka it is a dedicated topic). To enable this for RabbitMQ set the {spring-cloud-stream-docs}#_rabbitmq_consumer_properties[consumer properties] republishtoDlq and autoBindDlq and the {spring-cloud-stream-docs}#_rabbit_producer_properties[producer property] autoBindDlq 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 will find extensive declarative support for all the native QOS options.

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 using Spring’s MVC framework and the Spring HATEOAS library to create REST representations that follow the HATEOAS principle.

Each Data Flow Server executable jar targets a single runtime by delegating to the implementation of the deployer Service Provider Interface found on the classpath.

We provide a Data Flow Server executable jar that targets a single runtime. The Data Flow server delegates to the implementation of the deployer Service Provider Interface found on the classpath. In the current version, there are no endpoints specific to a target runtime, but may be available in future releases as a convenience to access runtime specific features

While we provide a server executable for each of the target runtimes you can also create your own customized server application using Spring Initialzr. This let’s you add or remove functionality relative to the executable jar we provide. For example, adding additional security implementations, custom endpoints, or removing Task or Analytics REST endpoints. You can also enable or disable some features through the use of feature toggles.

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.

Authorization via groups is planned for a future release.

The target runtimes supported by Data Flow all have the ability to restart a long lived application should it fail. Spring Cloud Data Flow sets up whatever health probe is required by the runtime environment when deploying the application.

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

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

When deploying a stream, you can set the instance count for each individual application that comprises 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. Future work will provide a portable command in the Data Flow Server to perform this operation.

Currently, this is not supported with the Kafka binder (based on the 0.8 simple consumer at the time of the release), as well as partitioned streams, for which the suggested workaround is redeploying the stream with an updated number of instances. Both cases require a static consumer set up based on information about the total instance count and current instance index, a limitation intended to be addressed in future releases. For example, Kafka 0.9 and higher provides good infrastructure for scaling applications dynamically and will be available as an alternative to the current Kafka 0.8 based binder in the near future. One specific concern regarding scaling partitioned streams is the handling of local state, which is typically reshuffled as the number of instances is changed. This is also intended to be addressed in the future versions, by providing first class support for local state management.

Application versioning, that is upgrading or downgrading an application from one version to another, is not directly supported by Spring Cloud Data Flow. You must rely on specific target runtime features to perform these operational tasks.

The roadmap for Spring Cloud Data Flow will deploy applications that are compatible with Spinnaker to manage the complete application lifecycle. This also includes automated canary analysis backed by application metrics. Portable commands in the Data Flow server to trigger pipelines in Spinnaker are also planned.

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 that 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 will encounter when all three components are involved.

A stream is defined 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 will connect the standard ouput of the program 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 via 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 either 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:

A stream that involves some processing would be expresed as:

Stream definitions can be created using the shell’s create stream command. For example:

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

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

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 shell is calling the Data Flow Servers' REST API. For more information on making HTTP request 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 which allows the data ingestion port to be changed from the default value.

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.

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 and also the shell command app info <appType>:<appName> provides additional documentation for all the supported properties.

Register a Stream App with the App Registry using the Spring Cloud Data Flow Shell 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". 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 [eg: stream-apps.properties]:

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

For convenience, we have the static files with application-URIs (for both maven and docker) available for all the out-of-the-box stream and task/batch app-starters. You can point to this file and import all the application-URIs in bulk. Otherwise, as explained in previous paragraphs, 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.

List of available Stream Application Starters:

List of available Task Application Starters:

You can find more information about the available task starters in the Task App Starters Project Page and related reference documentation. For more information about the available stream starters look at the Stream App Starters Project Page and related reference documentation.

As an example, ff you would like to register all out-of-the-box stream applications built with the Kafka binder in bulk, you can with the following command.

Alternatively you can register all the stream applications with the Rabbit binder

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.

While there are out of the box source, processor, sink applications available, one can extend these applications or write a custom Spring Cloud Stream application.

The process of creating Spring Cloud Stream applications via Spring Initializr is detailed in the Spring Cloud Stream {spring-cloud-stream-docs}#_getting_started[documentation]. It is possible to include multiple binders to an application. If doing so, refer the instructions in Passing Spring Cloud Stream properties on how to configure them.

For supporting property whitelisting, Spring Cloud Stream applications running in Spring Cloud Data Flow may include the Spring Boot configuration-processor as an optional dependency, as in the following example.

Once a custom application has been created, it can be registered as described in Register a Stream App.

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 via the Spring Cloud Data Flow shell. Start the shell as described in the Getting Started section.

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

This defines a stream named ticktock 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. In both cases, the description of how deployment properties applies to both approaches of Stream deployment.

Give the ticktock stream definition:

You can deploy the stream using the following command: Then to deploy the stream execute the following shell command

The Data Flow Server resolves time and log to maven coordinates and uses those to launch the time and log applications of the stream.

In this example, the time source simply sends the current time as a message each second, and the log sink outputs it 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 above.

You can also create an deploy the stream in one step by passing the --deploy flag when creating the stream.

However, it is not very common in real world use cases to do 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, e.g. what is the memory size of the container to use, the number of each application to run, or to enable data partitioning features. Properties can also override application properties which were set when creating the stream. The next sections cover this in detail.

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

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

Often you will 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.

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 as follows:

If you want to pass deployment properties, you can create and deploy a stream in two steps:

The command stream info shows useful information about the stream including the deployment properties.

There is an important optional command argument to the stream deploy command, which is --platformName. Skipper can be configured to deploy to multiple platforms. Skipper is pre-configured with a platform named default which will deploys applications to the local machine where Skipper is running. The default value of the command line argument --platformName is default. If you are commonly deploying 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 command stream platform-list

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

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 an "manifest" of the all the applications, their application properties and 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 using the 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 majority of the deployment and application properties were set by Data Flow in order to enable the applications to talk to each other and sending application metrics with identifying labels.

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

There is an optional --releaseVersion command argument which is 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 will be deployed.

Skipper has a simple 'red/black' upgrade strategy. It deploys the new version of the applications, 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 not successful.

The upgrade strategy is not a rolling upgrade, so if 5 applications of the application to upgrade are runningn, then in a sunny day scenario, 5 of the new applications will also be running before the older version is undeployed. Future versions of Skipper will support rolling upgrades and other types of checks, e.g. manual, to continue to upgrade process.

Taps can be created at various producer endpoints in a stream. For a stream like this:

taps can be created at the output of http, step1 and step2.

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

To create a tap at the output of http in the stream above, 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 looks like this:

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

When a stream is comprised of multiple apps with the same name, they must be qualified with labels:

Instead of referencing a source or sink applications, you can use a named destination. A named destination corresponds to a specific destination name in the middleware broker (Rabbit, Kafka, etc.,). When using the | symbol, applications are connected to each other using 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 < greater-than > charaters. 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:

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

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

It is also possible to connect two different destinations (source and sink positions) at the broker in a stream.

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

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. For example

Would direct 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 DSL expression

would have all the data from those three sources sent to the file sink.

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.

The classes you will encounter using 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.

We will now walk though a quick example, using the definition style.

The method create 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 as a single string for the stream definition, just like 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 has the methods getStatus, destroy and undeploy 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.

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 what 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 will create a simple stream using both approaches. A complete sample for you to get started can be found in the Spring Cloud Data Flow Samples Repository

The waitAndDestroy method uses the getStatus method to poll for the stream’s status.

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

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 the deployment properties are added using the method addDeploymentProperty, e.g. 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 to structure it is to have an @Configuration class define the StreamBuilder and StreamApplications.

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

This style allows you to easily 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 using the stream definitions

To create this stream enter the following command in the shell

Posting some data (using a shell command)

Will result in an uppercased 'HELLO' in the log

To demonstrate the data partitioning functionality, let’s deploy the following stream with Kafka as the binder.

You’ll see the following in the server logs.

Review the words.log instance 0 logs:

Review the words.log instance 1 logs:

This shows that payload splits that contain the same word are routed to the same application instance.

Let’s try something a bit more complicated and swap 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 above to

which will produce the following output from the server

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

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, e.g.:

and the stream will then funnel the data from the http source to the output log implemented by the log sink

Of course, 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 apps which are available. You can also define your own apps.

While Spring Cloud Task does provide a number of out of the box applications (via the spring-cloud-task-app-starters), most task applications will be custom developed. In order to create a custom task application:

Register a Task App with the App Registry 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". Here are a few examples:

When providing a URI with the maven scheme, the format should conform to 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, this would be a valid properties file:

Then use the app import command and provide the location of the properties file via --uri:

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 in previous paragraphs, 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.

List of available static property files:

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

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 will not be overridden by default. If you would like to override the pre-existing task app, then include the --force option.

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 via the restful API or the shell. To create a task definition using the shell, use the task create command to create the task definition. For example:

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

An adhoc task can be launched via the restful API or via the shell. To launch an ad-hoc task via the shell use the task launch command. For example:

When a task is launched, any properties that need to be passed as the 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 using a --properties option. Format of this option is a comma delimited string of properties prefixed with app.<task definition name>.<property>. Properties are passed to TaskLauncher as application properties and 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.

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 via the restful API or by the shell. To display the latest task executions via 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 display command with the id of the task execution, for example task display --id 549.

Destroying a Task Definition will remove the definition from the definition repository. This can be done via the restful API or via the shell. To destroy a task via the shell use the task destroy command. For example:

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

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

Conditional execution is expressed using a double ampersand symbol &&. This allows each task in the sequence to be launched only if the previous task successfully completed. For example:

When the composed task my-composed-task is launched, it will launch the task foo and if it completes successfully, then the task bar will be launched. If the foo task fails, then the task bar will 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. For example:

The diagram above is a screen capture of the directed graph as it being created using the Spring Cloud Data Flow Dashboard. We see that are 4 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 for 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 || . For example:

The example above will launch tasks foo, bar and baz in parallel.

Using the Spring Cloud Data Flow Dashboard to create the same "split execution" would look like:

With the task DSL a user may also execute multiple split groups in succession. For example:

In the example above tasks foo, bar and baz will be launched in parallel, once they all complete then tasks qux, quux will be launched in parallel. Once they complete the composed task will end. However if foo, bar, or baz fails then, the split containing qux and quux will not launch.

Using the Spring Cloud Data Flow Dashboard to create the same "split with multiple groups" would look like:

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

One way to launch a task using the task-launcher is to use the triggertask source. The triggertask source will emit a message with a TaskLaunchRequest object containing the required launch information. The triggertask can be added to the available sources by executing the app register command as follows (for the Rabbit Binder):

An example of this would be to launch the timestamp task once every 60 seconds, the stream to implement this would look like:

If you execute runtime apps you can find the log file for the task launcher sink. Tailing that file you can find the log file for the launched tasks. The setting of triggertask.environment-properties is so that all the task executions can be collected in the same H2 database used in the local version of the Data Flow Server. You can then see the list of task executions using the shell command task execution list

Another option to start a task using the task-launcher would be to create a stream 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):

For example:

A composed task can be launched using one of the task-launcher sinks as discussed here. Since we will be using the ComposedTaskRunner directly we will need to setup the task definitions it will use prior to the creation of the composed task launching stream. So let’s say that we wanted to create the following composed task definition AAA && BBB. The first step would be to create the task definitions. For example:

Now that the task definitions we need for composed task definition are ready, we need to create a stream that will launch ComposedTaskRunner. So in this case we will create a stream that has a trigger that will emit a message once every 30 seconds, a transformer that will create a TaskLaunchRequest for each message received, and a task-launcher-local sink that will launch a the ComposedTaskRunner on our local machine. The stream should look something like this:

In the example above we see that the tasklaunchrequest-transform is establishing 2 primary components:

For now let’s focus on the configuration that is required to launch the ComposedTaskRunner:

The bulk import applications page provides numerous options for defining and importing a set of applications in one go. For bulk import the application definitions are expected to be expressed in a properties style:

<type>.<name> = <coordinates>

For example:

task.timestamp=maven://org.springframework.cloud.task.app:timestamp-task:1.2.0.RELEASE

processor.transform=maven://org.springframework.cloud.stream.app:transform-processor-rabbit:1.2.0.RELEASE

At the top of the bulk import page an Uri can be specified that points to a properties file stored elsewhere, it should contain properties formatted as above. Alternatively, using the textbox labeled Apps as Properties it is possible to directly list each property string. Finally, if the properties are stored in a local file the Select Properties File option will open a local file browser to select the file. After setting your definitions via one of these routes, click Import.

At the bottom of the page there are quick links to the property files for common groups of stream apps and task apps. If those meet your needs, simply select your appropriate variant (rabbit, kafka, docker, etc) and click the Import action on those lines to immediately import all those applications.

Apps encapsulate a unit of work into a reusable component. Within the Data Flow runtime environment Apps allow users to create definitions for Streams as well as Tasks. Consequently, the Apps tab within the Tasks section allows users to create Task definitions.

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 using simple textual input, indicated by the bulk define tasks button.

This page lists the Batch Job Executions and provides the option to restart or stop a specific job execution, provided the operation is available. Furthermore, you have the option to view the Job execution details.

The list of Job Executions also shows the state of the underlying Job Definition. Thus, if the underlying definition has been deleted, deleted will be shown.

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 using the Data Flow Template the only needed Data Flow dependency is the Spring Cloud Data Flow Rest Client:

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

When instantiating the DataFlowTemplate, you will 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 will suffice:

Now you can instantiate the DataFlowTemplate with:

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

If you have custom Spring XD modules, you’d have to refactor them to use Spring Cloud Stream and Spring Cloud Task annotations, with updated dependencies and built 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, including Spring Batch jobs, and they end/stop at some point. Task applications can be majorly used for on-demand use-cases such as database migration, machine learning, scheduled operations etc. Using Spring Cloud Task, users can build Spring Batch jobs as microservice applications.

The Admin-UI is now renamed as 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, there are less peripherals that are necessary to operationalize Spring Cloud Data Flow.

To support centralized and consistent management of an application’s configuration properties, Spring Cloud Config client libraries have been included into 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/push it to the runtime platform (cloud foundry, apache yarn, kubernetes, or apache mesos). For example, if you’re running Spring Cloud Data Flow on Cloud Foundry, you’d download the Cloud Foundry server implementation and do a cf push as explained in the reference guide.

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

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

Let’s review some use-cases to compare and contrast the differences between Spring XD and Spring Cloud Data Flow.

There is a "full" profile that will generate documentation. You can build just the documentation by executing

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

If you don’t have an IDE preference we would recommend that you use Spring Tools Suite or Eclipse when working with the code. We use the m2eclipe eclipse plugin for maven support. Other IDEs and tools should also work without issue.

Before we accept a non-trivial patch or pull request we will 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 given the ability to merge pull requests.

None of these is essential for a pull request, but they will all help. They can also be added after the original pull request but before a merge.