flink连接hbase

水深无声 2023-07-12 05:26 9阅读 0赞

flink在1.9之后在table api&sql中支持了hbase的连接器,，且是hbase1.4.3版本。之前版本需要自己实现hbase的连接。  
下面实现一个通用的hbase sink：

@Slf4j
    public class HBaseConnPool implements Serializable { 
    
        private static Configuration conf = null;
    
        //HBase中Connection已经实现了对连接的管理功能,是线程安全的，一个JVM中共用一个Connection对象即可，
        // 而Table和Admin则不是线程安全的，不同的线程中使用单独的Table和Admin对象。
        private static Connection conn = null;
    
        private static final HBaseConnPool instance = new HBaseConnPool();
    
        private HBaseConnPool() { 
            log.info("init HBaseConnPool");
        }
    
        /** * 获取pool实例 * @return HBaseConnPool */
        public static HBaseConnPool getInstance(){ 
            log.info("execute getInstance");
            if(instance == null){ 
                log.error("getInstance error, instance is null.");
            }
            return instance;
        }
    
        public synchronized Connection getConnection() throws Exception { 
            log.info("get connection");
            if (conn == null || conn.isClosed()) { 
                log.info("recreate HBase connection");
                closeConnectionPool();
                createConnectionPool();
            }
            if (conn == null) { 
                log.error("getConnection error, conn is null");
            }
            return conn;
        }
    
        private void createConnectionPool() throws IOException { 
            log.info("create connection pool");
            loadConfiguration();
            conn = ConnectionFactory.createConnection(conf);
        }
    
        private void closeConnectionPool() throws IOException { 
            log.info("close connection pool");
            if (conn == null) { 
                log.info("conn is null, end close");
                return;
            }
            conn.close();
            log.info("end close");
        }
    
        /** * 加载配置文件 */
        private void loadConfiguration(){ 
            conf = HBaseConfiguration.create();
            //获取路径:classes/
            String filePath = this.getClass().getClassLoader().getResource("").getPath();
            String hbaseFilePath = filePath + "hbase" + File.separator;
    
            //读取hbase客户端配置文件
            conf.addResource(new Path(hbaseFilePath + "core-site.xml"));
            conf.addResource(new Path(hbaseFilePath + "hbase-site.xml"));
        }
    }

@Slf4j
    public class HBaseSink extends RichSinkFunction<List<Put>> { 
    
        @Override
        public void open(Configuration parameters) throws Exception { 
            super.open(parameters);
            Admin admin = HBaseConnPool.getInstance().getConnection().getAdmin();
            if (!admin.tableExists(TableName.valueOf("tableName"))) { 
                log.info("create hbase table: tableName");
                HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf("tableName"));
                tableDescriptor.addFamily(new HColumnDescriptor("cf"));
                admin.createTable(tableDescriptor);
            }
            admin.close();
        }
    
        @Override
        public void close() throws Exception { 
            super.close();
        }
    
        @Override
        public void invoke(List<Put> putList, Context context) throws Exception{ 
            try (Table table = HBaseConnPool.getInstance().getConnection().getTable(TableName.valueOf("tableName"))) { 
                table.put(putList);
            } catch (IOException e) { 
                log.error("put error", e);
            }
        }
    }

public class HBaseDemo { 
    
        public static void main(String[] args) { 
            EnvironmentSettings settings = EnvironmentSettings
                    .newInstance()
                    .useBlinkPlanner()
                    .inStreamingMode()
                    .build();
    
            // set up the streaming execution environment
            final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
    
            Properties prop = new Properties();
            prop.put("bootstrap.servers", "localhost:9092");
            prop.put("group.id", "flink-streaming-job");
    
            FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("start_log", new SimpleStringSchema(), prop);
            //source
            DataStream<String> source = env.addSource(consumer);
            DataStream<String> filterDs =  source.filter((FilterFunction<String>) value -> value != null && value.length() > 0);
            /* filterDs.map((MapFunction<String, List<Put>>) value -> { List<Put> list = new ArrayList<>(); String[] args1 = value.split(","); String rowKey = args1[0] + "_" + args1[1]; Put put = new Put(Bytes.toBytes(rowKey)); put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("column0"),Bytes.toBytes(args1[2])); put.addColumn(Bytes.toBytes("cf"),Bytes.toBytes("column1"),Bytes.toBytes(args1[3])); list.add(put); return list; }) .addSink(new HBaseSink()); */
    
            //HBaseSink中入参是List<Put>，但是上面的实现只传一条数据，明显效率很低
            //用下面的方法批量Sink到HBase
            filterDs.countWindowAll(100)
                    .apply((AllWindowFunction<String, List<Put>, GlobalWindow>) (window, message, out) -> { 
                        List<Put> putList1 = new ArrayList<>();
                        for (String value : message) { 
                            String[] columns = value.split(",");
                            String rowKey = columns[0] + "_" + columns[1];
                            Put put = new Put(Bytes.toBytes(rowKey));
                            for (int i = 2; i < columns.length; i++) { 
                                put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("column" + i), Bytes.toBytes(columns[i]));
                            }
                            putList1.add(put);
                        }
                         out.collect(putList1);
                     })
                    .addSink(new HBaseSink());
        }
    }