Flink操作Hbase

现在有这样一个场景，我们需要将hbase做成一个数据流，而不是数据集。根据Flink自带的Flink-Hbase只能帮我们做到数据集，所以这个时候选择了重写Hbase的数据源。

 package com.yjp.flink.demo11; import org.apache.flink.streaming.api.functions.source.SourceFunction; import org.apache.flink.table.shaded.org.joda.time.DateTime; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.KeyValue; import org.apache.hadoop.hbase.client.HTable; import org.apache.hadoop.hbase.client.Result; import org.apache.hadoop.hbase.client.ResultScanner; import org.apache.hadoop.hbase.client.Scan; import org.apache.hadoop.hbase.util.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.IOException; import java.util.ArrayList; import java.util.List; /** * 以Hbase为数据源 * 从Hbase中获取数据，然后以流的形式发射 * Date : 9:50 2018/3/12 */ public class HbaseSource implements SourceFunction<String> { private static Logger loggerFactory = LoggerFactory.getLogger(HbaseSource.class); private static final long serialVersionUID = 1; private volatile boolean isRunning = true; /** * 开始的时间戳 */ private long startTime; /** * 每次查询多长时间的数据 */ private long interval; /** * 需要查询的列名 */ private ArrayList<String> columns; /** * 需要查询的表名 */ private String tableName; public HbaseSource(long startTime, long interval, ArrayList<String> columns, String tableName) { this.startTime = startTime; this.interval = interval; this.columns = columns; this.tableName = tableName; } public HbaseSource() { } @Override public void run(SourceContext<String> out) { if (isRunning) { long endTime = DateTime.now().getMillis() - interval; ResultScanner rs = new HbaseSource().getHbaseData(tableName, startTime, endTime - startTime, columns); new HbaseSource().transmitData(rs, out); startTime = endTime; } while (isRunning) { ResultScanner rs = new HbaseSource().getHbaseData(tableName, startTime, interval, columns); new HbaseSource().transmitData(rs, out); startTime += interval; try { Thread.sleep(interval); } catch (InterruptedException e) { throw new RuntimeException("休眠异常", e); } } } @Override public void cancel() { } /** * 获取数据集 * * @param startTime 时间戳开始的时间 * @param interval 间隔时间 * @return 对应的结果集 */ private ResultScanner getHbaseData(String tableName, long startTime, long interval, List<String> columns) { Configuration conf = HBaseConfiguration.create(); HTable table; Scan scan; try { table = new HTable(conf, tableName); scan = new Scan(); scan.setTimeRange(startTime, startTime + interval); for (String column : columns) { String[] columnName = column.split(":"); scan.addColumn(Bytes.toBytes(columnName[0]), Bytes.toBytes(columnName[1])); } return table.getScanner(scan); } catch (IOException e) { throw new RuntimeException("读取数据异常", e); } } private void transmitData(ResultScanner rs, SourceContext<String> out) { Result result; try { while ((result = rs.next()) != null && isRunning) { KeyValue[] kvs = result.raw(); for (KeyValue kv : kvs) { String value = new String(kv.getValue()); out.collect(value); } } } catch (IOException e) { throw new RuntimeException("结果集遍历异常", e); } } } 然后将数据结果加工和处理存入Hbase中 package com.yjp.flink.hbase; import org.apache.flink.api.common.functions.FlatMapFunction; import org.apache.flink.api.java.tuple.Tuple2; import org.apache.flink.api.java.tuple.Tuple3; import org.apache.flink.streaming.api.datastream.DataStream; import org.apache.flink.streaming.api.datastream.DataStreamSource; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.Table; import org.apache.flink.table.api.java.StreamTableEnvironment; import org.apache.flink.util.Collector; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.ArrayList; import java.util.List; public class HbaseToHbase { public static Logger logger = LoggerFactory.getLogger(HbaseToHbase.class); public static void main(String[] args) throws Exception { StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv = StreamTableEnvironment.getTableEnvironment(sEnv); sEnv.getConfig().disableSysoutLogging(); List<String> getColumns = new ArrayList<String>(3); getColumns.add("cf1_name"); getColumns.add("cf2_amount"); getColumns.add("cf3_groupId"); List<String> columnFamily = new ArrayList<>(3); columnFamily.add("cf1"); columnFamily.add("cf2"); columnFamily.add("cf3"); List<String> setColumns = new ArrayList<>(3); setColumns.add("cf2:result"); DataStreamSource<Orders> orderDataStream = sEnv.addSource(new HbaseStreamDataSource("Orders", 0L, 2000L, getColumns, Orders.class)); DataStream<Tuple3<String, Double, Integer>> dataStream = orderDataStream.flatMap( new FlatMapFunction<Orders, Tuple3<String, Double, Integer>>() { @Override public void flatMap(Orders value, Collector<Tuple3<String, Double, Integer>> out) throws Exception { out.collect(new Tuple3<String, Double, Integer>(value.getCf1_name(), value.getCf2_amount(), value.getCf3_groupId())); } }); dataStream.keyBy(2).sum(1).addSink( new SinkHbase<Tuple3<String, Double, Integer>>( "OrderResult", columnFamily, setColumns, "result")); sEnv.execute("test Hbase"); } } package com.yjp.flink.hbase; import org.apache.flink.api.java.tuple.*; import org.apache.flink.configuration.Configuration; import org.apache.flink.streaming.api.functions.sink.RichSinkFunction; import org.apache.hadoop.hbase.HColumnDescriptor; import org.apache.hadoop.hbase.HTableDescriptor; import org.apache.hadoop.hbase.TableName; import org.apache.hadoop.hbase.client.Admin; import org.apache.hadoop.hbase.client.Table; import java.lang.reflect.Method; import java.util.*; /** * 自定义Sink * * Date : 17:23 2018/3/12 */ public class SinkHbase<T> extends RichSinkFunction<T> { private static final long serialVersionUID = 1L; /** * 表名 */ private String tableName; /** * 列族名 */ private List<String> columnFails; /** * 列名 以 family:column的形式传入 column与tuple中的值一一对应 */ private List<String> columns; /** * 行名 */ private String rowKey; /** * @param tableName 表名 * @param columnFamily 列族名 当表存在时不用输入 * @param columns 储存的列名 列族:列名 * @param rowKey 传入的行名 */ public SinkHbase(String tableName, List<String> columnFamily, List<String> columns, String rowKey) { this.tableName = tableName; this.columnFails = columnFamily; this.columns = columns; this.rowKey = rowKey; } /** * @param tableName 表名 * @param columns 储存的列名 列族:列名 * @param rowKey 传入的行名 */ public SinkHbase(String tableName, List<String> columns, String rowKey) { this.tableName = tableName; this.columns = columns; this.rowKey = rowKey; } public SinkHbase() { } /** * 初始化完成连接 当表不存在的时候 新建表和family列 * * @param parameters 调用父类的方法 * @throws Exception 创建连接失败 */ @Override public void open(Configuration parameters) throws Exception { super.open(parameters); Admin admin = FactoryConnect.getConnection().getAdmin(); final TableName tableName1 = TableName.valueOf(tableName); if (!admin.tableExists(tableName1)) { HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName1); for (String columnFamily : columnFails) { hTableDescriptor.addFamily(new HColumnDescriptor(columnFamily)); } admin.createTable(hTableDescriptor); } } /** * 执行方法 将数据存入hbase * * @param value 传入的结果 */ @Override public void invoke(T value, Context context) throws Exception { Map<Class, Method> map = new HashMap<>(25); new SinkHbase<T>().initMap(map); Table table = FactoryConnect.getConnection().getTable(TableName.valueOf(tableName)); Set<Class> keys = map.keySet(); for (Class key : keys) { if (value.getClass() == key) { map.get(key).invoke(new AssignmentTuple(), value, rowKey, columns, table); return; } } } private void initMap(Map<Class, Method> map) { try { map.put(Tuple1.class, AssignmentTuple.class.getMethod("setTuple1", Tuple1.class, String.class, ArrayList.class, Table.class)); map.put(Tuple2.class, AssignmentTuple.class.getMethod("setTuple2", Tuple2.class, String.class, ArrayList.class, Table.class)); map.put(Tuple3.class, AssignmentTuple.class.getMethod("setTuple3", Tuple3.class, String.class, ArrayList.class, Table.class)); map.put(Tuple4.class, AssignmentTuple.class.getMethod("setTuple4", Tuple4.class, String.class, ArrayList.class, Table.class)); map.put(Tuple5.class, AssignmentTuple.class.getMethod("setTuple5", Tuple5.class, String.class, ArrayList.class, Table.class)); map.put(Tuple6.class, AssignmentTuple.class.getMethod("setTuple6", Tuple6.class, String.class, ArrayList.class, Table.class)); map.put(Tuple7.class, AssignmentTuple.class.getMethod("setTuple7", Tuple7.class, String.class, ArrayList.class, Table.class)); map.put(Tuple8.class, AssignmentTuple.class.getMethod("setTuple8", Tuple8.class, String.class, ArrayList.class, Table.class)); map.put(Tuple9.class, AssignmentTuple.class.getMethod("setTuple9", Tuple9.class, String.class, ArrayList.class, Table.class)); map.put(Tuple10.class, AssignmentTuple.class.getMethod("setTuple10", Tuple10.class, String.class, ArrayList.class, Table.class)); map.put(Tuple11.class, AssignmentTuple.class.getMethod("setTuple11", Tuple11.class, String.class, ArrayList.class, Table.class)); map.put(Tuple12.class, AssignmentTuple.class.getMethod("setTuple12", Tuple12.class, String.class, ArrayList.class, Table.class)); map.put(Tuple13.class, AssignmentTuple.class.getMethod("setTuple13", Tuple13.class, String.class, ArrayList.class, Table.class)); map.put(Tuple14.class, AssignmentTuple.class.getMethod("setTuple14", Tuple14.class, String.class, ArrayList.class, Table.class)); map.put(Tuple15.class, AssignmentTuple.class.getMethod("setTuple15", Tuple15.class, String.class, ArrayList.class, Table.class)); map.put(Tuple16.class, AssignmentTuple.class.getMethod("setTuple16", Tuple16.class, String.class, ArrayList.class, Table.class)); map.put(Tuple17.class, AssignmentTuple.class.getMethod("setTuple17", Tuple17.class, String.class, ArrayList.class, Table.class)); map.put(Tuple18.class, AssignmentTuple.class.getMethod("setTuple18", Tuple18.class, String.class, ArrayList.class, Table.class)); map.put(Tuple19.class, AssignmentTuple.class.getMethod("setTuple19", Tuple19.class, String.class, ArrayList.class, Table.class)); map.put(Tuple20.class, AssignmentTuple.class.getMethod("setTuple20", Tuple20.class, String.class, ArrayList.class, Table.class)); map.put(Tuple21.class, AssignmentTuple.class.getMethod("setTuple21", Tuple21.class, String.class, ArrayList.class, Table.class)); map.put(Tuple22.class, AssignmentTuple.class.getMethod("setTuple22", Tuple22.class, String.class, ArrayList.class, Table.class)); map.put(Tuple23.class, AssignmentTuple.class.getMethod("setTuple23", Tuple23.class, String.class, ArrayList.class, Table.class)); map.put(Tuple24.class, AssignmentTuple.class.getMethod("setTuple24", Tuple24.class, String.class, ArrayList.class, Table.class)); map.put(Tuple25.class, AssignmentTuple.class.getMethod("setTuple25", Tuple25.class, String.class, ArrayList.class, Table.class)); } catch (NoSuchMethodException e) { throw new RuntimeException("反射失败", e); } } } package com.yjp.flink.hbase; import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.client.Connection; import org.apache.hadoop.hbase.client.ConnectionFactory; import java.io.IOException; import java.io.Serializable; /** * 单例模式 安全的拿到连接 * * Date : 16:45 2018/3/16 */ public class FactoryConnect implements Serializable { private static volatile Connection connection; private FactoryConnect() { } public static Connection getConnection() throws IOException { if (null == connection) { synchronized (FactoryConnect.class) { try { if (null == connection) { org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create(); connection = ConnectionFactory.createConnection(conf); } } catch (Exception e) { System.err.println("读取配置文件异常"); } } } return connection; } } package com.yjp.flink.hbase; import org.apache.flink.api.java.tuple.*; import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.Table; import org.apache.hadoop.hbase.util.Bytes; import java.io.IOException; import java.time.Instant; import java.util.ArrayList; import java.util.List; /** * 将tuple中的存放在Hbase中 * * Date : 16:49 2018/3/12 */ public class AssignmentTuple { /** * tuple 为1 * * @param tuple1 传入tuple的值 * @param rowKey 传入的rowkey的值 * @param columns 需要赋值的列 * @param table put的table对象 */ public void setTuple1(Tuple1<Object> tuple1, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple1, rowKey, columns, table); } public void setTuple2(Tuple2<Object, Object> tuple2, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple2, rowKey, columns, table); } public void setTuple3(Tuple3<Object, Object, Object> tuple3, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple3, rowKey, columns, table); } public void setTuple4(Tuple4<Object, Object, Object, Object> tuple4, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple4, rowKey, columns, table); } public void setTuple5(Tuple5<Object, Object, Object, Object, Object> tuple5, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple5, rowKey, columns, table); } public void setTuple6(Tuple6 tuple6, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple6, rowKey, columns, table); } public void setTuple7(Tuple7 tuple7, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple7, rowKey, columns, table); } public void setTuple8(Tuple8 tuple8, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple8, rowKey, columns, table); } public void setTuple9(Tuple9 tuple9, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple9, rowKey, columns, table); } public void setTuple10(Tuple10 tuple10, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple10, rowKey, columns, table); } public void setTuple11(Tuple11 tuple11, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple11, rowKey, columns, table); } public void setTuple12(Tuple12 tuple12, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple12, rowKey, columns, table); } public void setTuple13(Tuple13 tuple13, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple13, rowKey, columns, table); } public void setTuple14(Tuple14 tuple14, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple14, rowKey, columns, table); } public void setTuple15(Tuple15 tuple15, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple15, rowKey, columns, table); } public void setTuple16(Tuple16 tuple16, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple16, rowKey, columns, table); } public void setTuple17(Tuple17 tuple17, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple17, rowKey, columns, table); } public void setTuple18(Tuple18 tuple18, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple18, rowKey, columns, table); } public void setTuple19(Tuple19 tuple19, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple19, rowKey, columns, table); } public void setTuple20(Tuple20 tuple20, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple20, rowKey, columns, table); } public void setTuple21(Tuple21 tuple21, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple21, rowKey, columns, table); } public void setTuple22(Tuple22 tuple22, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple22, rowKey, columns, table); } public void setTuple23(Tuple23 tuple23, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple23, rowKey, columns, table); } public void setTuple24(Tuple24 tuple24, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple24, rowKey, columns, table); } public void setTuple25(Tuple25 tuple25, String rowKey, ArrayList<String> columns, Table table) { new AssignmentTuple().putData(tuple25, rowKey, columns, table); } /** * 将tuple中的数据一一对应的赋值给列 * * @param tuple tuple中的数据 * @param rowKey 设置的行值 * @param columns 对应的列名 * @param table 对应的table对象 */ public void putData(Tuple tuple, String rowKey, List<String> columns, Table table) { Put put = new Put(Bytes.toBytes(rowKey)); Long timeStamp = Instant.now().toEpochMilli(); for (int i = 0; i < columns.size(); i++) { String[] split = columns.get(i).split(":"); put.addColumn(Bytes.toBytes(split[0]), Bytes.toBytes(split[1]), timeStamp, Bytes.toBytes(tuple.getField(i).toString())); } try { table.put(put); } catch (IOException e) { throw new RuntimeException("存放失败", e); } } } 

为了做到一个通用的数据源和数据存储，于是采用了反射的方法。