Spring Cloudデータフローを使用したETL

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

CREATE DATABASE dataflow;

$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

$ 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:>

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

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

$ dataflow:> app list

CREATE DATABASE crm;

CREATE TABLE customer (
    id bigint NOT NULL,
    imported boolean DEFAULT false,
    customer__name character varying(50),
    PRIMARY KEY(id)
)

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

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
</dependency>

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
}

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;
    }
}

import org.springframework.data.mongodb.core.mapping.Document;

@Document(collection="customer")
public class Customer {

    private Long id;
    private String name;

   //Getters and Setters
}

@EnableBinding(Sink.class)
public class CustomerListener {

    @Autowired
    private CustomerRepository repository;

    @StreamListener(Sink.INPUT)
    public void save(Customer customer) {
        repository.save(customer);
    }
}

import org.springframework.data.mongodb.repository.MongoRepository;
import org.springframework.stereotype.Repository;

@Repository
public interface CustomerRepository extends MongoRepository<Customer, Long> {

}

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

app list

http --port=8181 | log

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

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

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"

stream deploy --name jdbc-to-mongodb

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