ETL mit Spring Cloud-Datenfluss

1. Übersicht

Spring Cloud Data Flow ist ein Cloud-natives Toolkit zum Erstellen von Echtzeit-Datenpipelines und Batch-Prozessen. Der Spring Cloud-Datenfluss kann für eine Reihe von Anwendungsfällen für die Datenverarbeitung verwendet werden, z. B. für einfachen Import / Export, ETL-Verarbeitung, Ereignis-Streaming und Predictive Analytics.

In diesem Tutorial lernen wir ein Beispiel für ETL (Extract Transform and Load) in Echtzeit mithilfe einer Stream-Pipeline, die Daten aus einer JDBC-Datenbank extrahiert, in einfache POJOs transformiert und in eine MongoDB lädt.

2. ETL- und Event-Stream-Verarbeitung

ETL - Extrahieren, Transformieren und Laden - wurde allgemein als ein Prozess bezeichnet, der Daten aus mehreren Datenbanken und Systemen stapelweise in ein gemeinsames Data Warehouse lädt. In diesem Data Warehouse ist es möglich, umfangreiche Datenanalyseverarbeitungen durchzuführen, ohne die Gesamtleistung des Systems zu beeinträchtigen.

Neue Trends verändern jedoch die Art und Weise, wie dies getan wird. ETL spielt weiterhin eine Rolle bei der Übertragung von Daten an Data Warehouses und Data Lakes.

Heutzutage kann dies mit Streams in einer Event-Stream-Architektur mithilfe von Spring Cloud Data Flow erfolgen .

3. Spring Cloud-Datenfluss

Mit Spring Cloud Data Flow (SCDF) können Entwickler Datenpipelines in zwei Varianten erstellen:

  • Langlebige Echtzeit-Stream-Anwendungen mit Spring Cloud Stream
  • Kurzlebige Batch-Task-Anwendungen mit Spring Cloud Task

In diesem Artikel behandeln wir die erste, langlebige Streaming-Anwendung, die auf Spring Cloud Stream basiert.

3.1. Spring Cloud Stream-Anwendungen

Die SCDF Stream-Pipelines bestehen aus Schritten, wobei jeder Schritt eine Anwendung ist, die im Spring Boot-Stil unter Verwendung des Spring Cloud Stream-Mikroframeworks erstellt wurde. Diese Anwendungen werden von einer Messaging-Middleware wie Apache Kafka oder RabbitMQ integriert.

Diese Anwendungen sind in Quellen, Prozessoren und Senken unterteilt. Im Vergleich zum ETL-Prozess können wir sagen, dass die Quelle der „Extrakt“ ist, der Prozessor der „Transformator“ und die Senke der „Last“ -Teil.

In einigen Fällen können wir einen Anwendungsstarter in einem oder mehreren Schritten der Pipeline verwenden. Dies bedeutet, dass wir für einen Schritt keine neue Anwendung implementieren müssen, sondern stattdessen einen bereits implementierten vorhandenen Anwendungsstarter konfigurieren müssen.

Eine Liste der Anwendungsstarter finden Sie hier.

3.2. Spring Cloud Datenflussserver

Der letzte Teil der Architektur ist der Spring Cloud Data Flow Server . Der SCDF-Server führt die Bereitstellung der Anwendungen und des Pipeline-Streams mithilfe der Spring Cloud Deployer-Spezifikation durch. Diese Spezifikation unterstützt die SCDF-Cloud-native Version, indem sie für eine Reihe moderner Laufzeiten wie Kubernetes, Apache Mesos, Yarn und Cloud Foundry bereitgestellt wird.

Wir können den Stream auch als lokale Bereitstellung ausführen.

Weitere Informationen zur SCDF-Architektur finden Sie hier.

4. Umgebung einrichten

Bevor wir beginnen, müssen wir die Teile dieser komplexen Bereitstellung auswählen . Das erste zu definierende Stück ist der SCDF-Server.

Zum Testen verwenden wir SCDF Server Local für die lokale Entwicklung . Für die Produktionsbereitstellung können wir später eine Cloud-native Laufzeit wie SCDF Server Kubernetes auswählen. Die Liste der Serverlaufzeiten finden Sie hier.

Lassen Sie uns nun die Systemanforderungen überprüfen, um diesen Server auszuführen.

4.1. System Anforderungen

Um den SCDF-Server auszuführen, müssen zwei Abhängigkeiten definiert und eingerichtet werden:

  • die Messaging-Middleware und
  • das RDBMS.

Für die Messaging-Middleware arbeiten wir mit RabbitMQ und wählen PostgreSQL als RDBMS zum Speichern unserer Pipeline-Stream-Definitionen.

For running RabbitMQ, download the latest version here and start a RabbitMQ instance using the default configuration or run the following Docker command:

docker run --name dataflow-rabbit -p 15672:15672 -p 5672:5672 -d rabbitmq:3-management

As the last setup step, install and run the PostgreSQL RDBMS on the default port 5432. After this, create a database where SCDF can store its stream definitions using the following script:

CREATE DATABASE dataflow;

4.2. Spring Cloud Data Flow Server Local

For running the SCDF Server Local, we can choose to start the server using docker-compose, or we can start it as a Java application.

Here, we'll run the SCDF Server Local as a Java application. For configuring the application, we have to define the configuration as Java application parameters. We'll need Java 8 in the System path.

To host the jars and dependencies, we need to create a home folder for our SCDF Server and download the SCDF Server Local distribution into this folder. You can download the most recent distribution of SCDF Server Local here.

Also, we need to create a lib folder and put a JDBC driver there. The latest version of the PostgreSQL driver is available here.

Finally, let's run the SCDF local server:

$java -Dloader.path=lib -jar spring-cloud-dataflow-server-local-1.6.3.RELEASE.jar \ --spring.datasource.url=jdbc:postgresql://127.0.0.1:5432/dataflow \ --spring.datasource.username=postgres_username \ --spring.datasource.password=postgres_password \ --spring.datasource.driver-class-name=org.postgresql.Driver \ --spring.rabbitmq.host=127.0.0.1 \ --spring.rabbitmq.port=5672 \ --spring.rabbitmq.username=guest \ --spring.rabbitmq.password=guest

We can check if it's running by looking at this URL:

//localhost:9393/dashboard

4.3. Spring Cloud Data Flow Shell

The SCDF Shell is a command line tool that makes it easy to compose and deploy our applications and pipelines. These Shell commands run over the Spring Cloud Data Flow Server REST API.

Download the latest version of the jar into your SCDF home folder, available here. Once it is done, run the following command (update the version as needed):

$ java -jar spring-cloud-dataflow-shell-1.6.3.RELEASE.jar ____ ____ _ __ / ___| _ __ _ __(_)_ __ __ _ / ___| | ___ _ _ __| | \___ \| '_ \| '__| | '_ \ / _` | | | | |/ _ \| | | |/ _` | ___) | |_) | | | | | | | (_| | | |___| | (_) | |_| | (_| | |____/| .__/|_| |_|_| |_|\__, | \____|_|\___/ \__,_|\__,_| ____ |_| _ __|___/ __________ | _ \ __ _| |_ __ _ | ___| | _____ __ \ \ \ \ \ \ | | | |/ _` | __/ _` | | |_ | |/ _ \ \ /\ / / \ \ \ \ \ \ | |_| | (_| | || (_| | | _| | | (_) \ V V / / / / / / / |____/ \__,_|\__\__,_| |_| |_|\___/ \_/\_/ /_/_/_/_/_/ Welcome to the Spring Cloud Data Flow shell. For assistance hit TAB or type "help". dataflow:>

If instead of “dataflow:>” you get “server-unknown:>” in the last line, you are not running the SCDF Server at localhost. In this case, run the following command to connect to another host:

server-unknown:>dataflow config server //{host}

Now, Shell is connected to the SCDF Server, and we can run our commands.

The first thing we need to do in Shell is to import the application starters. Find the latest version here for RabbitMQ+Maven in Spring Boot 2.0.x, and run the following command (again, update the version, here “Darwin-SR1“, as needed):

$ dataflow:>app import --uri //bit.ly/Darwin-SR1-stream-applications-rabbit-maven

For checking the installed applications run the following Shell command:

$ dataflow:> app list

As a result, we should see a table containing all the installed applications.

Also, SCDF offers a graphical interface, named Flo, that we can access by this address: //localhost:9393/dashboard. However, its use isn't in the scope of this article.

5. Composing an ETL Pipeline

Let's now create our stream pipeline. For doing this, we'll use the JDBC Source application starter to extract information from our relational database.

Also, we'll create a custom processor for transforming the information structure and a custom sink to load our data into a MongoDB.

5.1. Extract – Preparing a Relational Database for Extraction

Let's create a database with the name of crm and a table with the name of customer:

CREATE DATABASE crm;
CREATE TABLE customer ( id bigint NOT NULL, imported boolean DEFAULT false, customer_name character varying(50), PRIMARY KEY(id) )

Note that we're using a flag imported, which will store which record has already been imported. We could also store this information in another table, if necessary.

Now, let's insert some data:

INSERT INTO customer(id, customer_name, imported) VALUES (1, 'John Doe', false);

5.2. Transform – Mapping JDBC Fields to the MongoDB Fields Structure

For the transformation step, we'll do a simple translation of the field customer_name from the source table, to a new field name. Other transformations could be done here, but let's keep the example short.

To do this, we'll create a new project with the name customer-transform. The easiest way to do this is by using the Spring Initializr site to create the project. After reaching the website, choose a Group and an Artifact name. We'll use com.customer and customer-transform, respectively.

Once this is done, click on the button “Generate Project” to download the project. Then, unzip the project and import it into your favorite IDE, and add the following dependency to the pom.xml:

 org.springframework.cloud spring-cloud-stream-binder-rabbit 

Now we're set to start coding the field name conversion. To do this, we'll create the Customer class to act as an adapter. This class will receive the customer_name via the setName() method and will output its value via getName method.

The @JsonProperty annotations will do the transformation while deserializing from JSON to Java:

public class Customer { private Long id; private String name; @JsonProperty("customer_name") public void setName(String name) { this.name = name; } @JsonProperty("name") public String getName() { return name; } // Getters and Setters }

The processor needs to receive data from an input, do the transformation and bind the outcome to an output channel. Let's create a class to do this:

import org.springframework.cloud.stream.annotation.EnableBinding; import org.springframework.cloud.stream.messaging.Processor; import org.springframework.integration.annotation.Transformer; @EnableBinding(Processor.class) public class CustomerProcessorConfiguration { @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT) public Customer convertToPojo(Customer payload) { return payload; } }

In the above code, we can observe that the transformation occurs automatically. The input receives the data as JSON and Jackson deserialize it into a Customer object using the set methods.

The opposite is for the output, the data is serialized to JSON using the get methods.

5.3. Load – Sink in MongoDB

Similarly to the transform step, we'll create another maven project, now with the name customer-mongodb-sink. Again, access the Spring Initializr, for the Group choose com.customer, and for the Artifact choose customer-mongodb-sink. Then, type MongoDB in the dependencies search box and download the project.

Next, unzip and import it to your favorite IDE.

Then, add the same extra dependency as in the customer-transform project.

Now we'll create another Customer class, for receiving input in this step:

import org.springframework.data.mongodb.core.mapping.Document; @Document(collection="customer") public class Customer { private Long id; private String name; // Getters and Setters }

For sinking the Customer, we'll create a Listener class that will save the customer entity using the CustomerRepository:

@EnableBinding(Sink.class) public class CustomerListener { @Autowired private CustomerRepository repository; @StreamListener(Sink.INPUT) public void save(Customer customer) { repository.save(customer); } }

And the CustomerRepository, in this case, is a MongoRepository from Spring Data:

import org.springframework.data.mongodb.repository.MongoRepository; import org.springframework.stereotype.Repository; @Repository public interface CustomerRepository extends MongoRepository { } 

5.4. Stream Definition

Now, both custom applications are ready to be registered on SCDF Server. To accomplish this, compile both projects using the Maven command mvn install.

We then register them using the Spring Cloud Data Flow Shell:

app register --name customer-transform --type processor --uri maven://com.customer:customer-transform:0.0.1-SNAPSHOT
app register --name customer-mongodb-sink --type sink --uri maven://com.customer:customer-mongodb-sink:jar:0.0.1-SNAPSHOT

Finally, let's check if the applications are stored at SCDF, run the application list command in the shell:

app list

As a result, we should see both applications in the resulting table.

5.4.1. Stream Pipeline Domain-Specific Language – DSL

A DSL defines the configuration and data flow between the applications. The SCDF DSL is simple. In the first word, we define the name of the application, followed by the configurations.

Also, the syntax is a Unix-inspired Pipeline syntax, that uses vertical bars, also known as “pipes”, to connect multiple applications:

http --port=8181 | log

This creates an HTTP application served in port 8181 which sends any received body payload to a log.

Now, let's see how to create the DSL stream definition of the JDBC Source.

5.4.2. JDBC Source Stream Definition

The key configurations for the JDBC Source are query and update.query will select unread records while update will change a flag to prevent the current records from being reread.

Also, we'll define the JDBC Source to poll in a fixed delay of 30 seconds and polling maximum 1000 rows. Finally, we'll define the configurations of connection, like driver, username, password and connection URL:

jdbc  --query='SELECT id, customer_name FROM public.customer WHERE imported = false' --update='UPDATE public.customer SET imported = true WHERE id in (:id)' --max-rows-per-poll=1000 --fixed-delay=30 --time-unit=SECONDS --driver-class-name=org.postgresql.Driver --url=jdbc:postgresql://localhost:5432/crm --username=postgres --password=postgres

More JDBC Source configuration properties can be found here.

5.4.3. Customer MongoDB Sink Stream Definition

As we didn't define the connection configurations in application.properties of customer-mongodb-sink, we'll configure through DSL parameters.

Our application is fully based on the MongoDataAutoConfiguration. You can check out the other possible configurations here. Basically, we'll define the spring.data.mongodb.uri:

customer-mongodb-sink --spring.data.mongodb.uri=mongodb://localhost/main

5.4.4. Create and Deploy the Stream

First, to create the final stream definition, go back to the Shell and execute the following command (without line breaks, they have just been inserted for readability):

stream create --name jdbc-to-mongodb --definition "jdbc --query='SELECT id, customer_name FROM public.customer WHERE imported=false' --fixed-delay=30 --max-rows-per-poll=1000 --update='UPDATE customer SET imported=true WHERE id in (:id)' --time-unit=SECONDS --password=postgres --driver-class-name=org.postgresql.Driver --username=postgres --url=jdbc:postgresql://localhost:5432/crm | customer-transform | customer-mongodb-sink --spring.data.mongodb.uri=mongodb://localhost/main" 

This stream DSL defines a stream named jdbc-to-mongodb. Next, we'll deploy the stream by its name:

stream deploy --name jdbc-to-mongodb 

Finally, we should see the locations of all available logs in the log output:

Logs will be in {PATH_TO_LOG}/spring-cloud-deployer/jdbc-to-mongodb/jdbc-to-mongodb.customer-mongodb-sink Logs will be in {PATH_TO_LOG}/spring-cloud-deployer/jdbc-to-mongodb/jdbc-to-mongodb.customer-transform Logs will be in {PATH_TO_LOG}/spring-cloud-deployer/jdbc-to-mongodb/jdbc-to-mongodb.jdbc

6. Conclusion

In this article, we've seen a full example of an ETL data pipeline using Spring Cloud Data Flow.

Am bemerkenswertesten ist, dass wir die Konfigurationen eines Anwendungsstarters gesehen, eine ETL-Stream-Pipeline mithilfe der Spring Cloud Data Flow Shell erstellt und benutzerdefinierte Anwendungen für das Lesen, Transformieren und Schreiben von Daten implementiert haben.

Der Beispielcode befindet sich wie immer im GitHub-Projekt.