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 1.2.0.RELEASE#_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 during stream deployment for additional strategies to partition streams during deployment and how they map onto the underlying 1.2.0.RELEASE#_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 1.2.0.RELEASE#_persistent_publish_subscribe_support[persistent publish-subscribe semantics].

The 1.2.0.RELEASE#_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 1.2.0.RELEASE#_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 1.2.0.RELEASE#_rabbitmq_consumer_properties[consumer properties] republishtoDlq and autoBindDlq and the 1.2.0.RELEASE#_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 1.2.0.RELEASE#_kafka_consumer_properties[Consumer] and 1.2.0.RELEASE#_kafka_producer_properties[Producer] and Rabbit 1.2.0.RELEASE#_rabbitmq_consumer_properties[Consumer] and 1.2.0.RELEASE#_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.

These requirements are not something yarn runtime needs but generally what dataflow core needs.

Download the Spring Cloud Data Flow YARN distribution ZIP file which includes the Server and the Shell apps:

Unzip the distribution ZIP file and change to the directory containing the deployment files.

Generic runtime settings can changed in config/servers.yml. Dedicated section Configuring Runtime Settings and Environment contains detailed information about configuration.

servers.yml file is a central place to share common configuration as it is added to Boot based jvm processes via option -Dspring.config.location=servers.yml.

If this is the first time deploying make sure the user that runs the Server app has rights to create and write to /dataflow directory in hdfs. If there is an existing deployment on hdfs remove it using:

Start the Spring Cloud Data Flow Server app for YARN

start spring-cloud-dataflow-shell

Shell in a distribution package contains extension commands for a hdfs file system.

By default, the application registry will be empty. If you would like to register all out-of-the-box stream applications built with the RabbitMQ binder in bulk, you can with the following command. For more details, review how to register applications.

Create a stream:

List streams:

After some time, destroy the stream:

The YARN application is pushed and started automatically during a stream deployment process. Once all streams are destroyed the YARN application will exit.

Create and launch task:

Launch tasks from streams:

task-launcher-yarn-sink itself bundles a YARN Deployer but doesn’t push any apps into hdfs, thus pushed app needs to exist and match a deployer version task-launcher-yarn-sink uses.

In below sample we use tasklaunchrequest processor to pass needed properties into task-launcher-yarn sink. We explicitely defined appVersion as appv1 which you would have pushed into hdfs prior running this stream. With this processor you also need to define a uri for a task application itself.

To fire up a task just post a dummy message into http source.

Overall app status can be seen from YARN Resource Manager UI or using Spring YARN CLI which gives more info about running containers within an app itself.

We package three different metrics collector implementations, one for RabbitMQ and two for different Kafka versions. There can be started using shell scripts, dataflow-server-metrics-collector-kafka-09, dataflow-server-metrics-collector-kafka-10 and dataflow-server-metrics-collector-rabbit respectively. These applications are not using servers.yml file for config, instead collectors.yml is used where custom settings can be placed.

Generally it is only needed to install scdf-plugin-hdp plugin into ambari server which adds needed service definitions.

When you create your cluster and choose a stack, make sure that redhat6 or/and redhat7 sections contains repository named SCDF-1.2.0 and that it points to repo.spring.io/yum-release-local/scdf/1.2.0.

Ambari 2.4 contains major rewrites for stack definitions and how it is possible to integrate with those from external contributions. Our plugin will eventually integrate via extensions or management packs, but for now you need to choose stack marked as a Default Version Definition which contains correct yum repository. For example with HDP 2.5 you have two default choices, HDP-2.5.0.0 and HDP-2.5 (Default Version Definition). As mentioned you need to pick latter. With older ambari versions you don’t have these new options.

From services choose Spring Cloud Data Flow and Kafka. Hdfs, Yarn and Zookeeper are forced dependencies.

Then in Customize Services what is really left for user to do is to customise settings if needed. Everything else is automatically configured. Technically it also allows you to switch to use rabbit by leaving Kafka out and defining rabbit settings there. But generally use of Kafka is a good choice.

servers.yml file is also used to store common configuration with Ambari. Settings in Advanced scdf-site and Custom scdf-site are used to dynamically create a this file which is then copied over to hdfs when needed application files are deployd.

Every additional entry added via Custom scdf-site is added into servers.yml as is and overrides everything else in it.

All applications whether those are stream apps or task apps can be centrally configured with servers.yml as that file is passed to apps using --spring.config.location='servers.yml'.

Stream and task processes for application master and containers can be further tuned by setting memory and cpu settings. Also java options allow to define actual jvm options.

Base directory where all needed files are kept defaults to /dataflow and can be changed using baseDir property.

Spring Cloud Data Flow app registration is based on URI’s with various different endpoints. As mentioned in section How YARN Deployment Works all applications are first stored into hdfs before application container is launched. Server can use http, file, http and maven based uris as well direct hdfs uris.

It is possible to place these applications directly into HDFS and register application based on that URI.

Logging for all components is done centrally via servers.yml file using normal Spring Boot properties.

If metrics are enabled, needed settings are written into servers.yml files used by applications. Also specific settings are written into collectors.yml used by SCDF Metrics Collector service. You need to choose a correct collector type, its service port and output channel name.

YARN Nodemanager is continously tracking how much memory is used by individual YARN containers. If containers are using more memory than what the configuration allows, containers are simply killed by a Nodemanager. Application master controlling the app lifecycle is given a little more freedom meaning that Nodemanager is not that aggressive when making a desicion when a container should be killed.

Lets take a quick look of memory related settings in YARN cluster and in YARN applications. Below xml config is what a default vanilla Apache Hadoop uses for memory related settings. Other distributions may have different defaults.

Enabling kerberos is relatively easy when existing kerberized cluster exists. Just like with every other hadoop related service, use a specific user and a keytab.

Generic settings for dataflow components to work with HA setup can be seen below where id is set to mycluster.

On default a dataflow server will start embedded H2 database using in-memory storage and effectively using configuration.

Distribution package contains a bundled self-contained H2 executable which can be used instead. This allows to persist data throughout server restarts and is not limited to single host.

This bundled H2 database is also used in ambari to have a default out of a box functionality. Any database supported by a dataflow itself can be used by changing datasource settings.

YARN Deployer has to be able to talk with Application Master which then is responsible controlling containers running stream and task applications. The way this work is that Application Master tries to discover its own address which YARN Deployer is then able to use. If YARN cluster nodes have multiple NICs or for some other reason address is discovered wrongly, some settings can be changed to alter default discovery logic.

Below is a generic settings what can be changed.

Install plugin repository.

Install plugin.

Add needed services together spring Spring Cloud Data Flow. Tune server jvm options. Spring Cloud Data Flow → Configs → Advanced scdf-server-env → scdf-server-env template:

Tune jvm options for application masters and container. Spring Cloud Data Flow → Configs → Custom scdf-site:

Stream and Task applications are Spring Boot applications which are aware of many Common application properties, e.g. server.port but also families of properties such as those with the prefix spring.jmx and logging. When creating your own application it is desirable to whitelist properties so that the shell and the UI can display them first as primary properties when presenting options via TAB completion or in drop-down boxes.

To whitelist application properties create a file named spring-configuration-metadata-whitelist.properties in the META-INF resource directory. There are two property keys that can be used inside this file. The first key is named configuration-properties.classes. The value is a comma separated list of fully qualified @ConfigurationProperty class names. The second key is configuration-properties.names whose value is a comma separated list of property names. This can contain the full name of property, such as server.port or a partial name to whitelist a category of property names, e.g. spring.jmx.

The Spring Cloud Stream application starters are a good place to look for examples of usage. Here is a simple example of the file sink’s spring-configuration-metadata-whitelist.properties file

If we also wanted to add server.port to be white listed, then it would look like this:

You can go a step further in the process of describing the main properties that your stream or task app supports by creating a so-called metadata companion artifact. This simple jar file contains only the Spring boot JSON file about configuration properties metadata, as well as the whitelisting file described in the previous section.

Here is the contents of such an artifact, for the canonical log sink:

Note that the spring-configuration-metadata.json file is quite large. This is because it contains the concatenation of all the properties that are available at runtime to the log sink (some of them come from spring-boot-actuator.jar, some of them come from spring-boot-autoconfigure.jar, even some more from spring-cloud-starter-stream-sink-log.jar, etc.) Data Flow always relies on all those properties, even when a companion artifact is not available, but here all have been merged into a single file.

To help with that (as a matter of fact, you don’t want to try to craft this giant JSON file by hand), you can use the following plugin in your build:

The benefits of a companion artifact are manifold:

Remember though, that this is entirely optional when dealing with uberjars. The uberjar itself also includes the metadata in it already.

Application properties are the properties associated with each application in the stream. When the application is deployed, the application properties are applied to the application via command line arguments or environment variables based on the underlying deployment implementation.

When deploying the stream, properties that control the deployment of the apps into the target platform are known as deployment properties. For instance, one can specify how many instances need to be deployed for the specific application defined in the stream using the deployment property called count.

In addition to configuration via DSL, Spring Cloud Data Flow provides a mechanism for setting common properties to all the streaming applications that are launched by it. This can be done by adding properties prefixed with spring.cloud.dataflow.applicationProperties.stream when starting the server. When doing so, the server will pass all the properties, without the prefix, to the instances it launches.

For example, all the launched applications can be configured to use a specific Kafka broker by launching the Data Flow server with the following options:

This will cause the properties spring.cloud.stream.kafka.binder.brokers and spring.cloud.stream.kafka.binder.zkNodes to be passed to all the launched applications.

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.

To add a driver for the database that will store the Task Execution information, a dependency for the driver will need to be added to a maven pom file and the Spring Cloud Data Flow will need to be rebuilt. Since Spring Cloud Data Flow is comprised of an SPI for each environment it supports, please review the SPI’s documentation on which POM should be updated to add the dependency and how to build. This document will cover how to setup the dependency for local SPI.

To configure the datasource Add the following properties to the dataflow-server.yml or via environment variables:

For example adding postgres would look something like this:

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:

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

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 builds upon Spring Cloud Deployer SPI and the platform specific dataflow-server uses the respective SPI implementations. Specifically, if we were to troubleshoot deployment specific issues; such as the network errors, it’d be useful to enable the DEBUG logs at the underlying deployer and the libraries used by it.

The streaming applications in Spring Cloud Data Flow are Spring Boot applications and they can be independently setup with logging configurations.

For instance, if you’d have to troubleshoot the header and payload specifics that are being passed around source, processor and sink channels, you’d be deploying the stream with the following options.

(where, org.springframework.integration is the global package for everything Spring Integration related, which is responsible for messaging channels)

These properties can also be specified via deployment properties when deploying the stream.

One of the powerful features of SpEL expressions is functions. Spring Integration provides jsonPath() and xpath() out-of-the-box SpEL-functions, if appropriate libraries are in the classpath. All the provided Spring Cloud Stream application starters are supplied with the json-path and spring-integration-xml jars, thus we can use those SpEL-functions in Spring Cloud Data Flow streams whenever expressions are possible. For example we can transform JSON-aware payload from the HTTP request using some jsonPath() expression:

In this sample we apply jsonPath for the incoming payload to extract just only the price field value. Similar syntax can be used with splitter or filter expression options. Actually any available SpEL-based option has access to the built-in SpEL-functions. For example we can extract some value from JSON data to calculate the partitionKey before sending output to the Binder:

The same syntax can be applied for xpath() SpEL-function when you deal with XML data. Any other custom SpEL-function can also be used, but for this purpose you should build a library with the @Configuration class containing an appropriate SpelFunctionFactoryBean @Bean definition. The target Spring Cloud Stream application starter should be re-packaged to supply such a custom extension via built-in Spring Boot @ComponentScan mechanism or auto-configuration hook.

The JDBC-sink can be used to insert message payload data into a relational database table. By default, it inserts the entire payload into a table named after the jdbc.table-name property, and if it is not set, by default the application expects to use a table with the name messages. To alter this behavior, the JDBC sink accepts several options that you can pass using the --foo=bar notation in the stream, or change globally. The JDBC sink has a jdbc.initialize property that if set to true will result in the sink creating a table based on the specified configuration when the it starts up. If that initialize property is false, which is the default, you will have to make sure that the table to use is already available.

A stream definition using jdbc sink relying on all defaults with MySQL as the backing database looks like the following. In this example, the system time is persisted in MySQL for every second.

For this to work, you’d have to have the following table in the MySQL database.

For situations where the data is consumed and processed between two different message brokers, Spring Cloud Data Flow provides easy to override global configurations, out-of-the-box bridge-processor, and DSL primitives to build these type of topologies.

Let’s assume we have data queueing up in RabbitMQ (e.g., queue = fooRabbit) and the requirement is to consume all the payloads and publish them to Apache Kafka (e.g., topic = barKafka), as the destination for downstream processing.

Follow the global application of configurations to define multiple binder configurations.

These properties can be supplied in a ".properties" file that is accessible to the server directly or via config-server.

Spring Cloud Data Flow internally uses bridge-processor to directly connect different named channel destinations. Since we are publishing and subscribing from two different messaging systems, you’d have to build the bridge-processor with both RabbitMQ and Apache Kafka binders in the classpath. To do that, head over to start-scs.cfapps.io/ and select Bridge Processor, Kafka binder starter, and Rabbit binder starter as the dependencies and follow the patching procedure described in the reference guide. Specifically, for the bridge-processor, you’d have to import the BridgeProcessorConfiguration provided by the starter.

Once you have the necessary adjustments, you can build the application. Let’s register the name of the application as multiBinderBridge.

It is time to create a stream definition with the newly registered processor application.

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-yarn-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.