实时数仓（四）业务数据从ods到dwd中自定义sink

package com.yyds.app.function;

import com.alibaba.fastjson.JSONObject;
import com.yyds.common.FlinkConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Set;



public class DimSinkFunction extends RichSinkFunction {


    private Connection connection;

    // 初始化连接
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName(FlinkConfig.PHOENIX_DRIVER);
        connection = DriverManager.getConnection(
                FlinkConfig.PHOENIX_SERVER
        );
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // value = {"sinkTable":"", "database":"","before":{},"after":{},"type":"insert","tableName":""}
        // phoenix sql语句
        // upsert into db.tn (id,name) values('...','...')
        PreparedStatement preparedStatement = null;

        try {
            //1、获取sql
            String upsertSql = getUpsertSql(
                    value.getString("sinkTable"),
                    value.getJSONObject("after")
            );
            System.out.println("phoenix sql ===> " + upsertSql);
            // 2、预编译sql语句
            preparedStatement = connection.prepareStatement(upsertSql);
            // 3、执行插入操作
            preparedStatement.executeUpdate();

            connection.commit();
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            if(preparedStatement != null){
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    // upsert into db.tn (id,name) values('...','...')
    private String getUpsertSql(String sinkTable, JSONObject after) {
        String sql = " upsert into " + FlinkConfig.Hbase_SCHEMA + "." + sinkTable + " ( " ;
        Set keySet = after.keySet();
        Collection