CogStack data processing engine

There are multiple configurable parameters available to tailor the CogStack data processing pipeline to the specific data processing needs and available resources. CogStack data processing pipeline design follows the principles behind Spring Batch. That is, there are different Jobs and Steps and custom processing components called ItemReaders , ItemWriters and ItemProcessors . A Job has one to many Step, which has exactly one ItemReader, ItemProcessor, and ItemWriter. A Job needs to be launched by JobLauncher, and meta data about the currently running process needs to be stored in JobRepository. The picture below presents a simplified version of a reference batch processing architecture. 

In CogStack engine multiple data processing operations can be chained together in one ItemProcesor. All the selected data processing operations with ItemReaders and ItemWriters are later assembled in compositeSlaveStep Bean (defined in uk/ac/kcl/batch/JobConfiguration.java) according to the active Spring profiles specified in the user-defined CogStack job properties file. This allows for a significant degree of flexibility when defining data processing pipelines using CogStack and which allows for future extendability with new components. 

Moreover, apart from data processing components, there are also other aspects that CogStack pipeline supports, such as data partitioning, parallel and multithreaded execution of jobs, job scheduling, etc.. All the currently implemented functionality is briefly described below in the following sections.

For more details on how does Spring Batch framework works and what functionality it offers, please refer to the official Spring Batch documentation.

Document model

CogStack data processing engine uses an internal document model to perform mapping between the input data source (supplied to item readers), item processors and output sink (supplied through item writers). The currently available document model is defined in uk/ac/kcl/model/Document.groovy file.

The current document model defines a number of available fields which are used when defining mappings in CogStack job properties files. Please refer to Quickstart tutorial to see some examples working directly with the data.

Records data mapping

The document model exposes such fields to perform the mapping from the input data source:

• String srcTableName
• String srcColumnFieldName
• String primaryKeyFieldName
• String primaryKeyFieldValue
• Timestamp timeStamp

These fields are used to provide direct access to the underlying record data from the source database table to the desired item processors.

In the current data model implementation, it is mandatory to provide mappings for primaryKeyFieldValue and timeStamp fields only. The remaining fields are used in docman profile.

Moreover, some database writers also rely on the document model to map the record data into output database table – please see Target SQL query part.

Documents data mapping

When processing documents, in addition to the standard records data mapping fields, some fields are available to perform the mapping from the input data source:

• byte[] binaryContent

The binaryContent field stores the binary data of the document to be processed by selected item processor. The document can be of any of the formats that are supported by the desired item processor(s). The processed document data is later saved in associativeArray field by the active item processor (see below).

The profiles that use the functionality of these fields:

• tika
• metadata
• pdfbox
• pdfgen
• jdbc_out
• jdbc_out_map

Associative array for JSON mapping

In addition to the fields which can be mapped by the user, CogStack data model defines an additional field for storing auxiliary information:

• HashMap<String,Object> associativeArray

The information which cannot be handled by the standard record fields nor stored in the documents mapping fields can be represented in the associativeArray field as a key:value pair by item processors. However, some document writers, such as mapJdbcItemWriter, can directly access the elements from the associativeArray field when writing records into output database table.

Moreover, all the record fields from the specified input database table are directly mapped into this array (with key being the name of the column). Thanks to that, it is possible to generate an output JSON file (see: JSONMakerItemProcessor) which contains all the record information both originating from the input database table, containing processed document data and containing any useful auxiliary information.

In the current implementation, the data model provides such fields to store the resulting record data in JSON format:

• String outputData

Such generated JSON file can be then sent directly to ElasticSearch (see: ElasticSearch sink connector).

Job properties file

CogStack data processing engine is highly configurable with a large number of parameters available to tailor the data processing requirements. All the parameters are set in the Java properties file which is later parsed when launching CogStack instance. Each CogStack instance should use it's own separate properties file to avoid possible conflict with the used resources.

The properties file used by CogStack is usually divided into sections, each covering one aspect of the CogStack job or data processing pipeline. For the ease of readability, it should be organised in the following sections:

• Active profiles
• Data source with reader configuration
• Data sink with writers configuration
• Data processors with configuration(s)
• Data partitioner and data step processing configuration
• General job, execution, job repository and scheduler configuration

For an example job properties file file please refer to Quickstart or Examples part.

Spring profiles

CogStack configuration file uses Spring profiles, which enable different components of the data processing pipeline. The profiles can be divided into categories regarding the functionality they cover. They are a part of CogStack job configuration and need to be specified before launching the CogStack instance. The profiles are enabled by specifying them as a comma separated values in the property:

• spring.profiles.active

The currently available profiles are listed below.

Reading the data

jdbc_in

This profile enables JDBC input database connector used for reading records from the input databases. It is one of the most commonly used profiles.

Components activated by this profile:

• documentItemReader 

docman

This profile enables processing documents read as files from the filesystem while keeping their path in the database. It is used together with other profiles, as multiple components implement functionality offered by this profile.

Processing data

deid

This profile enables de-identification process using GATE or ElasticGazetteer services. 

Components activated by this profile:

• deIdDocumentItemProcessor

Services activated by this profile:

• ElasticGazetteerService
• gate

tika

This profile enables extraction of text from documents of different formats (PDF, DOCX, ...) and from images (JPEG, PNG, ...) using Apache Tika. Please refer to Apache Tika documentation (v.1.7) for a list of supported formats.

Components activated by this profile:

• tikaDocumentItemProcessor

gate

This profile enables NLP data processing using generic applications provided in GATE NLP suite.

Components activated by this profile:

• gateDocumentItemProcessor

Services activated by this profile:

• GateService

pdfGeneration

This profile enables generation PDF documents from the input records.

Components activated by these profiles:

• PDFGenerationItemProcessor

thumbnailGeneration

This profile enable generation of documents thumbnails for preview.

Components activated by these profiles:

• ThumbnailGenerationItemProcessor

dbLineFixer

This profile enables concatenation of records which are split across multiple rows in the input database.

Components activated by this profile:

• DbLineFixerItemProcessor

pdfbox

This profile enables processing of the documents using Apache PDFBox.

Components activated by this profile:

• PdfBoxItemProcessor

metadata

This profile enables extraction of metadata information from the documents using ImageMagick as an external application.

Components activated by this profile: 

• MetadataItemProcessor

webservice

This profile enables processing of the documents using an external webservice.

Components activated by this profile:

• WebserviceDocumentItemProcessor

Writing the data

jdbc_out

This profile enables JDBC output database connector used for writing the processed EHRs to databases to the table specified in the properties file. It uses the Document model to map directly the record data with processed fields (e.g., by Tika) from the input table directly into the output table. It uses target.Sql property directive to perform the mapping (see: Target SQL query). The record fields not covered by the provided SQL query are output as text in the JSON format as outputData field which requires manual parsing to access it.

Components activated by this profile:

• simpleJdbcItemWriter

jdbc_out_map

This profile enables a JDBC output database connector used for writing to databases using direct row/document mapping. Selected fields from the input table alongside the processed data are mapped to the output table according to the SQL query (as defined by target.Sql property) using the Document model. In contrast to jdbc_out , here parsing of the JSON output is not required, as only the specified fields are being mapped. Access to the other fields from the input record or access to some auxiliary fields, such as, status of Tika parsing, is therefore not available.

Components activated by this profile are:

• mapJdbcItemWriter

elasticsearch

This profile is used to insert documents to ElasticSearch using the ElasticSearch Java Transport Client API.

Components activated by this profile are:

• esDocumentWriter

elasticsearchRest

This profile is used to insert documents to ElasticSearch through exposed ElasticSearch REST endpoint using the ElasticSearch Java Low Level REST Client.

Components activated by this profile:

• esRestDocumentWriter

Services activated by this profile: 

• EsRestService

jsonFileItemWriter

This profile enables writing documents directly to files in JSON format. It can be used in restricted scenarios when one cannot use a database as a data sink or just when one wants to operate directly on the files.

Components activated by this profile are: 

• jsonFileItemWriter

Partitioning and scaling

localPartitioning

This profile enables partitioning functionality when running jobs locally on JVM. For more details please see the Data Partitioning and Running jobs sections.

remotePartitioning

This profile enables partitioning functionality when running jobs by multiple JVMs coordinated using JMS in a master-slave fashion. For more details please see the Data Partitioning and Running jobs sections.

Data source and item readers

CogStack data processing engine can read records and documents from multiple sources. For the moment, however, only reading from databases and filesystem is supported using a generic document reader.

Generic document reader

CogStack data processing engine implements a generic document item reader documentItemReader . It is used both for reading the documents from a database table (or view) and/or filesystem. The reader extends Java Spring JdbcPagingItemReader and provides capabilities of processing documents in batches. Hence, it requires some additional configuration, such as partitioning schemes, batch size, etc., – for more information see Data partitioning.

In order to read the documents from the input database, the user needs to define the SQL block that specifies the data to process. This block is composed of three parts – defined by mandatory properties:

• source.selectClause – the SELECT statement clause (possibly followed by the names of the columns), e.g.: SELECT *

• source.fromClause – the FROM clause followed the name of the table, e.g.: FROM <in_table>,

• source.sortKey – the name of the column in the source database table by which the records are ordered.

The available optional properties:

• .source.pageSize (if not specified, defaults to step.chunkSize) – the number of rows to retrieve (per read) at a time from the input database.

Database connectors

CogStack supports multiple JDBC connectors as provided in Java Spring framework. The most commonly used ones here are MySQL and PostgreSQL. However, using the other ones implemented in Java Spring framework should be also possible, such as: Derby, DB2 HSQL, ORACLE, Sybase, H2 or SQLite connectors.

Spring profiles that are using source JDBC connectors:

• jdbc_in

Connector configuration

There are multiple mandatory properties which need to be provided in the job configuration to set up the source database connector handled by Hikari Connection Pool:

• source.Driver – source JDBC driver class name,
• source.JdbcPath – JDBC path to the source database,
• source.username – DB user name,
• source.password – DB user password,
• source.poolSize – the size of the connection pool.

The available optional properties:

• source.poolSize (default: 10) – the size of the connection pool,
• source.idleTimeout (default: 30000) – max. idle timeout value (in ms) that a client will wait for the connection,
• source.maxLifetime (default: 60000) – max connection life time value (in ms) of a connection in the pool,
• source.leakDetectionThreshold (default: 0 ) – the max. amount of time (in ms) that a connection can be out of the pool before a potential connection resource leak.

Document row mapper configuration

The main component in CogStack engine providing mapping between the source database, Document model and item processors is DocumentRowMapper. It uses multiple properties, configuration of which depends on the workflow, selected Spring profiles and active components.

The available mandatory properties are:

• source.srcTableName – the name of the database table where the current document to be processed reside,
• source.srcColumnFieldName – the name of the column which stores the name or filename of the document,
• source.primaryKeyFieldName – the name of the column which stores the primary key for the current document to be processed (used for partitioning),
• source.primaryKeyFieldValue – the name of the column which stores the primary key value (used for partitioning),
• source.timeStamp – the name of the field which stores the timestamp value (used for partitioning).

The available optional properties (used with different item processor components) are:

• elasticsearch.datePattern (default: yyyy-MM-dd'T'HH:mm:ss.SSS) – the date pattern used for compatibility when indexing documents to ElasticSearch (but it is also used by the row mapper),
• reindex (default: false) – used for enabling re-indexing the data (see below),
• reindexField (using: reindex) – the name of the field to reindex (generated from the previous step),
• tika.binaryContentSource (possible values: database | fileSystemWithDBPath ; default: database) – specifies where the binary documents to be processed by Tika reside,
• tika.binaryFieldName – the name of the field where the Tika binary document output is stored,
• tika.binaryPathPrefix (using: fileSystemWithDBPath) – prefix of the filesystem path where the documents reside,
• tika.binaryFileExts (using: fileSystemWithDBPath) – possible binary file extensions to include.

Item processors and services

CogStack implements a range of different data processors and services to be able to build fully customisable pipelines. Thanks to following the Spring Batch data processing patterns, it is possible to extend CogStack pipeline with custom data processors or adding additional services.

Item processors

CogStack pipeline implements compositeItemProcessor (based on Spring Batch CompositeItemProcessor) that is used to launch delegate item processors chained together. Each processor extends the functionality of the abstract class TLItemProcessor and implements the functionality of Java abstract class ItemProcessor<Document, Document>. Below are described more in details the available item processors.

CogStack engine provides the following processors:

dBLineFixerItemProcessor

Allows to merge records from a specified table in a database spanning into multiple lines (and records) into a single record.

Active in profiles:

• dBLineFixer

Available mandatory properties:

• lf.documentKeyName – primary key of original document,
• lf.lineKeyName – primary key from table with multi row documents,
• lf.srcTableName – source table of multi row documents,
• lf.lineContents – text content from table with multi row documents,
• lf.fieldName – output field name.

gateDocumentItemProcessor

This components implements NLP data processing using generic applications from GATE suite.

Active in profiles:

• gate

Available mandatory GATE document item processor properties:

• gate.gateFieldName – the name of the output JSON field into which the GATE application will output the result,
• gate.fieldsToGate – comma separated names of fields in SQL that will be processed by the GATE application.

Available mandatory GATE service properties:

• gate.gateHome – GATE home directory for loading plugins,
• gate.gateApp – the target GAPP or XAPP application to be run as a GATE component of the pipeline,
• gate.deIdApp (with profile: deid) – the name of the GAPP or XAPP application to perform de-identification,

Available optional GATE service properties:

• gate.poolSize (default: 1) – number of GATE pipelines to be loaded when in multithreading mode.
• gate.gateAnnotationSets (when not specified – using default set for extraction) – comma separated Annotation Sets to be extracted; need to exclude to use annotations in the default set,
• gate.gateAnnotationTypes (when not specified – all types will be extracted) – comma separated Annotation Types to extract.

deIdDocumentItemProcessor

This component implements de-identification of records using GATE or custom ElasticGazetteer service.

Active in profiles:

• deid

The property deid.useGateApp controls whether to use GATE (value: true) or ElasticGazetteer (value: false) for performing de-identification.

Available mandatory properties when using GATE application:

• deid.useGateApp – this needs to be set to true,
• gate.deIdApp – the name of the GAPP or XAPP application to perform de-identification,

• deid.fieldsToDeId – comma separated list of field names to process, it maps the resulting fields to database column labels,

• deid.replaceFields – if value set to true, it replaces the text strings in the fields listed by deid.fieldsToDeId otherwise (value: false), creates a new field called as deidentified_<originalFieldName>.

Available mandatory properties when using ElasticGazetteer service:

• deid.useGateApp – this needs to be set to false,

• deid.stringTermsSQLFront – SQL query to retrieve strings of text for masking – e.g., names, addresses, etc., ; must surround document primary key,

• deid.timestampTermsSQLFront –  SQL query to retrieve strings of text for masking date/time values, similar as in deid.stringTermsSQLFront.

The available optional properties:

• deid.levDistance (default: 30 ) – Levenshtein distance between 1 and 100 for performing matching – the higher value the greater fuzziness,
• deid.minWordLength (default: 3 ) – don't mask words shorter than this many characters, 
• deid.stringTermsSQLBack – second part of the corresponding above Front SQL string (if required),
• deid.timestampTermsSQLBack – second part of the corresponding above Front SQL string (if required).

JSONMakerItemProcessor

This components is used to generate the document output in JSON format using the Document model. It is active in all profiles and used by default (to meet requirements of composite processor).

MetadataItemProcessor

This component implements extraction of document metadata, primarily the number of pages. It uses ImageMagick as an external application for this task.

Available optional properties:

• metadata.eagerGetPageCount (default: false) – when set to true instructs to get the page count of the document.

Active in profiles:

• metadata

PdfBoxItemProcessor

This component implements processing of documents using Apache PDFBox.

Active in profiles:

• pdfbox

Available optional properties:

• pdfbox.namespaces – namespaces to perform mapping.
• pdfbox.mapping – used mappings.

pdfGenerationItemProcessor

This component implements generation of documents in PDF format from the provided input. It uses LibreOffice and ImageMagick as external applications.

Active in profiles:

• pdfGeneration
• thumbnailGeneration

Available mandatory properties:

• pdfGenerationItemProcessor.fileOutputDirectory – the path where the generated PDF files will be stored.

thumbnailGenerationItemProcessor 

 This component implements generation of thumbnails of PDF documents for preview. It uses ImageMagick as an external applications.

Active in profiles:

• thumbnailGeneration

Available mandatory properties:

• thumbnailGenerationItemProcessor.fileOutputDirectory – the path where the generated thumbnail files will be stored.

Available optional properties:

• thumbnailGenerationItemProcessor.thumbnailDensity (default: 150) – the pixel density used when generated thumbnail images.

tikaDocumentItemProcessor

This component implements processing of binary documents using Apache Tika. It implements a modified Tika documents parser for processing PDF documents. It supports processing the documents either reading them directly from the specified input database table or from filesystem (see: docman profile). It uses ImageMagick (for image conversion) and TesseractOCR (for performing OCR) external applications.

Active in profiles:

• tika

Available mandatory properties:

• tika.binaryFieldName – the name of the field (in the input database table) containing the document binary data,
• tika.binaryPathPrefix (using: fileSystemWithDBPath) – the path prefix where the document files are stored.

Available optional properties:

• tika.tikaFieldName (default: outTikaField ) – the name of the output field in the output database table or JSON key into which the parsed text document will be stored,
• tika.binaryContentSource – (values: database | fileSystemWithDBPath; default: database) – specifies where the documents to be processed are kept (setting the fileSystemWithDBPath requires also setting the docman profile),
• tika.keepTags (default: false) – specifies whether the document shall be parsed into XHTML (true) format or plaintext (false),

• tika.tesseract.timeout (default: 120) – the max. timeout value (in s) when extracting the text from image documents using TesseractOCR,

• tika.convert.timeout (default: 120) – the max. timeout value (in s) when converting the documents using ImageMagick.

webserviceDocumentItemProcessor 

This component implements processing of documents through an external webservice. It uses the REST API to communicate with the webservice and sends and receives the documents in JSON format.

Active in profiles:

• webservice

Available mandatory properties:

• webservice.endPoint – the URL for the webservice REST endpoint,

• webservice.fieldsToSendToWebservice – a comma seperated field names (SQL columns) to send.

Available optional properties:

• webservice.name (default: defaultWSName) – the custom name of the webservice,

• webservice.fieldName (default: outWSField) – the name of the output field to be created,

• webservice.connectTimeout (default: 10000) – max. timeout value (in ms) for connection to the service,

• webservice.readTimeout (default: 60000) – max. timeout value (in ms) to receive the response from the service,

• webservice.retryTimeout (default: 5000) – max. time value (in ms) before performing a (first) retry,

• webservice.retryBackoff (default: 3000) – increase the last retry time using this value (in ms) before performing the next retry. 

Services

ElasticGazetteer

This service implements basic de-identification functionality based on configurable Levenshtein distance between analysed fields in the records.

Active in profiles:

• deid

Used by components:

• deIdDocumentItemProcessor

Available mandatory properties when using ElasticGazetteer:

• deid.useGateApp – this needs to be set to false,

• deid.stringTermsSQLFront – SQL query to retrieve strings of text for masking – e.g., names, addresses, etc., ; must surround document primary key,

• deid.timestampTermsSQLFront –  SQL query to retrieve strings of text for masking date/time values, similar as in deid.stringTermsSQLFront.

Available optional properties:

• deid.levDistance (default: 30 ) – Levenshtein distance between 1 and 100 for performing matching – the higher value the greater fuzziness,
• deid.minWordLength (default: 3) – don't mask words shorter than this many characters, 
• deid.stringTermsSQLBack – second part of the corresponding above Front SQL string (if required),
• deid.timestampTermsSQLBack – second part of the corresponding above Front SQL string (if required).

EsRestService

This service implements a REST client for reading and writing documents into ElasticSearch using ElasticSearch Java Low Level REST Client.

Active in profiles:

• elasticsearchRest

Used by components:

• esRestDocumentWriter

Available mandatory properties:

• elasticsearch.cluster.host – the host name of the ElasticSearch cluster, 
• elasticsearch.cluster.port – the port number of the cluster.

Available optional properties:

• elasticsearch.index.name (default: default_index) – the name of the index in ElasticSearch under which documents are or will be stored,
• elasticsearch.type (default: doc)– the type of the documents (deprecated in ElasticSearch 6.0+),
• elasticsearch.cluster.name (default: elasticsearch) – the name of the ElasticSearch cluster,
• elasticsearch.cluster.slaveNodes – the list of comma-separated <host:port> values for multi-node deployments,
• elasticsearch.connect.timeout (default: 5000) – max. time value (in ms) for connection timeout,

• elasticsearch.response.timeout (default: 60000) – max. timeout value (in ms) when waiting for response,

• elasticsearch.retry.timeout (default: 60000) – max. time value (in ms) before performing retry,

• elasticsearch.xpack.enabled (default: false) – specifies whether to enable security module in X-Pack secirity module,

Available properties when using the ElasticSearch X-Pack security module:

• elasticsearch.xpack.enabled – needs to be set to true,
• elasticsearch.xpack.user – the user name for logging in to the ElasticSearch cluster,

• elasticsearch.xpack.password – the user password,

• elasticsearch.xpack.security.transport.ssl.enabled (default: false) – specifies whether to use SSL encryption for communication between ElasticSearch nodes,
• elasticsearch.xpack.ssl.keystore.path – the path to the Java Keystore file that contains a private key and certificate,
• elasticsearch.xpack.ssl.keystore.password – the password to the keystore,

• elasticsearch.xpack.ssl.truststore.path – the path to the Java Keystore file that contains the certificates to trust,

• elasticsearch.xpack.ssl.truststore.password – the password to the truststore.

GateService

This service implements a GATE client for interacting with GATE NLP suite and running native XAPP and GAPP applications.

Active in profiles:

• gate
• deid

Used by components:

• gateDocumentItemProcessor

Available mandatory properties:

• gate.gateHome – GATE home directory for loading plugins,
• gate.gateApp – the target GAPP or XAPP application to be run as a GATE component of the pipeline,
• gate.deIdApp (with profile: deid) – the name of the GAPP or XAPP application to perform de-identification.

Available optional properties:

• gate.gateAnnotationSets (when not specified – using default set for extraction) – comma separated Annotation Sets to be extracted; need to exclude to use annotations in the default set,
• gate.gateAnnotationTypes (when not specified – all types will be extracted) – comma separated Annotation Types to extract,
• gate.poolSize (default: 1) – number of GATE pipelines to be loaded when in multithreading mode.

Data sink and item writers

CogStack data processing engine implements the functionality to store the processed documents both in different data sinks. This can be either databases, ElasticSearch or files. To do this. CogStack engine provides a number of item writers implementing Java abstract class ItemWriter<Document>. Depending on the job configuration, they are later chained together using Java CompositeItemWriter<> . Below are presented currently implemented database connectors with possible item writers.

Database connectors

Similarly as in case of source database connectors, CogStack supports multiple JDBC sink connectors as provided in Java Spring framework. Here also the most commonly used ones are MySQL and PostgreSQL. However, using the other ones implemented in Java Spring framework should be also possible, such as: Derby, DB2 HSQL, ORACLE, Sybase, H2 or SQLite connectors.

Spring profiles that are using sink JDBC connectors:

• jdbc_out
• jdbc_out_map

Database connector configuration

There are multiple mandatory properties which need to be provided in the job configuration to set up the source database connector handled by Hikari Connection Pool:

• target.Driver – target JDBC driver class name,
• target.JdbcPath – JDBC path to the target database,
• target.username – DB user name,
• target.password – DB user password.

Available optional properties:

• target.poolSize (default: 10) –  the size of the connection pool,
• target.idleTimeout (default: 30000) – max. idle timeout value (in ms) that a client will wait for the connection,
• target.maxLifetime (default: 60000) – max connection life time value (in ms) of a connection in the pool.

Target SQL query

In addition to specifying the connector configuration, it is essential to also specify the target SQL query exposed by the property 

• target.sql

This SQL query is used to perform the mapping between the Document model and the target database table storing the processed document with selected input table fields. The query can only address the fields available in the document model and in the target database table schema. 

Using the current Document model only selected fields are available to be used in the target SQL query to perform the mapping. The rest of the fields from the input table are represented as key:value in the associativeArray in the document model. Alternatively, the final JSON data can be accessed through the field outputData. Depending on the data pipeline design and requirements, the target SQL query can perform the mapping of the document data using either of the fields. Please see simpleJdbcItemWriter and mapJdbcItemWriter for more information.

ElasticSearch connector

CogStack engine also allows for writing processed directly to ElasticSearch. The documents are generated into JSON format using JSONMakerItemProcessor which allows then for easy indexing.

Spring profiles that are using ElasticSearch connectors:

• elasticsearch
• elasticsearchRest

Item writers

CogStack implements a range of item writers to be able to build fully customisable pipelines – the currently implemented ones are listed below.

simpleJdbcItemWriter 

This component implements writing of documents to the specified output database table according to the specified target SQL query. It uses Java BeanPropertyItemSqlParameterSourceProvider to perform the mapping between the direct Document model and the output table. Usually, this writer uses outputData field from the document model.

Active in profiles:

• jdbc_out

Available mandatory properties:

• target.sql – SQL query to perform the mapping between the input database table, Document model and output database table

Requires also specifying the configuration properties for DB connector.

mapJdbcItemWriter 

This component implements writing of documents to the specified output database table according to the specified target SQL query. It uses a custom implementation of Java ItemSqlParameterSourceProvider<Document> to perform the mapping between the Document model and the output table. To perform the mapping, it uses available key:value in the associativeArray field from Document model.

Active in profiles:

• jdbc_out_map

Available mandatory properties:

• target.sql – SQL query to perform the mapping between the input database table, Document model and output database table.

Requires also specifying the configuration properties for DB connector.

esDocumentWriter

This profile is used to insert documents to ElasticSearch using the native ElasticSearch Java Transport Client.

Active in profiles:

• elasticsearch

Requires also specifying the configuration properties for ElasticSearch connector (the same as for ElasticSearch REST service).

esRestDocumentWriter

This component is used to insert documents to ElasticSearch through exposed ElasticSearch REST endpoint using the ElasticSearch Java Low Level REST Client.

Active in profiles:

• elasticsearchRest

Requires also specifying the configuration properties fo ElasticSearch REST service.

jsonFileItemWriter

This component writes documents directly to files in JSON format. It can be used in restricted scenarios when one cannot use a database as a data sink or just when one wants to operate directly on the files.

Active in profiles: 

• jsonFileItemWriter

Available mandatory properties:

• fileOutputDirectory – the output directory where the generated JSON files will be stored

Running jobs

CogStack data processing engine offers a broad range of options to tailor the execution of the pipeline jobs according to the requirements, available resources and desired performance. All the user has to do is to set essential parameters and Spring Batch framework will take care of running the jobs, scaling their work and parallelisation. There are multiple ways to configure the running jobs where the main concepts are covered below. How to run CogStack pipeline either as as a standalone application or as a part of ecosystem is covered in Running CogStack part.

General job configuration

There are multiple parameters to configure the pipeline batch jobs execution, with the mandatory properties to set covered in the sections below.

However, there are some optional properties available:

• job.jobName (default: defaultJob) – the name of the running job, useful for controlling the execution state and auditing.
• configuredStart.firstJobStartDate – if set, starts processing the records from the specified date, 
• configuredStart.datePatternForSQL – the date time patter for the particular DBMS that can be cast to an appropriate timestamp variable.

Execution modes

In principle, there are two modes in which CogStack pipeline can be executed: local and master-slave mode.

Local mode

In this mode jobs are executed inside a single JVM instance. All the data partitioning, job execution and coordination is being handled by a single process and multiple jobs can be running in parallel or in a multithreaded mode. This is the basic mode and the recommended one to use.

This mode is activated by enabling Spring profiles:

• localPartitioning

Master-slave mode

In this mode jobs are executed in a master-slave fashion, running multiple worker JVM instances and a coordinating master JVM instance. All the step execution is being performed by the slave workers running as separate processes. Data partitioning, job controlling with workers coordination is being handled by the master JVM instance. The coordination of and communication between JVM instances uses Apache ActiveMQ messaging and integration patterns server.

This mode is activated by enabling Spring profiles:

• remotePartitioning

Available mandatory properties:

• jms.ip – the IP address of the Java Messaging Service (here: ApacheMQ instance),
• jms.username – the username to connect to the JMS,
• jms.password – the password to connect to the JMS,
• jms.requestQueueName – the name of the JMS request channel (should be unique per job),
• jms.replyQueueName – the name of the JMS reply channel (should be unique per job),

Available optional properties:

• job.jobTimeout (default: 10000000) – the max. time value (in ms) that should complete before if will be listed as failed,
• jms.closeTimeout (default: 100000) –  timeout value (in ms) before closing the connection,
• globalSocketTimeout – the max. time value (in ms) to specify the global socket timeout (as some JDBS drivers don't support sockets timeouts),

In addition, it requires indicating whether the running CogStack instance will be running as a master or a slave. This is being handled by configuring properties: 

• execution.mode (default: local ) – can be local (default) or remote (master-slave mode) and specifies the execution mode,

• execution.instanceType – which can be master or slave and it is required when set‌ execution.mode=remote .

Job repository

In order to control the batch jobs execution, CogStack engine requires setting up a job repository. It stores the current and historical job execution states and allows to check and re-start the failed ones or resume jobs execution from a previously stopped state. Usually a database is used as a job repository and CogStack engine can use JDBC connectors as provided in Java Spring framework, which can be one of: PostgreSQL, MySQL, Derby, DB2 HSQL, ORACLE, Sybase, H2 or SQLite connectors. The most commonly used one however is PostgreSQL .

Available mandatory properties:

• jobRepository.Driver – target JDBC driver class name,
• jobRepository.JdbcPath – JDBC path to the target database,
• jobRepository.username – DB user name,
• jobRepository.password – DB user password.

Available optional properties:

• jobRepository.poolSize (optional, default: 10) - the size of the connection pool,
• jobRepository.idleTimeout (default: 60000) – max. idle timeout value (in ms) that a client will wait for the connection,
• jobRepository.maxLifetime (default: 30000) – max connection life time value (in ms) of a connection in the pool.

Data partitioning

CogStack engine follows the batch processing principles as provided by Spring Batch framework. It uses data partitioning allows to split the input data into multiple smaller chunks according to the requirements and user-defined job configuration. This allows to process the data by running multiple processing jobs either as multi-threaded (local execution mode) or running in parallel as separate JVM instances (master-slave execution mode) – see Execution modes.

Available mandatory properties:

• partitioner.tableToPartition – the name of the table in the input database that contains the columns used to partition the data (see: partitioner.pkColumnName and partitioner.timeStampColumnName),
• partitioner.pkColumnName – the name of the column containing the records primary key used when partitioning the data,
• partitioner.timeStampColumnName (optional) – the column name containing the records timestamp used when partitioning the data,
• source.dbmsToJavaSqlTimestampType – since different DBMS interpret the SQL standard for time differently, is is necessary to explicitly specify, the date type that the database is using. E.g.: PostgreSQL: TIMESTAMP, SQL Server: DATETIME.

Available optional properties:

• partitioner.gridSize (default: 1) – specifies the number of data blocks to create to be processed by (usually) the same number of workers – e.g., for 10 rows to be processed, setting gridSize to 5 will give 10 / 5 = 2 rows per thread,
• partitioner.partitionType (values: PK | PKTimeStamp ; default: PKTimeStamp ) – the type of partitioning method ; for the moment PKTimeStamp only is fully supported.
• partitioner.preFieldsSQL – required prefix on some types of used databases (e.g., SQL Server: TOP 100 PERCENT),
• partitioner.maxPartitionSize – if set, overrides the grid size and sets the max number of rows per partition (warning: if used, can result in generating large numbers of partitions per job).

Moreover, there are available some optional parameters, but used when performing remote partitioning in master-slave execution mode:

• partitioner.partitionHandlerTimeout (default: 0) – the timeout value (in ms) to receive messages from the server.

Data processing step control

CogStack engine also exposes parameters to tailor the step execution for when configuring pipeline jobs. 

Available optional properties:

• step.chunkSize (default: 50) – defines commit interval in step, i.e., to process this many rows before writing results to the output sink and updating the jobs repository,
• step.skipLimit (default: 5) – the total number of exceptions that can be thrown by a job before it fails,
• step.concurrencyLimit (default: 2) – the max. size of the thread pool for running multithreading partitions.

Scheduling

Another useful feature that CogStack implements is running jobs in a scheduled fashion according to the user configuration. By default, scheduler is off and CogStack Pipeline is run in a local execution mode.

Available optional parameters:

• scheduler.useScheduling (default: false) – specifies whether run CogStack with or without job scheduling,
• scheduler.rate (when: scheduler.useScheduling = true)  – the cron scheduling rate pattern, a list of six single space-separated fields representing second, minute, hour, day, month, weekday; E.g., "0 0/30 8-10 * * *"  – 8:00, 8:30, 9:00, 9:30 and 10 o'clock every day,
• scheduler.processingPeriod (default: 777600000000 ms (9000 days)) –  if set, process records in the time frame of the provided number of milliseconds from last successful job/first timestamp/start date.
  
  

Known issues

Incorrect document timestamp in ElasticSearch

CogStack Pipeline is preferably being run inside a container. The docker containers are by default using UTC time. Hence CogStack jobs in Docker containers are running in UTC, and when the documents are ingested into the Elastic index, they’re ending up with the date fields being one hour ahead of expected. Eg. the document added field would be actually 1 hour in the future compared to BST.

The solution here would be to run the CogStack Pipeline container with setting explicitly the time zone, e.g. in the appropriate docker-compose file (thanks Phil Davidson (Unlicensed)). More information can be found on ServerFault discussion .