Flink CDC介绍及KADB支持_flink cdc opensearch

<dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-connector-jdbc</artifactId>

    <version>3.1.0-1.18</version>

</dependency>

CREATE TABLE MyUserTable (

  id BIGINT,

  name STRING,

  age INT,

  status BOOLEAN,

  PRIMARY KEY (id) NOT ENFORCED

) WITH (

   'connector' = 'jdbc',

   'url' = 'jdbc:postgresql://dbhost:5432/postgresdb’

   'table-name' = 'users'

);

INSERT INTO MyUserTable

SELECT id, name, age, status FROM T;

-- scan data from the JDBC table

SELECT id, name, age, status FROM MyUserTable;

<dependency>

    <groupId>org.apache.flink</groupId>

    <artifactId>flink-connector-jdbc</artifactId>

    <version>3.1.0-1.18</version>

</dependency>

public class JdbcSinkExample {

    static class Book {

        public Book(Long id, String title, String authors, Integer year) {

            this.id = id;

            this.title = title;

            this.authors = authors;

            this.year = year;

        }

        final Long id;

        final String title;

        final String authors;

        final Integer year;

    }

    public static void main(String[] args) throws Exception {

        var env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(

                new Book(101L, "Stream Processing with Apache Flink", "Fabian Hueske, Vasiliki Kalavri", 2019),

                new Book(102L, "Streaming Systems", "Tyler Akidau, Slava Chernyak, Reuven Lax", 2018),

                new Book(103L, "Designing Data-Intensive Applications", "Martin Kleppmann", 2017),

                new Book(104L, "Kafka: The Definitive Guide", "Gwen Shapira, Neha Narkhede, Todd Palino", 2017)

        ).addSink(

                JdbcSink.sink(

                        "insert into books (id, title, authors, year) values (?, ?, ?, ?)",

                        (statement, book) -> {

                            statement.setLong(1, book.id);

                            statement.setString(2, book.title);

                            statement.setString(3, book.authors);

                            statement.setInt(4, book.year);

                        },

                        JdbcExecutionOptions.builder()

                                .withBatchSize(1000)

                                .withBatchIntervalMs(200)

                                .withMaxRetries(5)

                                .build(),

                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()

                                .withUrl("jdbc:postgresql://dbhost:5432/postgresdb")

                                .withDriverName("org.postgresql.Driver")

                                .withUsername("someUser")

                                .withPassword("somePassword")

                                .build()

                ));

        env.execute();

    }

}

-- MySQL

CREATE DATABASE mydb;

USE mydb;

CREATE TABLE products (

  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,

  name VARCHAR(255) NOT NULL,

  description VARCHAR(512)

);

ALTER TABLE products AUTO_INCREMENT = 101;

INSERT INTO products

VALUES (default,"scooter","Small 2-wheel scooter"),

       (default,"car battery","12V car battery"),

       (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),

       (default,"hammer","12oz carpenter's hammer"),

       (default,"hammer","14oz carpenter's hammer"),

       (default,"hammer","16oz carpenter's hammer"),

       (default,"rocks","box of assorted rocks"),

       (default,"jacket","water resistent black wind breaker"),

       (default,"spare tire","24 inch spare tire");

CREATE TABLE orders (

  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,

  order_date DATETIME NOT NULL,

  customer_name VARCHAR(255) NOT NULL,

  price DECIMAL(10, 5) NOT NULL,

  product_id INTEGER NOT NULL,

  order_status BOOLEAN NOT NULL -- Whether order has been placed

) AUTO_INCREMENT = 10001;

INSERT INTO orders

VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),

       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),

       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

-- PG

CREATE TABLE shipments (

  shipment_id SERIAL NOT NULL PRIMARY KEY,

  order_id SERIAL NOT NULL,

  origin VARCHAR(255) NOT NULL,

  destination VARCHAR(255) NOT NULL,

  is_arrived BOOLEAN NOT NULL

);

ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001;

ALTER TABLE public.shipments REPLICA IDENTITY FULL;

INSERT INTO shipments

VALUES (default,10001,'Beijing','Shanghai',false),

       (default,10002,'Hangzhou','Shanghai',false),

       (default,10003,'Shanghai','Hangzhou',false);

./bin/sql-client.sh

-- Flink SQL                  

Flink SQL> SET execution.checkpointing.interval = 3s;

- Flink SQL

Flink SQL> CREATE TABLE products (

    id INT,

    name STRING,

    description STRING,

    PRIMARY KEY (id) NOT ENFORCED

  ) WITH (

    'connector' = 'mysql-cdc',

    'hostname' = 'localhost',

    'port' = '3306',

    'username' = 'root',

    'password' = '123456',

    'database-name' = 'mydb',

    'table-name' = 'products'

  );

Flink SQL> CREATE TABLE orders (

   order_id INT,

   order_date TIMESTAMP(0),

   customer_name STRING,

   price DECIMAL(10, 5),

   product_id INT,

   order_status BOOLEAN,

   PRIMARY KEY (order_id) NOT ENFORCED

 ) WITH (

   'connector' = 'mysql-cdc',

   'hostname' = 'localhost',

   'port' = '3306',

   'username' = 'root',

   'password' = '123456',

   'database-name' = 'mydb',

   'table-name' = 'orders'

 );

Flink SQL> CREATE TABLE shipments (

   shipment_id INT,

   order_id INT,

   origin STRING,

   destination STRING,

   is_arrived BOOLEAN,

   PRIMARY KEY (shipment_id) NOT ENFORCED

 ) WITH (

   'connector' = 'postgres-cdc',

   'hostname' = 'localhost',

   'port' = '5432',

   'username' = 'postgres',

   'password' = 'postgres',

   'database-name' = 'postgres',

   'schema-name' = 'public',

   'table-name' = 'shipments'

 );

-- Flink SQL

Flink SQL> CREATE TABLE enriched_orders (

   order_id INT,

   order_date TIMESTAMP(0),

   customer_name STRING,

   price DECIMAL(10, 5),

   product_id INT,

   order_status BOOLEAN,

   product_name STRING,

   product_description STRING,

   shipment_id INT,

   origin STRING,

   destination STRING,

   is_arrived BOOLEAN,

   PRIMARY KEY (order_id) NOT ENFORCED

 ) WITH (

    'connector' = 'jdbc',

   'url' = 'jdbc:postgresql://dbhost:5432/postgresdb’

   'table-name' = 'enriched_orders'

 );

-- Flink SQL

Flink SQL> INSERT INTO enriched_orders

 SELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrived

 FROM orders AS o

 LEFT JOIN products AS p ON o.product_id = p.id

 LEFT JOIN shipments AS s ON o.order_id = s.order_id;

--MySQL

INSERT INTO orders

VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);

--PG

INSERT INTO shipments

VALUES (default,10004,'Shanghai','Beijing',false);

--MySQL

UPDATE orders SET order_status = true WHERE order_id = 10004;

--PG

UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;

--MySQL

DELETE FROM orders WHERE order_id = 10004;