1.一文搞定Flink自定义JDBC

1.前言

本论坛发送的所有内容，都是笔者在自己的个人的笔记上优化后誊写而来，希望自己所拥有的知识能够帮助更多有学习欲望的人。但值得一提的是，由于本人接触行业时间有限，可能会出现一些技术上的纰漏，如果有问题欢迎私信、评论指出，大家共同进步。

---

2.简单介绍

在Flink对数据进行计算的时候，一般会按照阶段的不同，将处理过程分为 source->transform->sink
借此来完成数据从读取到计算再到写出的全过程。
本章节当中要介绍的FlinkJDBC其实就是Sink阶段的成员之一，它能够帮助Flink达成从数据流到存储介质保存的全过程（存储介质需要支持JDBC）。如果SINK方的这个存储介质支持XA事务的话，那么FlinkJDBC还能够对其提供精准一次性语义。

---

3.FlinkJDBC使用

3.1 引入依赖

<dependency>

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

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

    <version>1.15.0</version>

</dependency>

3.2 直接指定JDBCSink即可

这里以ClickHouse为例->本代码可直接粘贴使用,因为是在文档中手写的，没有用编辑器，所以可能会有错别单词.

//1.声明静态方法

public static <T> SinkFunction<T> getJdbcSink(String sql){

    return JdbcSink.<T>sink(

        sql,

        new JdbcStatementBuilder<T>(){

            //这个方法主要是完成对sql语句中的数据内容对PreparedStatement对象中占位符的赋值

            @Overwrite

            public void accept(PreparedStatement preparedStatement,T obj) throws SQLException{

                    //通过反射来完成赋值，本段代码结束之后有关于这部分内容反射相关知识的介绍

                    Field[] declaredFields = obj.getClass().getDeclaredFields();

                    for(int i=0; i<declaredFields.length; i++){

                        Filed declaredField = declaredFields[i];

                        declaredField.setAccessible(true);

                        try{

                            Object value = declaredField.get(obj);

                            preparedStatement.setObject(i,value);

                            }catch(IllegalAccessException e){

                                e.printStackTrace();

                            }

                        }

                }

            },

        JdbcExecutionOptions.bulider()

        .withBatchIntervalMs(5000L) //指定多长时间发送一次

        .withBatchSize(5) //指定攒够多少条数发送一次

        .build(),

        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()

        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver") //Drvier

        .withUrl("jdbc:clickhouse://hadoop102:8123/table") //url

        .build()

    );   

}

---

4.反射相关知识的描述

反射机制对于我来说，实际上就是一种能够为使用者提供针对未知对象或者未知类来进行内容读取的一个功能。这是我个人对于反射的浅显理解，如果有错误欢迎指正。接下来我就要用我所理解的内容，编写一个简单的例子，来解释3.2程序段中accept方法是如何完成实体类对占位符进行赋值的过程。

4.1 反射的小例子

思路：主程序想要通过对一个方法传入不同的实体类，来获得所有实体类中的所有属性的字段信息。

准备：主程序（用来调用方法）、两个不同的实体类（用来对公共方法做验证）、泛型方法（输出实体类中的字段信息）

//实体类1

@Data

@AllArgsConstructor

public class Student {

//用来表示学生信息

    private String name;

    private String banji;

    private String score;

}

//实体类2

@Data

@AllArgsConstructor

public class Teacher {

    //用来表示老师信息

    private String dept;

    private String classHeader;

}

//泛型方法

import java.lang.reflect.Field;

import java.util.ArrayList;

import java.util.List;

public class ReadInfoMethod {

    //通过传入参数，然后循环的将作为参数的对象中的所有属性的值信息添加到数组中

    public static <T> List<T> getInfo(T t){

        Class<?> aClass = t.getClass();

        Field[] declaredFields = aClass.getDeclaredFields();

        ArrayList<T> result = new ArrayList<T>();

        for (Field declaredField : declaredFields) {

            int num = 0;

            declaredField.setAccessible(true);

            try {

                T o = (T)declaredField.get(t);

                result.add(num,o);

                num++;

            }catch (IllegalAccessException e){

                e.printStackTrace();

            }

        }

        return  result;

    }

}

//主程序

public class test {

    public static void main(String[] args) {

        Student student = new Student("弗林克", "三年二班", "95");

        Teacher teacher = new Teacher("办公室部门", "三年二班班主任");

        //调用泛型方法，获得传入对象的所有属性字段的值信息的列表

        List<Student> info = ReadInfoMethod.getInfo(student);

        List<Teacher> teachers = ReadInfoMethod.getInfo(teacher);

        System.out.println(info);

        System.out.println(teachers);

    }

}

泛型方法中调用的方法的方式，与FlinkJDBC中的accept方法中的内容如出一辙。二者在表现形式上的区别就是accept方法在对占位符进行赋值的时候，需要指定对应字段的索引位置，因此 才有了preparedStatement.setObject(i,value);的这种方式。

Flink官网中针对这部分内容进行描述的地址是：https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/datastream/jdbc/