The Spring XD distributed runtime (DIRT) supports distribution of processing tasks across multiple nodes. See Getting Started for information on running Spring XD as a single node.

The XD distributed runtime architecture consists of the following distributed components:

In addition, XD provides a Command Line Interface (CLI), XD Shell as well as a web application, XD-UI to interact with the XD runtime.

The distributed runtime requires an RDBMS. The XD distrubution comes with an HSQLDB in memory database for testing purposes, but an alternate is expected. To start HSQLDB:

To configure XD to connect to a different RDBMS, have a look at xd/config/servers.yml in the spring:datasource section for details. Note that spring.batch.initializer.enabled is set to true by default which will initialize the Spring Batch schema if it is not already set up. However, if those tables have already been created, they will be unaffected.

If the provided schemas are customized, other values may need to be customized. In the xd/config/servers.yml the following block exposes database specific values for the batch job repository.

Currently XD does not ship with ZooKeeper. At the time of this writing, the compliant version is 3.4.6 and you can download it from here. Please refer to the ZooKeeper Getting Started Guide for more information. A ZooKeeper ensemble consisting of at least three members is recommended for production installations, but a single server is all that is needed to have XD up and running.

You can configure the root path in Zookeeper where an XD cluster’s top level nodes will be created. This allows you to run multiple independent clusters of XD that share a single ZK instance. Add the following to servers.yml to configure. You can also set as an environment variable, system property in the standard manner.

Additionally, various time related settings may be optionally configured for ZooKeeper:

Redis is the default transport when running in distributed mode.

Spring XD consists of two servers

You can start the xd-container and xd-admin servers individually as follows:

Spring XD supports the following Hadoop distributions:

To specify the distribution libraries to use for Hadoop client connections, use the option
--hadoopDistro for the xd-container and xd-shell commands:

Pass in the --help option to see other configuration properties.

If you wish to use a XD-Shell that is on a different machine than where you deployed your admin server.

1) Open your shell

2) From the xd shell use the "admin config server" command i.e.

The Spring XD distributed runtime (DIRT) supports distribution of processing tasks across multiple nodes. See Running Distributed Mode for information on running Spring XD in distributed mode. One option is to run these nodes on a Hadoop YARN cluster rather than on VMs or physical servers managed by you.

To begin with, you need to have access to a Hadoop cluster running a version based on Apache Hadoop version 2. This includes Apache Hadoop 2.6.0, Pivotal HD 2.1, Hortonworks HDP 2.2 and Cloudera CDH5.

You need a supported transport, see Running Distributed Mode for installation of Redis or Rabbit MQ. Spring XD on YARN currently uses Redis as the default data transport.

You also need Zookeeper running. If your Hadoop cluster doesn’t have Zookeeper installed you need to install and run it specifically for Spring XD. See the Setting up ZooKeeper section of the "Running Distributed Mode" chapter.

Lastly, you need an RDBMs to support batch jobs and JDBC operations.

In addition to the regular spring-xd-<version>-dist.zip files we also distribute a zip file that includes all you need to deploy on YARN. The name of this zip file is spring-xd-<version>-yarn.zip. You can download the zip file for the current release from Spring release repo or a milestone build from the Spring milestone repo. Unzip the downloaded file and you should see a spring-xd-<version>-yarn directory.

Configuration options are contained in a config/servers.yml file in the Spring XD YARN install directory. You need to configure the hadoop settings, the transport choice plus redis/rabbit settings, the zookeeper settings and the JDBC datasource properties.

Depending on the distribution used you might need to change the siteYarnAppClasspath and siteMapreduceAppClasspath. We have provided basic settings for the supported distros, you just need to uncomment the ones for the distro you use.

These are the settings used for Hadoop 2.6.0:

The configurations for all modules can be customized by modifying the file modules.yml in the config directory and then adding it to the modules-config.zip archive in the same directory.

You can run the following command from the config directory to achieve this:

There is an empty archive file named custom-modules.zip. You can replace this with your own ZIP archive with the same name and it will be uploaded as part of the deployment to YARN. Place custom module definitions in a modules directory inside this new ZIP archive. Module definitions must follow the Spring XD module semantics.

Logging configuration for XD admins and containers are defined in files config/xd-admin-logger.properties and config/xd-container-logger.properties respectively. These two files are copied over to hdfs during the deployment. If you want to modify logging configuration either modify source files and do a deployment again or modify files in hdfs directly.

Change current directory to be the directory that was unzipped spring-xd-<version>-yarn. To read about runtime configuration and more sophisticated features see section Working with container groups.

XD admins will register its runtime information into zookeeper and you can use the admininfo command to query this information:

Then connect xd-shell to this instance:

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.

We have three memory settings for components participating XD YARN runtime. You can use configuration properties spring.xd.appmasterMemory, spring.xd.adminMemory and spring.xd.containerMemory respectively.

Container grouping and clustering is more sophisticated feature which allows better control of XD admins and containers at runtime. Basic features are:

XD YARN Runtime has a few built-in groups to get you started. There are two groups admin and container created by default which both are lauching exactly one container chosen randomly from YARN cluster.

This section contains examples of usual use cases for custom configurations.

This section describes some background of how YARN resource allocation works, what are the limitations of it and more importantly how it reflects into XD YARN runtime.

YARN as having a strong roots from original MapReduce framework is imposing relatively strange concepts of where containers are about to be executed. In a MapReduce world every map and reduce tasks are executed in its own container where colocation is usually determined by a physical location of a HDFS file block map or reduce tasks are accessing. This is introducing a concepts of allocating containers on any hosts, specific hosts or specific racks. Usually YARN is trying to place container as close as possible to a physical location to minimize network IO so i.e. if host cannot be chosen, rack is chosen instead assuming a whole rack is connected together with a fast switch.

For custom YARN applications like XD YARN runtime this doesn’t necessarily make that much sense because we’re not hard-tied to HDFS file blocks. What makes sense is that we can still place containers on different racks to get better high availability in case whole rack goes down or if specific containers needs to exist on specific hosts to access either custom physical or network resources. Good example of having a need to execute something on a specific host is either a disk access or outbound internet access if cluster is highly secured.

One other YARN resource allocation concept worth mentioning is relaxation of container locality. This simply means that if resources are requested from hosts or racks, YARN will relax those requests if resources cannot be allocated immediately. Turning relax flag off guarantees that containers will be allocated from hosts or racks. Though these requests will then wait forever if allocation cannot be done.

Profiles provide a way to segregate parts of your application configuration and change their availability and/or values based on the environment. This lets you have different configuration settings for qa and prod environments and to easily switch between them.

To activate a profile, set the OS environment variable SPRING_PROFILES_ACTIVE to a comma delimited list of profile names. The server looks to load profile specific variants of the servers.yml file based on the naming convention servers-{profile}.yml. For example, if SPRING_PROFILES_ACTIVE=prod the following files would be searched for in the following order.

You may also put multiple profile specific configuration in a single servers.yml file by using the key spring.profiles in different sections of the configuration file. See Multi-profile YAML documents for more information.

Spring XD saves the state of the batch job workflows in a relational database. When running xd-singlenode a embedded HSQLDB database is run. When running in distributed mode a standalone HSQLDB instance can be used, the startup script hsqldb-server is in is provided the installation directory under the folder hsqldb/bin. It is recommended to use HSQLDB only for development and learning.

When deploying in a production environment, you will need to select another database. Spring XD is actively tested on MySql (Version: 5.1.23) and Postgres (Version 9.2-1002). All batch workflow tables are automatically created, if they do not exist, for HSQLDB, MySQL and Postgres. The JDBC driver jars for the HSQLDB and Postgres are already on the XD classpath.

The provided configuration file servers.yml located in $XD_HOME\config has commented out configuration for some commonly used databases. You can use these as a basis to support your database environment. XD also utilizes the Tomcat jdbc connection pool and these settings can be configured in the servers.yml.

If you want to use Redis for analytics or data transport you should set the host and port of the Redis server.

To override these settings set an OS environment variable such as spring_redis_port to the value you require.

You can also configure redis to use Sentinel.

In addition, the following default settings for the rabbit message bus can be modified in servers.yml…​

If you want to use RabbitMQ as a data transport use the following configuration setting

To override these settings set an OS environment variable such as spring_rabbitmq_host to the value you require.

See Message Bus regarding SSL configuration.

In addition, the following default settings for the rabbit message bus can be modified in servers.yml…​

If you want to use Kafka as a data transport, the following connection settings, as well as defaults for the kafka message bus can be modified in servers.yml.

The default HTTP port of the xd-admin server is 9393. To change the value use the following configuration setting

The XD servers provide general health and JMX exported management endpoints via Jolokia.

By default the management and health endpoints are available on port 9393. To change the value of the port use the following configuration setting to servers.yml.

You can also disable http management endpoints by setting the port value to -1.

By default JMX MBeans are exported. You can disable JMX by setting spring.jmx.enabled=false.

The section on Monitoring and management over HTTP provides details on how to configure these endpoint.

By default, the Spring XD admin server is unsecured and runs on an unencrypted HTTP connection. You can secure your administration REST endpoints, as well as the Admin UI by enabling HTTPS and requiring clients to authenticate.

Local transport uses a QueueChannel to pass data between modules. There are a few properties you can configure on the QueueChannel

When resolving property file names, the server will look to load profile specific variants based on the naming convention <name>-{profile}.properties. For example, if given the OS environment variable spring_profiles_active=default,qa the following configuration file names for the twittersearch module would be searched in this order

Also, the shared module configuration file is refernced using profile variants, so given the OS environment variable spring_profiles_active=default,qa the following shared module configuration files would be searched for in this order

Another common case is access to a relational database from a job or the JDBC Sink module.

As an example, to provide the properties for the batch job jdbchdfs the file XD_MODULE_CONFIG_LOCATION\job\jdbchdfs\jdbchdfs.properites should contain

A property file with the same keys, but likely different values would be located in XD_MODULE_CONFIG_LOCATION\sink\jdbc\jdbc.properites.

Arguably, the most complex component when it comes to quotes is the shell. The rules can be laid out quite simply, though:

For example, the XD shell supports the ! command to execute native shell commands. The ! accepts a single, key-less argument. This is why the following works:

The argument here is the whole rm foo string, which is passed as is to the underlying shell.

As another example, the following commands are strictly equivalent, and the argument value is foo (without the quotes):

At the XD parser level (that is, inside the body of a stream or job definition) the rules are the following:

As such, the values of the --expression option to the filter module are semantically equivalent in the following examples:

Arguably, the last one is more readable. It is made possible thanks to the surrounding quotes. The actual expression is payload > 5 (without quotes).

Now, let’s imagine we want to test against string messages. If we’d like to compare the payload to the SpEL literal string, "foo", this is how we could do:

Please note that the examples above are to be considered outside of the Spring XD shell. When entered inside the shell, chances are that the whole stream definition will itself be inside double quotes, which would need escaping. The whole example then becomes:

The last piece of the puzzle is about SpEL expressions. Many modules accept options that are to be interpreted as SpEL expressions, and as seen above, String literals are handled in a special way there too. Basically,

As a last example, assume you want to use the transform module. That module accepts an expression option which is a SpEL expression. It is to be evaluated against the incoming message, with a default of payload (which forwards the message payload untouched).

It is important to understand that the following are equivalent:

but very different from the following:

and other variations.

The first series will simply evaluate to the message payload, while the latter examples will evaluate to the actual literal string payload (again, without quotes).

As a last, complete example, let’s review how one could force the transformation of all messages to the string literal hello world, by creating a stream in the context of the XD shell:

When you start the shell you can type help to show all the commands that are available. Note, that since the XD shell is based on Spring Shell you can contribute you own commands into the shell. The general groups of commands are related to the management of

For example to see what modules are available issue the command

The shell also provides extensive command completion capabilities. For example, if you type mod and hit TAB, you will be presented with all the matching commands.

The command module list shows all the modules available

Suppose we want to create a stream that uses the http source and file sink. How do we know what options are available to use? There are two ways to find out. The first is to use the command module info. Pressing TAB after typiing moudle info will complete the command with the --name option and then present all the modules prefixed by their type.

The module info command for the http source shows the option names, a brief description, and default values.

For the file sink the options are

Note that the default value ${xd.stream.name} will be resolved to the name of the stream that contains the module.

When creating a stream defintion tab completion after --definition will enable you to see all the options that are available for a given module as well as a list of candidate modules for the subsequent module in the stream. For example, hitting TAB after http as shown below

shows the options outputType and port in addition to any processors and sinks. Hitting TAB after entering -- after the http module will provide a list of only the http options

Entering the port number and also the pipel | symbol and hitting tab will show completions for candidate processor and sink modules. The same process of tab completion for module options applies to each module in the chain.

You can execute a script by either passing in the --cmdfile argument when starting the shell or by executing the script command inside the shell. When using scripts it is common to add comments using either // or ; characters at the start of the line for one line comments or use /* and */ for multiline comments

There are often three layers of parsing when passing entering commands to the shell. The shell parses the command to recognize -- options, inside the body of a stream/job definition the values are parsed until the first space character, and inside some command options SpEL is used (e.g. router). Understanding the interaction between these layers can cause some confusion. The DSL Guide section on quotes and escaping will help you if you run into any issues.

Modules encapsulate a unit of work into a reusable component. Within the XD runtime environment Modules allow users to create definitions for Streams as well as Batch Jobs. Consequently, the Modules tab within the Jobs section allows users to create Batch Job definitions. In order to learn more about Modules, please see the chapter on Modules.

This page lists the XD batch job definitions and provides actions to deploy, un-deploy or destroy those jobs.

This page lists all the deployed jobs and provides option to launch or schedule the deployed job.

On this screen, you can view additional deployment details. Besides viewing the stream definition, the available Module Metadata is shown as well, e.g. on which Container the definition has been deployed to.

This page lists the batch job executions and provides option to restart specific job executions, provided the batch job is restartable and stopped/failed.

Furthermore, you have the option to view the Job execution details.

The key components in Spring XD are the XD Admin and XD Container Servers. Using a high-level DSL, you post the description of the required processing tasks to the Admin server over HTTP. The Admin server then maps the processing tasks into processing modules. A module is a unit of execution and is implemented as a Spring ApplicationContext. A distributed runtime is provided that will assign modules to execute across multiple XD Container servers. A single XD Container server can run multiple modules. When using the single node runtime, all modules are run in a single XD Container and the XD Admin server is run in the same process.

A single node runtime is provided that runs the Admin and Container servers, ZooKeeper, and HSQLDB in the same process. the single node runtime is primarily intended for testing and development purposes but it may also appropriate to use in small production use-cases. The communication to the XD Admin server is over HTTP and the XD Admin server communicates to an in-process XD Container using an embedded ZooKeeper server.

The Admin Server uses an embedded servlet container and exposes REST endpoints for creating, deploying, undeploying, and destroying streams and jobs, querying runtime state, analytics, and the like. The Admin Server is implemented using Spring’s MVC framework and the Spring HATEOAS library to create REST representations that follow the HATEOAS principle. The Admin Server and Container Servers monitor and update runtime state using ZooKeeper (see XD Distributed Runtime).

The key components of data processing in Spring XD are

Streams define how event driven data is collected, processed, and stored or forwarded. For example, a stream might collect syslog data, filter, and store it in HDFS.

Jobs define how coarse grained and time consuming batch processing steps are orchestrated, for example a job could be be defined to coordinate performing HDFS operations and the subsequent execution of multiple MapReduce processing tasks.

Taps are used to process data in a non-invasive way as data is being processed by a Stream or a Job. Much like wiretaps used on telephones, a Tap on a Stream lets you consume data at any point along the Stream’s processing pipeline. The behavior of the original stream is unaffected by the presence of the Tap.

The programming model for processing event streams in Spring XD is based on the well known Enterprise Integration Patterns as implemented by components in the Spring Integration project. The programming model was designed so that it is easy to test components.

A Stream consist of the following types of modules: * An Input source * Processing steps * An Output sink

An Input source produces messages from an external source. XD supports a variety of sources, e.g. syslog, tcp, http. The output from a module is a Spring Message containing a payload of data and a collection of key-value headers. Messages flow through message channels from the source, through optional processing steps, to the output sink. The output sink delivers the message to an external resource. For example, it is common to write the message to a file system, such as HDFS, but you may also configure the sink to forward the message over tcp, http, or another type of middleware, or route the message to another stream.

A stream that consists of a input source and a output sink is shown below

A stream that incorporates processing steps is shown below

For simple linear processing streams, an analogy can be made with the UNIX pipes and filters model. Filters represent any component that produces, processes or consumes events. This corresponds to the modules (source, processing steps, and sink) in a stream. Pipes represent the way data is transported between the Filters. This corresponds to the Message Channel that moves data through a stream.

A simple stream definition using UNIX pipes and filters syntax that takes data sent via a HTTP post and writes it to a file (with no processing done in between) can be expressed as

The pipe symbol represents a message channel that passes data from the HTTP source to the File sink. The message channel implementation can either be backed with a local in-memory transport, Redis queues, or RabbitMQ. The message channel abstraction and the XD architecture are designed to support a pluggable data transport. Future releases will support other transports such as JMS.

Note that the UNIX pipes and filter syntax is the basis for the DSL that Spring XD uses to describe simple linear flows. Non-linear processing is partially supported using named channels which can be combined with a router sink to effectively split a single stream into multiple streams (see Dynamic Router Sink). Additional capabilities for non-linear processing are planned for future releases.

The programming model for processing steps in a stream originates from the Spring Integration project and is included in the core Spring Framework as of version 4. The central concept is one of a Message Handler class, which relies on simple coding conventions to Map incoming messages to processing methods. For example, using an http source you can process the body of an HTTP POST request using the following class

The payload of the incoming Message is passed as a string to the method process. The contents of the payload is the body of the http request as we are using a http source. The non-void return value is used as the payload of the Message passed to the next step. These programming conventions make it very easy to test your Processor component in isolation. There are several processing components provided in Spring XD that do not require you to write any code, such as a filter and transformer that use the Spring Expression Language or Groovy. For example, adding a processing step, such as a transformer, in a stream processing definition can be as simple as

For more information on processing modules, refer to the Processors section.

The Container Server listens for module deployment events initiated from the Admin Server via ZooKeeper. When the container node handles a module deployment event, it connects the module’s input and output channels to the data bus used to transport messages during stream processing. In a single node configuration, the data bus uses in-memory direct channels. In a distributed configuration, the data bus communications are backed by the configured transport middleware. Redis and Rabbit are both provided with the Spring XD distribution, but other transports are envisioned for future releases.

In the http | file example, the Admin assigns each module to a separate Container instance, provided there are at least two Containers available. The file module is deployed to one container and the http module to another. The definition of a module is stored in a Module Registry. A module definition consists of a Spring XML configuration file, some classes used to validate and handle options defined by the module, and dependent jars. The module definition contains variable placeholders, corresponding to DSL parameters (called options) that allow you to customize the behavior of the module. For example, setting the http listening port would be done by passing in the option --port, e.g. http --port=8090 | file, which is in turn used to substitute a placeholder value in the module definition.

The Module Registry is backed by the filesystem and corresponds to the directory <xd-install-directory>/modules. When a module deployment is handled by the Container, the module definition is loaded from the registry and a new Spring ApplicationContext is created in the Container process to run the module. Dependent classes are loaded via the Module Classloader which first looks at jars in the modules /lib directory before delegating to the parent classloader.

Using the DIRT runtime, the http | file example would map onto the following runtime architecture

Data produced by the HTTP module is sent over a Redis Queue and is consumed by the File module. If there was a filter processing module in the stream definition, e.g http | filter | file that would map onto the following DIRT runtime architecture.

Let’s walk through the simple example above. If you don’t have a Spring XD cluster set up, this example can be easily executed running Spring XD in a single node configuration. The single node application includes an embedded ZooKeeper server by default and allocates a random unused port. The embedded ZooKeeper connect string is reported in the console log for the single node application:

For our purposes, we will use the ZooKeeper CLI tool to inspect the contents of ZooKeeper nodes reflecting the current state of Spring XD. First, we need to know the port to connect the CLI tool to the embedded server. For convenience, we will assign the ZooKeeper port (5555 in this example) when starting the single node application. From the XD install directory:

In another terminal session, start the ZooKeeper CLI included with ZooKeeper to connect to the embedded server and inspect the contents of the nodes (NOTE: tab completion works) :

After some console output, you should see a prompt:

navigate using the ls command. This will reflect the schema shown in the figure above, the unique container ID will be different for you.

The above reflects the initial state of Spring XD with a running admin and container instance. Nothing is deployed yet and there are no existing stream or job definitions. Note that xd/deployments/modules/allocated has a persistent child corresponding to the id of the container at xd/containers. If you are running in a distributed configuration and connected to one of the ZooKeeper servers in the same ensemble that Spring XD is connected to, you might see multiple nodes under /xd/containers, and xd/admins. Because the external ensemble persists the state of the Spring XD cluster, you will also see any deployments that existed when the Spring XD cluster was shut down.

Start the XD Shell in a new terminal session and create a stream:

Back to the ZK CLI session:

using the get command on the new stream node, we can see the stream definition represented as JSON, along with some standard ZooKeeper metadata.

Now, Using the Spring XD shell, let’s deploy the stream,

and verify with ZooKeeper:

Note the deployment state shown for the stream’s status node is deployed, meaning the deployment request was satisfied. Deployment states are discussed in more detail here.

Spring XD decomposes stream deployment requests to individual module deployment requests. Hence, we see that each module in the stream is associated with a container instance. The container instance in this case is the same since there is only one instance in the single node configuration. In a distributed configuration with more than one instance, the stream source and sink will each be deployed to a separate container. The node name itself is of the form <module_type>.<module_name>.<module_sequence_number>.<container_id>, where the sequence number identifies a deployed instance of a module if multiple instances of that module are requested.

The metadata and status nodes are ephemeral nodes which store details about the deployed module. This information is provided to XD shell queries. For example:

For this example we start two container instances and deploy and simple stream:

Now we will kill one of the container processes and observe that the affect module has been redeployed to the remaining container:

Now if we kill the remaining container, we see warnings in the xd-admin log:

Register a Job Module with the Module Registry by putting XML and/or jar files into the $XD_HOME/modules/jobs directory.

Create a Job Definition from a Job Module by providing a definition name as well as properties that apply to all Job Instances. At this point the job is not deployed, yet.

Deploy the Job Definition to one or more Spring XD containers. This will initialize the Job Definitions on those containers. The jobs are now "live" and a job can be created by sending a message to a job queue that contains optional runtime Job Parameters.

Launch a job by sending a message to the job queue with Job Parameters. A Job Instance is created, representing a specific run of the job. A Job Instance is the Job Definition plus the runtime Job Parameters. You can query for the Job Instances associated with a given job name.

The job is executed creating a Job Execution object that captures the success or failure of the job. You can query for Job Executions associated with a given job name.

This removes the job from the Spring XD container(s) preventing the launching of any new Job Instances. For reporting purposes, you will still be able to view historic Job Executions associated with the the job.

Destroying a Job Definition will not only un-deploy any still deployed Job Definitions but will also remove the Job Definition itself.

When creating jobs, the following options are available to all job definitions:

Also, similar to the stream create command, the job create command has an optional --deploy option to create the job definition and deploy it. --deploy option is false by default.

Below is an example of some of these options combined:

Remember that you can always find out about available options for a job by using the module info command.

To launch a job one time, use the launch option of the job command. So going back to our example above, we’ve created a job module instance named helloSpringXD. Launching that Job Module Instance would look like:

In the logging output of the XDContainer you should see the following

To re-launch the job just execute the launch command. For example:

To launch a batch job based on a cron scheduler is done by creating a stream using the trigger source.

A batch job can receive parameters from a source (in this case a trigger) or process. A trigger uses the --payload expression to declare its payload.

To pause/stop future scheduled jobs from running for this stream, the stream must be undeployed for example:

A fixed-delay-trigger is used to launch a Job on a regular interval. Using the --fixedDelay parameter you can set up the number of seconds between executions. In the example below we are running myXDJob every 10 seconds and passing it a payload containing a single attribute.

To pause/stop future scheduled jobs from running for this stream, you must undeploy the stream for example:

A batch job is always used as a sink, with that being said it can receive messages from sources (other than triggers) and processors. In the case below we see that the user has created an http source (http source receives http posts and passes the payload of the http message to the next module in the stream) that will pass the http payload to the "myHttpJob".

To test the stream you can execute a http post, like the following:

The stream receives aggregated event messages from all the default batch job listeners and sends those messages to the log.

Note: The syntax for the tap that receives the aggregated events is: tap:job:<job-name>

In the logging output of the container you should see something like the following when the job completes (with the aggregated events

Note: The syntax for the tap that receives the job execution events is: tap:job:<job-name>.job

In the logging output of the container you should see something like the following when the job completes

Note: The syntax for the tap that receives the step execution events is: tap:job:<job-name>.step

In the logging output of the container you should see something like the following when the job completes

Note: The syntax for the tap that receives the item events: tap:job:<job-name>.item,for skip events: tap:job:<job-name>.skip and for chunk events: tap:job:<job-name>.chunk

To select specific listeners, specify comma separated list in --listeners option. Following example illustrates the selection of job and step execution listeners only:

Note: List of options are: job, step, item, chunk and skip The aggregated channel is registered if at least one of these default listeners are enabled.

For a complete example, please see the Batch Notifications Sample which is part of the Spring XD Samples repository.

To use the hdfs based jobs below, XD needs to have append enabled for hdfs. Update the hdfs-site.xml with the following settings:

This module is designed to be driven by a stream polling a directory. It imports data from CSV files and requires that you supply a list of named columns for the data using the names parameter. For example:

You would then use a stream with a file source to scan a directory for files and drive the job. A separate job will be started for each file found:

The filepollhdfs job has the following options:

A module which loads CSV files into a JDBC table using a single batch job. By default it uses the internal HSQL DB which is used by Spring Batch. Refer to how module options are resolved for further details on how to change defaults (one can of course always use --foo=bar notation in the job definition to achieve the same effect).

The filejdbc job has the following options:

The job should be defined with the resources parameter defining the files which should be loaded. It also requires a names parameter (for the CSV field names) and these should match the database column names into which the data should be stored. You can either pre-create the database table or the module will create it for you if you use --initializeDatabase=true when the job is created. The table initialization is configured in a similar way to the JDBC sink and uses the same parameters. The default table name is the job name and can be customized by setting the tableName parameter. As an example, if you run the command

it will create the table "people" in the database with three varchar columns called "forename", "surname" and "address". When you launch the job it will load the files matching the resources pattern and write the data to this table. As with the filepollhdfs job, this module also supports the deleteFiles parameter which will remove the files defined by the resources parameter on successful completion of the job.

Launch the job using:

This module functions very similarly to the filejdbc one except that the resources you specify should actually be in HDFS, rather than the OS filesystem.

Launch the job using:

The hdfsjdbc job has the following options:

Performs the reverse of the previous module. The database configuration is the same as for filejdbc but without the initialization options since you need to already have the data to import into HDFS. When creating the job, you must either supply the select statement by setting the sql parameter, or you can supply both tableName and columns options (which will be used to build the SQL statement).

To import data from the database table some_table, you could use

You can customize how the data is written to HDFS by supplying the options directory (defaults to /xd/(job name)), fileName (defaults to job name), rollover (in bytes, default 1000000) and fileExtension (defaults to csv).

Launch the job using:

If you want to partition your job across multiple XD containers you can provide the partitionColumn and partitions option. When the job is launched the partitioner will query the database for the range of values and evenly divie the load between the partitions. This assumes that there is an even distribution of column values in the table. When using the partitioning support you must also use the tableName and columns options instead of the sql option. This is so the partitioner can construct the queries with the appropriate where clauses for the different partitions.

An example of a partitioned job could look like this:

The jdbchdfs job has the following options:

Exports CSV data from HDFS and stores it in a MongoDB collection which defaults to the job name. This can be overridden with the collectionName parameter. Once again, the field names should be defined by supplying the names parameter. The data is converted internally to a Spring XD Tuple and the collection items will have an id matching the tuple’s UUID. You can override this by setting the idField parameter to one of the field names if desired.

An example:

The hdfsmongodb job has the following options:

Copies files from FTP directory into HDFS. Job is partitioned in a way that each separate file copy is executed on its own partitioned step.

An example which copies files:

Full path is preserved so that above command would result files in HDFS shown below:

The ftphdfs job has the following options:

A Spark Application can be deployed and launched from Spring XD as a batch job. SparkTasklet submits the Spark application into Spark cluster manager using org.apache.spark.deploy.SparkSubmit. Through this approach, you can also launch a Spark application with specific criteria via Spring XD stream (for instance: A real time scoring algorithm through MLlib spark job can be triggered based on the streaming data events). To get started, please refer to Spark examples here: https://spark.apache.org/examples.html.

Lets run some Spark examples as Spring XD batch jobs:

Once the job is launched, go to Spring XD admin-ui to verify the job results. Jobs → Executions → Select the job to verify that execution context holds the log for Spark application results. If you launch the Spark application through Spark Master, then the results and application status can be verified from SparkUI as well.

The sparkapp job has the following options:

A Sqoop job can be deployed and launched from Spring XD as a batch job. The Sqoop job uses a SqoopTasklet and a SqoopRunner that submits a Sqoop job using org.apache.sqoop.Sqoop.runTool. The Spring XD Sqoop batch job aims to support most of the Sqoop functionality, but at this point we have tested a limited subset:

The intention is to eventually support all features of the Sqoop tool including running Sqoop jobs saved in the Sqoop metastore. See Sqoop User Guide for full documentation of the Sqoop features.

We can test the Sqoop job by just listing the tables in the database:

The definition contains the name of the provided job as sqoop and the --command option names the Sqoop command we want to run, which in this case is "list-tables".

Once the job is launched, go to Spring XD admin-ui to verify the job results. Jobs → Executions → Select the job to verify that step execution context holds the log for Sqoop Tool execution results. You should see some tables listed there. Since we didn’t provide any connection arguments Spring XD will by default use the batch respoitory database for the Sqoop Tool execution. We could provide options specifying a different database using the --url, --username and --password options for the job:

Here we connect to a local MySQL database. It’s important to note that you need to provide the MySQL JDBC driver jar in the Spring XD lib directory for this to work.

There also is an option to specify connection arguments using the --args option. This allows you to use the same arguments that you are used to provide on the command line when running the Sqoop Tool directly. To connect to the same MySQL database as above using --args we would use:

When importing data, you simply use "import" as the command to run. Here is an example:

In this example we provided the connection arguments using the -args option. We could also have used --url, --username and --password options like we did above for the "list-tables" example. The "import" command will use the spring.hadoop.fsUri that is specified when Spring XD starts up. You can override this by providing the --fsUri option when defining the job. The same is true for spring.hadoop.resourceManagerHost and spring.hadoop.resourceManagerPort. You can override the Spring XD configured values with --resourceManagerHost and --resourceManagerPort options.

For exports we use the "export" command. Here is an example:

Here we rely on the connection options to default to the same database used for the batch repository. Note that Sqoop requires that the table to export data into must already exist.

(For this example we have omitted the equal sign for the individual Sqoop arguments within the --args option. Either style works fine.)

The sqoop job has the following options:

The gpload utility can be deployed and launched from Spring XD as a batch job. The gpload job uses a GploadTasklet that submits a gpload job as an external process. The Spring XD gpload batch job aims to support most of the gpload functionality.

We need to provide the following required options:

Here is an example of a basic load job definition. Please note that some options like host, port, database and username could have been specified in the control file as well.

The content of the control file:

This is the command used to create and launch the job:

Once the job is launched, go to Spring XD admin-ui to verify the job results. Jobs → Executions → Select the job to verify that step execution context holds the log for gpload execution results.

We can override the file name for the source file by providing it as a job parameter like this:

This allows us to define a stream to capture new files created in a specific directory:

Now, any new file created in that directory will launch a gpload job for that new file.

The gpload job has the following options:

Sources, processors, and sinks are built using Spring Integration and are typically perform a single task that they may be easily reused in streams. Alternately, a custom module may be required to perform a specific function, such as integration with a legacy service. In Spring Integration terms:

For example, take a look at the file source which simply polls a directory using a file inbound adapter and file sink which appends an incoming message payload to a file using a file outbound adapter. On the surface, there is nothing special about these components. They are plain old Spring XML bean definition files.

Notice that modules adhere to an important convention: The input and output channels are always named input and output, in keeping with the KISS principle (let us know if you come up with some simpler names). The Spring XD runtime uses these names to bind these channels to the message transport.

A module is a packaged component containing artifacts used to create a Spring application context. In general, a module is not aware of its runtime environment. Each module’s application context is configured and connected to other modules via Plugins in order to support distributed processing. In this respect, modules may potentially be applied to purposes other than stream processing. The module types described here (source, processor, sink, and job) are specific to Spring XD, but the Module type is designed to act as a core component of any micro-service architecture built with Spring.

Physically, a Module is somewhat analogous to a war file in Servlet container. The Spring XD container configures and starts a module when it is deployed. Deploying a module in Spring XD terms means activating an instance for processing, not to be confused with deploying a web application in Servlet container. Consistent with the war analogy, a module has it’s own class loader to load resources provided by the module, notably the files found in its config in its lib directories. Another feature in common with a war file is that web applications are installed in a configured location and conform to a standard layout. Artifacts are installed in a known location, either in expanded form or as a single archive file. Spring XD modules work the same way. Spring XD module layout has evolved significantly as new features have been added to support custom module development. This evolution has generally led to increased flexibility with respect to individual artifacts. However, the module’s packaging structure is well defined:

For historical reasons, all modules included with Spring XD distribution are provided in expanded form and are commonly configured using XML bean definition files (<module-name>.xml) and property files (<module-name>.properties>. This is subject to change as this convention is no longer required. Meanwhile the out-of-the-box modules provide copious examples of module configuration and packaging.

A module’s contents typically includes:

If no configuration resource (config/*.xml or config/*.groovy) is present, Spring XD will expect a base_packages property containing a comma delimited list of package names to enable Spring component scanning scoped to the module.

As mentioned previously, a Spring XD module can be installed as an expanded directory tree or an archive. If the module requires dependent jars, which is the typical case, it may be packaged as an uberjar compatible with Spring Boot, and conforming to the above structure. The next section describes Spring XD’s support for module packaging and development.

Spring XD 1.1.x provides support for creating a module project to test and package the module with Maven or Gradle.

The sections Creating a Source Module, Creating a Processor Module, Creating a Sink Module, and Creating a Job Module each reference working examples of custom module projects including in-container tests.

To build the module:

or

See the si-dsl-module example for a complete working example.

A module definition requires the following attributes to uniquely define a module:

All modules included with Spring XD out-of-the-box are located in the xd/modules directory where Spring XD is installed. The Module Registry organizes modules by type in corresponding sub-directories, so a directory listing will look something like:

Spring XD provides a strategy interface ModuleRegistry used to locate a module of a given name and type. Currently Spring XD implements a ResourceModuleRegistry which is configured to locate modules in the following locations in this order:

Custom modules are located separately from out-of-the-box modules. The location is given by xd.customModule.home in servers.yml. The location defaults to ${xd.home}/custom-modules but we strongly recommend setting this to an external location on a network file system if you are using custom modules in production. There are two reasons for doing this. First, custom modules must be accessible to all nodes on the Spring XD cluster, including the XD Admin node. This allows any container instance to deploy the module. Second, if custom modules are registered within the Spring XD installation, they will not survive an upgrade to the Spring XD distribution and will need to be reinstalled.

Starting with Spring XD 1.2, a module can selectively add libraries from paths that are derived from module options. The aim is e.g. to support several alternative implementations in the same module. This works like the following:

This allows constructions like those (assuming for example that we want to create a jpa module that supports several JPA providers):

And, in jpa.properties:

Where the metadata class includes a provider option (of type String) that will take e.g. the values hibernate or eclipse-link. Note the presence of a third ${xd.home}/lib/jpa/${provider}/*.jpa entry that can be used for unforeseen provider implementations.

By convention, Spring XD defined properties are prefixed with xd. Below is the list of all available ${xd.xxx} keys that module authors may use in their declaration.

As we’ve seen so far, a module is a re-usable Spring Integration or Spring Batch application context that can be dynamically configured through the use of module options.

A module option is any value that the may be configured within a stream or job definition. Preferably, the module provides metadata to describe the available options. This section explains how default values are computed for each module option.

In a nutshell, actual values are resolved from the following sources, in order of precedence:

Going into more detail, the platform-wide defaults will resolve like so, assuming option <optionname> of a module <modulename> which is of type <moduletype>:

where

Note that YaML is particularly well suited for hierarchical configuration, so for example, instead of

one can write

Note that options in the .properties files can reference values that appear in the modules.yml file (this makes sharing common configuration easy). Also, the values that are used to configure the server runtimes (in servers.yml) are visible to modules.yml and .properties file (but the inverse is not true).

To create a composite module, use the module compose shell command:

Then, to verify the new module composition was successful, check if it exists:

Notice that the composed module shows up in the list of sink modules. That is because logically it acts as a sink: It provides an input channel (which is bridged to the filter processor’s input channel), but it provides no output channel (since the file sink has no output). Also notice that the module has a small (c) prefixed to it, to indicate that it is a composed module.

If a module were composed of two processors, it would be classified as a processor:

If a module were composed of a source and a processor, it would be classified as a source:

Based on the logical type of the composed module, it may be used in a stream as if it were a simple module instance. For example, to redefine the two streams from the first problem case above, now that the foo sink module has been composed, you can issue the following shell commands:

To test the httpfoo stream, try the following:

The first message should have been ignored due to the filter, but the second one should exist in the file:

To test the filefoo stream, echo "foo" to a file in the /tmp/xd/input/filefoo directory, then verify:

When you no longer need a composed module, you may delete it with the module delete shell command. However, if that composed module is currently being used in one or more stream definitions, Spring XD will not allow you to delete it until those stream definitions are destroyed. In this case, module delete will fail as shown below:

As you can see, the failure message shows which stream(s) depend upon the composed module you are trying to delete.

If you destroy both of those streams and try again, it will work:

When creating a module, if you duplicate the name of an existing module for the same type, you will receive an error. In the example below the user tried to compose a tcp module, however one already exists:

However, you can create a module for a given type even though a module of that name exists but as a different type. For example: I can create a sink module named filter, even though filter already exists as a processor.

Finally, it’s worth mentioning that in some cases duplication may be avoided by reusing an actual stream rather than a composed module. This is possible when named channels are used in the source and/or sink position of a stream definition. For example, the same overall functionality as provided by the two streams above could also be achieved as follows:

This approach is more appropriate for use-cases where individual streams on either side of the named channel may need to be deployed or undeployed independently. Whereas the queue typed channel will load-balance across multiple downstream consumers, the topic: prefix may be used if broadcast behavior is needed instead. For more information about named channels, refer to the Named Channels section.

The http source has the following options:

Here is an example

Post some data to the new port

Since this properties file contains sensitive information, it will typically be secured by the operating system with the XD container process having read access.

The sftp source has the following options:

The tail source has the following options:

Here is an example

Some platforms, such as linux, send status messages to stderr. The tail module sends these events to a logging adapter, at WARN level; for example…​

The file source has the following options:

The ref option is useful in some cases in which the file contents are large and it would be more efficient to send the file path.

The gemfire source has the following options:

Use of the gemfire source requires an external process (or a separate stream) that creates or updates entries in a GemFire region configured for a cache server. Such events may feed a Spring XD stream. To support such a stream, the Spring XD container must join a GemFire distributed client-server grid as a client, creating a client region corresponding to an existing region on a cache server. The client region registers a cache listener via the Spring Integration GemFire inbound channel adapter. The client region and pool are configured for a subscription on all keys in the region.

The following example creates two streams: One to write http messages to a Gemfire region named Stocks, and another to listen for cache events and record the updates to a file. This works with the Cache Server and sample configuration included with the Spring XD distribution:

Now send some messages to the stocks stream.

As updates are posted to the cache you should see them captured in the output file:

The last example creates connections to myhost1:10334, myhost2:10335, myhost3:10336

This source requires a cache server to be running in a separate process and its host and port, or a locator host and port must be configured. The XD distribution includes a GemFire server executable suitable for development and test purposes. This is a Java main class that runs with a Spring configured cache server. The configuration is passed as a command line argument to the server’s main method. The configuration includes a cache server port and one or more configured region. XD includes a sample cache configuration called cq-demo. This starts a server on port 40404 and creates a region named Stocks. A Logging cache listener is configured for the region to log region events.

Run Gemfire cache server by changing to the gemfire/bin directory and execute

The gemfire-cq source has the following options:

The example is similar to that presented for the gemfire source above, and requires an external cache server as described in the above section. In this case the query provides a finer filter on data events. In the example below, the cqtest stream will only receive events matching a single ticker symbol, whereas the gftest stream example above will receive updates to every entry in the region.

Now send some messages to the stocks stream.

The cqtest stream is now listening for any stock quote updates for the ticker symbol FAKE. As updates are posted to the cache you should see them captured in the output file:

The tcp source has the following options:

The following examples all use echo to send data to netcat which sends the data to the source.

The echo options -en allows echo to interpret escape sequences and not send a newline.

This uses the default (CRLF) decoder and port 1234; send some data

See if the data ended up in the file

Note that we configure the file sink with binary=true so that a newline is not appended.

The tcp-client source has the following options:

That "stimulus" counter concept bears some explanation. By default, the module will emit (at interval set by fixedDelay) an incrementing number, starting at 1. Given that the default is to use an expression of payload.toString(), this results in the module sending 1, 2, 3, ... to the remote server.

By using another expression, or more certainly a script, one can implement a simple conversation, assuming it is time based. As an example, let’s assume we want to join some kind of chat server where one first needs to authenticate, then specify which rooms to join. Lastly, all clients are supposed to send some keepalive commands to make sure that the connection is open.

The following groovy script could be used to that effect:

The rabbit source has the following options:

See the RabbitMQ MessageBus Documentation for more information about SSL configuration.

The jms source has the following options:

The mqtt source has the following options: