flink jdbc写gp，以多value的形式，快速写入

package com.ysservice.dataStreamApi.sink;

import com.ysservice.dataStreamApi.utils.GreenplumUtil;
import com.ysservice.dataStreamApi.utils.RegexUtils;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.types.Row;

import java.lang.reflect.Field;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;


public class GreenPlumSink extends RichSinkFunction>> {

    Connection connection;
    Statement statement;

    //任务开始时执行一次，用于获取Greenplum的链接
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = GreenplumUtil.getConnection();
        statement = connection.createStatement();
    }

    // 每条记录插入时调用一次
    public void invoke(List> value, SinkFunction.Context context) throws Exception {

        if (value.size() > 0) {
            String tableName = value.get(0).f1;
            String insertSql = "insert into "+tableName+" values";
            ArrayList> errorTuple3List = new ArrayList<>();
            for (Tuple3 tuple3 : value) {
                if (tuple3.f0) {
                    errorTuple3List.add(tuple3);
                    insertSql += (insertValue(tuple3.f2.getFieldNames(true), tuple3));
                } else {
                    if (!(("insert into "+tableName+" values").equals(insertSql))) {
                        try {
                            statement.execute(insertSql.substring(0, (insertSql.length() - 1)));
                        } catch (SQLException throwables) {
                            executeErrorSql(errorTuple3List);
                            errorTuple3List.clear();
                        }
                        insertSql = "insert into "+tableName+" values";
                    }
                    String deleteSql = deleteSql(tuple3.f2.getFieldNames(true), tuple3);
                    try {
                        statement.execute(deleteSql);
                    } catch (SQLException throwables) {
                        statement.execute("insert into cdc_log.cdc_error_sql(error_sql) values ('" + deleteSql + "')");
                    }
                }
            }
            if (!(("insert into "+tableName+" values").equals(insertSql))) {
                try {
                    statement.execute(insertSql.substring(0, (insertSql.length() - 1)));
                    errorTuple3List.clear();
                } catch (SQLException throwables) {
                    executeErrorSql(errorTuple3List);
                    errorTuple3List.clear();
                }
            }
        }

    }

    public void executeErrorSql(ArrayList> errorTuple3List) throws Exception {

        if (errorTuple3List.size() > 0) {
            for (Tuple3 tuple3 : errorTuple3List) {
                String deleteSql = deleteSql(tuple3.f2.getFieldNames(true), tuple3);
                try {
                    statement.execute(deleteSql);  //如果执行删除报错，就将错误的sql写入错误日志表
                } catch (SQLException e) {
                    statement.execute("insert into cdc_log.cdc_error_sql(error_sql) values ('" + deleteSql + "')");
                }
                String insertSql = insertSql(tuple3.f2.getFieldNames(true), tuple3);
                try {
                    statement.execute(insertSql);  //如果执行插入报错，就将错误的sql写入错误日志表
                } catch (SQLException e) {
                    statement.execute("insert into cdc_log.cdc_error_sql(error_sql) values ('" + insertSql + "')");
                }
            }
        }
    }

    //sql插入语句
    public String insertValue(Set names, Tuple3 value) {
        String values = "";//用于拼接数据
        for (String name : names) {//遍历字段，获取字段对于的数据后，拼接字段和数据，最终拼成一个完整的sql
            Object data = value.f2.getField(name);//获得字段对应的数据
            if (data != null) {
                if (RegexUtils.numberRegex(data.toString())) {//判断数据类型是否是数字的
                    data = data.toString();
                } else {
                    data = "'" + RegexUtils.timeStampRegex(data.toString()).replace("'", "''") + "'";//拼sql时如果遇到单引号会报错，将一个单引号变成两个就ok了
                }
            } else {
                data = "null";
            }
            values += data + ",";
        }
        //拼接插入的sql
        return "(" + values.substring(0, (values.length() - 1)) + "),";
    }

    //sql插入语句
    public String insertSql(Set names, Tuple3 value) {
        String datas = "";//用于拼接数据
        String fields = "";//用于拼接字段
        for (String name : names) {//遍历字段，获取字段对于的数据后，拼接字段和数据，最终拼成一个完整的sql
            Object data = value.f2.getField(name);//获得字段对应的数据
            if (data != null) {
                if (RegexUtils.numberRegex(data.toString())) {//判断数据类型是否是数字的
                    data = data.toString();
                } else {
                    data = "'" + RegexUtils.timeStampRegex(data.toString()).replace("'", "''") + "'";//拼sql时如果遇到单引号会报错，将一个单引号变成两个就ok了
                }
                datas += data + ",";
                fields += name + ",";
            }
        }
        //拼接插入的sql
        String sql = "insert into " + value.f1 + "(" + fields.substring(0, (fields.length() - 1)) + ") values(" + datas.substring(0, (datas.length() - 1)) + ");";
        return sql;
    }
    //sql删除语句
    public String deleteSql(Set names, Tuple3 value) throws Exception {
        //获得每张表的主键字段，使用主键去删除数据
        Class sourceClass = Class.forName("com.ysservice.dataStreamApi.source." + value.f1);
        Field primaryKey = sourceClass.getDeclaredField("primaryKey");
        String keys = (String) primaryKey.get(sourceClass);
        String[] keyArr = keys.split(",");

        String wheres = "";//用于拼接where sql
        for (String name : keyArr) {
            Object data = value.f2.getField(name);
            if (data != null) {
                if (RegexUtils.numberRegex(data.toString())) {
                    data = data.toString();
                } else {
                    data = "'" + RegexUtils.timeStampRegex(data.toString()).replace("'", "''") + "'";
                }
                wheres += name + "=" + data + " and ";
            }
        }
        //拼接删除的sql
        String sql = "delete from " + value.f1 + " where " + wheres.substring(0, (wheres.length() - 5)) + ";";
        return sql;
    }

    @Override
    public void close() throws Exception {
        super.close();
        if (statement != null) {
            statement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}