Flink CDC (MySQL as an example)

background

There are often some large tables of tens of millions or even billions in business , At this time, it may be considered to divide the database and table （Sharding-JDBC、MyCat And so on ）, Also often synchronize data entry ES in ; In the business scenario of synchronizing data ,Flink CDC It's a good solution .

programme

Such as mysql、postgresql、sqlserver etc. ,flink cdc By reading the binlog journal （ Be careful ： Please open first. binlog journal ）, Data synchronization , Good real-time performance .

The analysis and consumption of data are encapsulated twice , Users only need to add simple configuration , Realization FlinkConsumerListener Interface , Just focus on writing business code .

Code

show coding

flink: flink cdc For the time being mysql

test demo

Create a springboot project

Depend on the introduction of （ Introduce the dependencies after the packaging of the above projects ）

        <dependency>
            <groupId>com.kwin</groupId>
            <artifactId>flink</artifactId>
            <version>0.0.1-SNAPSHOT</version>
        </dependency>

The configuration file

flink:
  pipeline-name: flinkCDCTest
  mysqlDataSource:
        - port: 3306
          hostname: 127.0.0.1
          databaseList:
            - flinktest
          tableList:
            - flinktest.student
          username: root
          password: 123456

Above , in the light of flinktest Database student table binlog monitor .

flinktest.student The consumer

student Entity

import lombok.Data;

/**
 * @author kwin
 * @Date 2022/7/25 18:27
 **/
@Data
public class Student {
    private Long id;

    private String name;

    private Integer age;

    private Integer maxInx;
}

  consumer

import com.kwin.demo.server.module.flink.test.entity.Student;
import com.kwin.flink.sink.FlinkConsumerListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author kwin
 * @Date 2022/7/25 18:29
 **/
@Slf4j
@Component
public class StudentConsumerListener implements FlinkConsumerListener<Student> {
    @Override
    public String getDBName() {
        return "flinktest";
    }

    @Override
    public String getTable() {
        return "student";
    }

    @Override
    public void insert(Student data) {
        System.out.println("insert: " + data);
    }

    @Override
    public void update(Student srcData, Student destData) {
        System.out.println("update: \nsrc:" + srcData + "\ndest:" + destData);
    }

    @Override
    public void delete(Student data) {
        System.out.println("delete:"+data);
    }
}

Start project

flinktest.student When modifying data ：

flinktest.student When inserting data ：

 flinktest.student When deleting data ：

  Above , Users only need to implement FlinkConsumerListener Interface , You can perform consumption and business logic operations on the data of the specified table .