将Flink计算完毕后的数据Sink到Nebula

import lombok.val;

import lombok.var;

/**
 * Nebula工具类
 */
public class NebulaUtil {
    // Nebula会话
    private Session session = null;
    // Nebula连接池
    private NebulaPool pool = new NebulaPool();/**
     * 获得Nebula工具类单例
     *
     * @return NebulaUtil
     */
    public static NebulaUtil geTinstance() {
        return NebulaUtilHolder.instance;
    }

    /**
     * 执行NGQL
     *
     * @param nGQL NGQL
     * @return 返回执行结果
     */
    public ResultSet execute(String nGQL) {
        try {
            if (session != null) {
                return session.execute(nGQL);
            }
        } catch (IOErrorException E) {
            e.printStackTrace();
        } catch (UnsupportedEncodingException E) {
            e.printStackTrace();
        }

        return null;
    }
   /**
     * 释放会话
     */
    public void releaseSession() {
        // 释放连接
        if (session != null) {
            session.release();
        }

        // 关闭连接池
        pool.close();
    }

    private static class NebulaUtilHolder {
        private static final NebulaUtil instance = new NebulaUtil();
    }

    private NebulaUtil() {
        initSession();
    }

    /**
     * 初始化会话
     */
    private void initSession() {// 连接地址，多个间用逗号“,”隔开
        val host = "127.0.0.1";
        val port = 9669;
        val user = "user";
        val password = "password";
        val space = "MySpace";

        var nebulaPoolConfig = new NebulaPoolConfig();
        nebulaPoolConfig.setMaxConnSize(100);

        var hostAddressList = new ArrayList<HostAddress>();

        val hostArray = host.split(",");

        for (val hostAddress : hostArray) {
            hostAddressList.add(new HostAddress(hostAddress, port));
        }

        try {
            pool.init(hostAddressList, nebulaPoolConfig);
        } catch (UnknownHostException E) {
           e.printStackTrace();
        }

        try {
            session = pool.getSession(user, password, false);
        } catch (NotValidConnectionException E) {
            e.printStackTrace();
        } catch (IOErrorException E) {
           e.printStackTrace();
        } catch (AuthFailedException E) {
            e.printStackTrace();
        }

        // 切换图空间
        val resp = execute(String.format("USE %s;", spacE));

        if (resp == null || !resp.isSucceeded()) {
            System.out.println("切换图空间失败!" + spacE);
        }
    }
}

import lombok.val;/**
 * Sink到Nebula数据库
 */
public class NebulaSink extends RichSinkFunction<List<String>> {
    /**
     * 打开连接
     *
     * @param parameters 配置参数
     */
    @Override
    public void open(Configuration parameters) {
    }

    /**
     * 调用
     *
     * @param nGQLList NGQL列表
     * @param context  上下文
     */
    @Override
    public void invoke(List<String> nGQLList, Context context) {
        for (val nGQL : nGQLList) {
            NebulaUtil.geTinstance().execute(nGQL);
        }
    }

    /**
     * 关闭连接
     */
    @Override
    public void close() throws Exception {
        super.close();

        NebulaUtil.geTinstance().releaseSession();
    }
}