ETL con Spring Cloud Data Flow

1. Panoramica

Spring Cloud Data Flow è un toolkit nativo del cloud per la creazione di pipeline di dati in tempo reale e processi batch. Spring Cloud Data Flow è pronto per essere utilizzato per una serie di casi d'uso di elaborazione dati come semplice importazione / esportazione, elaborazione ETL, streaming di eventi e analisi predittiva.

In questo tutorial, impareremo un esempio di Extract Transform and Load (ETL) in tempo reale utilizzando una pipeline di flusso che estrae i dati da un database JDBC, li trasforma in semplici POJO e li carica in MongoDB.

2. Elaborazione ETL ed Event-Stream

ETL (estrazione, trasformazione e caricamento) era comunemente indicato come un processo che carica in batch i dati da diversi database e sistemi in un data warehouse comune. In questo data warehouse, è possibile eseguire pesanti elaborazioni di analisi dei dati senza compromettere le prestazioni complessive del sistema.

Tuttavia, le nuove tendenze stanno cambiando il modo in cui ciò viene fatto. ETL ha ancora un ruolo nel trasferimento dei dati a data warehouse e data lake.

Al giorno d'oggi questo può essere fatto con i flussi in un'architettura di flusso di eventi con l'aiuto di Spring Cloud Data Flow .

3. Flusso di dati Spring Cloud

Con Spring Cloud Data Flow (SCDF), gli sviluppatori possono creare pipeline di dati in due versioni:

  • Applicazioni di streaming in tempo reale di lunga durata che utilizzano Spring Cloud Stream
  • Applicazioni di attività in batch di breve durata utilizzando Spring Cloud Task

In questo articolo tratteremo la prima, un'applicazione di streaming di lunga durata basata su Spring Cloud Stream.

3.1. Applicazioni Spring Cloud Stream

Le pipeline SCDF Stream sono composte da passaggi, in cui ogni passaggio è un'applicazione costruita in stile Spring Boot utilizzando il micro framework Spring Cloud Stream. Queste applicazioni sono integrate da un middleware di messaggistica come Apache Kafka o RabbitMQ.

Queste applicazioni sono classificate in sorgenti, processori e sink. Confrontando il processo ETL, potremmo dire che la sorgente è l '"estratto", il processore è il "trasformatore" e il sink è la parte di "carico".

In alcuni casi, possiamo utilizzare uno starter dell'applicazione in uno o più passaggi della pipeline. Ciò significa che non avremmo bisogno di implementare una nuova applicazione per un passaggio, ma invece di configurare uno starter dell'applicazione esistente già implementato.

Un elenco di applicazioni per l'avvio può essere trovato qui.

3.2. Spring Cloud Data Flow Server

L'ultimo pezzo dell'architettura è Spring Cloud Data Flow Server . Il server SCDF esegue la distribuzione delle applicazioni e del flusso della pipeline utilizzando la specifica Spring Cloud Deployer. Questa specifica supporta la versione cloud-native di SCDF distribuendola a una gamma di runtime moderni, come Kubernetes, Apache Mesos, Yarn e Cloud Foundry.

Inoltre, possiamo eseguire il flusso come distribuzione locale.

Ulteriori informazioni sull'architettura SCDF sono disponibili qui.

4. Configurazione dell'ambiente

Prima di iniziare, dobbiamo scegliere i pezzi di questa complessa distribuzione . Il primo pezzo da definire è il server SCDF.

Per i test, utilizzeremo SCDF Server Local per lo sviluppo locale . Per la distribuzione in produzione, possiamo successivamente scegliere un runtime nativo del cloud, come SCDF Server Kubernetes. Possiamo trovare l'elenco dei tempi di esecuzione del server qui.

Ora, controlliamo i requisiti di sistema per eseguire questo server.

4.1. Requisiti di sistema

Per eseguire il server SCDF, dovremo definire e impostare due dipendenze:

  • il middleware di messaggistica e
  • l'RDBMS.

Per il middleware di messaggistica, lavoreremo con RabbitMQ e scegliamo PostgreSQL come RDBMS per l'archiviazione delle definizioni del flusso della pipeline.

Per eseguire RabbitMQ, scarica l'ultima versione qui e avvia un'istanza RabbitMQ utilizzando la configurazione predefinita o esegui il seguente comando Docker:

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

Come ultimo passaggio di configurazione, installa ed esegui PostgreSQL RDBMS sulla porta predefinita 5432. Successivamente, crea un database in cui SCDF possa memorizzare le sue definizioni di flusso utilizzando il seguente script:

CREATE DATABASE dataflow;

4.2. Spring Cloud Data Flow Server locale

Per eseguire SCDF Server Local, possiamo scegliere di avviare il server utilizzando docker-compose , oppure possiamo avviarlo come applicazione Java.

Qui, eseguiremo SCDF Server Local come un'applicazione Java. Per configurare l'applicazione, dobbiamo definire la configurazione come parametri dell'applicazione Java. Avremo bisogno di Java 8 nel percorso di sistema.

Per ospitare i jar e le dipendenze, dobbiamo creare una cartella home per il nostro server SCDF e scaricare la distribuzione locale del server SCDF in questa cartella. È possibile scaricare la distribuzione più recente di SCDF Server Local qui.

Inoltre, dobbiamo creare una cartella lib e inserire lì un driver JDBC. L'ultima versione del driver PostgreSQL è disponibile qui.

Infine, eseguiamo il server locale SCDF:

$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

Possiamo verificare se è in esecuzione guardando questo URL:

// localhost: 9393 / dashboard

4.3. Shell di flusso di dati Spring Cloud

La shell SCDF è uno strumento a riga di comando che semplifica la composizione e la distribuzione delle nostre applicazioni e pipeline . Questi comandi Shell vengono eseguiti sull'API REST di Spring Cloud Data Flow Server.

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.

Più degno di nota, abbiamo visto le configurazioni di uno starter dell'applicazione, creato una pipeline di flusso ETL utilizzando Spring Cloud Data Flow Shell e implementato applicazioni personalizzate per la nostra lettura, trasformazione e scrittura dei dati.

Come sempre, il codice di esempio può essere trovato nel progetto GitHub.