Apache HBase 是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统，利用 HBase 技术可在廉价服务器上搭建起大规模结构化存储集群。HBase不同于一般的关系数据库，它是一个适合于非结构化数据存储的数据库，HBase 基于列的而不是基于行的模式。

Apache Flink 官方未提供 HBase DataStream 的连接器。Apache StreamPark 基于 HBase client 封装了 HBaseSource、HBaseSink，支持依据配置自动创建连接，简化开发。StreamPark 读取 HBase 在开启 chekpoint 情况下可以记录读取数据的最新状态，通过数据本身标识可以恢复 source 对应偏移量。实现 source 端至少一次语义。

HBaseSource 实现了 Flink 的 Async I/O 接口，可以提升流处理的吞吐量。Sink 端默认支持至少一次的处理语义。在开启 checkpoint 情况下支持精确一次语义。

StreamPark 读取 HBase 在开启 chekpoint 情况下可以记录读取数据的最新状态，作业恢复后从是否可以恢复之前状态完全取决于数据本身是否有偏移量的标识，需要在代码手动指定。在 HBaseSource 的 getDataStream 方法 func 参数指定恢复逻辑。

HBase写入依赖

HBase Maven 依赖:

  1. <dependency>
  2.   <groupId>org.apache.hbase</groupId>
  3.   <artifactId>hbase-client</artifactId>
  4.   <version>${hbase.version}</version>
  5. </dependency>
  6. <dependency>
  7.   <groupId>org.apache.hbase</groupId>
  8.   <artifactId>hbase-common</artifactId>
  9.   <version>${hbase.version}</version>
  10. </dependency>

常规方式写入读取HBase

1.创建库表

  1.  create 'Student', {NAME => 'Stulnfo', VERSIONS => 3}, {NAME =>'Grades', BLOCKCACHE => true}

2.写入读取demo

读取数据

  2. import org.apache.flink.configuration.Configuration;
  3. import org.apache.flink.streaming.api.datastream.DataStream;
  4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  5. import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
  6. import org.apache.hadoop.hbase.Cell;
  7. import org.apache.hadoop.hbase.HBaseConfiguration;
  8. import org.apache.hadoop.hbase.TableName;
  9. import org.apache.hadoop.hbase.client.*;
  10. import org.apache.hadoop.hbase.util.Bytes;
  12. import java.util.List;
  15. public class FlinkHBaseReader {
  16.     public static void main(String[] args) throws Exception {
  17.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  18.         // 添加数据源
  19.         DataStream<String> stream = env.addSource(new HBaseReader());
  20.         stream.print();
  21.         env.execute("FlinkHBaseDemo");
  22.     }
  23. }
  26. class HBaseReader extends RichSourceFunction<String> {
  27.     private Connection connection = null;
  28.     private ResultScanner rs = null;
  29.     private Table table = null;
  31.     @Override
  32.     public void open(Configuration parameters) throws Exception {
  33.         org.apache.hadoop.conf.Configuration hconf = HBaseConfiguration.create();
  34.         hconf.set("hbase.zookeeper.quorum", "localhost:2181");
  35.         hconf.set("zookeeper.property.clientPort", "/hbase");
  36.         connection = ConnectionFactory.createConnection(hconf);
  37.     }
  39.     @Override
  40.     public void run(SourceContext<String> sourceContext) throws Exception {
  41.         table = connection.getTable(TableName.valueOf("Student"));
  42.         Scan scan = new Scan();
  43.         scan.addFamily(Bytes.toBytes("Stulnfo"));
  44.         rs = table.getScanner(scan);
  45.         for (Result result : rs) {
  46.             StringBuilder sb = new StringBuilder();
  47.             List<Cell> cells = result.listCells();
  48.             for (Cell cell : cells) {
  49.                 String value = Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  50.                 sb.append(value).append("-");
  51.             }
  52.             String value = sb.replace(sb.length() - 1, sb.length(), "").toString();
  53.             sourceContext.collect(value);
  54.         }
  55.     }
  57.     @Override
  58.     public void cancel() {
  60.     }
  62.     @Override
  63.     public void close() throws Exception {
  64.         if (rs != null) {
  65.             rs.close();
  66.         }
  67.         if (table != null) {
  68.             table.close();
  69.         }
  70.         if (connection != null) {
  71.             connection.close();
  72.         }
  73.     }
  74. }

写入数据

  1. import com.zhisheng.common.utils.ExecutionEnvUtil;
  2. import lombok.extern.slf4j.Slf4j;
  3. import org.apache.flink.api.java.utils.ParameterTool;
  4. import org.apache.flink.configuration.Configuration;
  5. import org.apache.flink.streaming.api.datastream.DataStream;
  6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  7. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  8. import org.apache.flink.streaming.api.functions.source.SourceFunction;
  9. import org.apache.hadoop.hbase.HBaseConfiguration;
  10. import org.apache.hadoop.hbase.TableName;
  11. import org.apache.hadoop.hbase.client.*;
  12. import org.slf4j.Logger;
  13. import org.slf4j.LoggerFactory;
  15. import java.io.IOException;
  17. /**
  18.  * Desc: 读取流数据，然后写入到 HBase
  19.  */
  20. @Slf4j
  21. public class HBaseStreamWriteMain {
  23.     public static void main(String[] args) throws Exception {
  24.         final ParameterTool parameterTool = ExecutionEnvUtil.createParameterTool(args);
  25.         StreamExecutionEnvironment env = ExecutionEnvUtil.prepare(parameterTool);
  26.         DataStream<String> dataStream = env.addSource(new SourceFunction<String>() {
  27.             private static final long serialVersionUID = 1L;
  29.             private volatile boolean isRunning = true;
  31.             @Override
  32.             public void run(SourceContext<String> out) throws Exception {
  33.                 while (isRunning) {
  34.                     out.collect("name" + Math.floor(Math.random() * 100));
  35.                 }
  36.             }
  38.             @Override
  39.             public void cancel() {
  40.                 isRunning = false;
  41.             }
  42.         });
  43.         dataStream.addSink(new HBaseWriter());
  44.         env.execute("Flink HBase connector sink");
  45.     }
  48. }
  50. /**
  51.  * 写入HBase
  52.  * 继承RichSinkFunction重写父类方法
  53.  * <p>
  54.  * 写入hbase时500条flush一次, 批量插入, 使用的是writeBufferSize
  55.  */
  56. class HBaseWriter extends RichSinkFunction<String> {
  57.     private static final Logger logger = LoggerFactory.getLogger(HBaseWriter.class);
  59.     private static org.apache.hadoop.conf.Configuration configuration;
  60.     private static Connection connection = null;
  61.     private static BufferedMutator mutator;
  62.     private static int count = 0;
  64.     @Override
  65.     public void open(Configuration parameters) throws Exception {
  66.         configuration = HBaseConfiguration.create();
  67.         configuration.set("hbase.zookeeper.quorum", "localhost:21981");
  68.         configuration.set("zookeeper.property.clientPort", "/hbase");
  69.         try {
  70.             connection = ConnectionFactory.createConnection(configuration);
  71.         } catch (IOException e) {
  72.             e.printStackTrace();
  73.         }
  74.         BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf("Student"));
  75.         params.writeBufferSize(2 * 1024 * 1024);
  76.         mutator = connection.getBufferedMutator(params);
  77.     }
  79.     @Override
  80.     public void close() throws IOException {
  81.         if (mutator != null) {
  82.             mutator.close();
  83.         }
  84.         if (connection != null) {
  85.             connection.close();
  86.         }
  87.     }
  89.     @Override
  90.     public void invoke(String values, Context context) throws Exception {
  91.         //Date 1970-01-06 11:45:55  to 445555000
  92.         long unixTimestamp = 0;
  93.         String RowKey = String.valueOf(unixTimestamp);
  94.         Put put = new Put(RowKey.getBytes());
  95.         put.addColumn("Stulnfo".getBytes(), "Name".getBytes(), values.getBytes());
  96.         mutator.mutate(put);
  97.         //每满500条刷新一下数据
  98.         if (count >= 500) {
  99.             mutator.flush();
  100.             count = 0;
  101.         }
  102.         count = count + 1;
  103.     }
  104. }

以方式读写HBase较繁琐，非常的不灵敏。StreamPark使用约定大于配置、自动配置的方式只需要配置HBase连接参数、flink运行参数，StreamPark 会自动组装source和sink，极大的简化开发逻辑，提升开发效率和维护性。

Apache StreamPark™ 读写 HBase

1. 配置策略和连接信息

  1. # hbase
  2. hbase:
  3.   zookeeper.quorum: test1,test2,test6
  4.   zookeeper.property.clientPort: 2181
  5.   zookeeper.session.timeout: 1200000
  6.   rpc.timeout: 5000
  7.   client.pause: 20

2. 读写入HBase

用 StreamPark 写入HBase非常简单,代码如下:

读取HBase

  2. import org.apache.streampark.common.util.ConfigUtils
  3. import org.apache.streampark.flink.core.java.wrapper.HBaseQuery
  4. import org.apache.streampark.flink.core.scala.FlinkStreaming
  5. import org.apache.streampark.flink.core.scala.request.HBaseRequest
  6. import org.apache.streampark.flink.core.scala.source.HBaseSource
  7. import org.apache.flink.api.scala.createTypeInformation
  8. import org.apache.hadoop.hbase.CellUtil
  9. import org.apache.hadoop.hbase.client.{Get, Scan}
  10. import org.apache.hadoop.hbase.util.Bytes
  12. import java.util
  14. object HBaseSourceApp extends FlinkStreaming {
  16.   override def handle(): Unit = {
  18.     implicit val conf = ConfigUtils.getHBaseConfig(context.parameter.toMap)
  20.     val id = HBaseSource().getDataStream[String](query => {
  21.         new HBaseQuery("person", new Scan())
  22.     },
  23.       //以下方法决定从checkpoint恢复偏移量的逻辑
  24.       r => new String(r.getRow), null)
  25. //flink Async I/O
  26.     HBaseRequest(id).requestOrdered(x => {
  27.       new HBaseQuery("person", new Get(x.getBytes()))
  28.     }, (a, r) => {
  29.       val map = new util.HashMap[String, String]()
  30.       val cellScanner = r.cellScanner()
  31.       while (cellScanner.advance()) {
  32.         val cell = cellScanner.current()
  33.         val q = Bytes.toString(CellUtil.cloneQualifier(cell))
  34.         val (name, v) = q.split("_") match {
  35.           case Array(_type, name) =>
  36.             _type match {
  37.               case "i" => name -> Bytes.toInt(CellUtil.cloneValue(cell))
  38.               case "s" => name -> Bytes.toString(CellUtil.cloneValue(cell))
  39.               case "d" => name -> Bytes.toDouble(CellUtil.cloneValue(cell))
  40.               case "f" => name -> Bytes.toFloat(CellUtil.cloneValue(cell))
  41.             }
  42.           case _ =>
  43.         }
  44.         map.put(name.toString, v.toString)
  45.       }
  46.       map.toString
  47.     }).print("Async")
  48.   }
  50. }

写入HBase

  1. import org.apache.streampark.flink.core.scala.FlinkStreaming
  2. import org.apache.streampark.flink.core.scala.sink.{HBaseOutputFormat, HBaseSink}
  3. import org.apache.flink.api.scala._
  4. import org.apache.streampark.common.util.ConfigUtils
  5. import org.apache.hadoop.hbase.client.{Mutation, Put}
  6. import org.apache.hadoop.hbase.util.Bytes
  8. import java.util.{Collections, Random}
  10. object HBaseSinkApp extends FlinkStreaming {
  12.   override def handle(): Unit = {
  13.     val source = context.addSource(new TestSource)
  14.     val random = new Random()
  16.     //定义转换规则...
  17.     implicit def entry2Put(entity: TestEntity): java.lang.Iterable[Mutation] = {
  18.       val put = new Put(Bytes.toBytes(System.nanoTime() + random.nextInt(1000000)), entity.timestamp)
  19.       put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("cid"), Bytes.toBytes(entity.cityId))
  20.       put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("oid"), Bytes.toBytes(entity.orderId))
  21.       put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("os"), Bytes.toBytes(entity.orderStatus))
  22.       put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("oq"), Bytes.toBytes(entity.quantity))
  23.       put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("sid"), Bytes.toBytes(entity.siteId))
  24.       Collections.singleton(put)
  25.     }
  26.     //source ===> trans ===> sink
  28.     //1）插入方式1
  29.     HBaseSink().sink[TestEntity](source, "order")
  30.     //2) 插入方式2
  31.     //1.指定HBase 配置文件
  32.     implicit val prop = ConfigUtils.getHBaseConfig(context.parameter.toMap)
  33.     //2.插入...
  34.     source.writeUsingOutputFormat(new HBaseOutputFormat[TestEntity]("order", entry2Put))
  37.   }
  39. }

StreamPark 写入HBase 需要创建HBaseQuery的方法、指定将查询结果转化为需要对象的方法、标识是否在运行、传入运行参数。具体如下：

  1. /**
  2.  * @param ctx
  3.  * @param property
  4.  */
  5. class HBaseSource(@(transient@param) val ctx: StreamingContext, property: Properties = new Properties()) {
  7.   /**
  8.    * @param query   指定创建HBaseQuery的方法
  9.    * @param func    查询结果转化为期望对方方法
  10.    * @param running 运行标识
  11.    * @param prop    作业参数
  12.    * @tparam R  返回类型
  13.    * @return
  14.    */
  15.   def getDataStream[R: TypeInformation](query: R => HBaseQuery,
  16.                                         func: Result => R,
  17.                                         running: Unit => Boolean)(implicit prop: Properties = new Properties()) = {
  18.     Utils.copyProperties(property, prop)
  19.     val hBaseFunc = new HBaseSourceFunction[R](prop, query, func, running)
  20.     ctx.addSource(hBaseFunc)
  21.   }
  23. }

StreamPark HBaseSource 实现了flink Async I/O 用于提升Streaming的吞吐量，先创建 DataStream 然后创建 HBaseRequest 调用 requestOrdered（） 或者 requestUnordered（） 创建异步流，建如下代码：

  1. class HBaseRequest[T: TypeInformation](@(transient@param) private val stream: DataStream[T], property: Properties = new Properties()) {
  3.   /**
  4.    *
  5.    * @param queryFunc
  6.    * @param resultFunc
  7.    * @param timeout
  8.    * @param capacity
  9.    * @param prop
  10.    * @tparam R
  11.    * @return
  12.    */
  13.   def requestOrdered[R: TypeInformation](queryFunc: T => HBaseQuery, resultFunc: (T, Result) => R, timeout: Long = 1000, capacity: Int = 10)(implicit prop: Properties): DataStream[R] = {
  14.     Utils.copyProperties(property, prop)
  15.     val async = new HBaseAsyncFunction[T, R](prop, queryFunc, resultFunc, capacity)
  16.     AsyncDataStream.orderedWait(stream, async, timeout, TimeUnit.MILLISECONDS, capacity)
  17.   }
  19.   /**
  20.    *
  21.    * @param queryFunc
  22.    * @param resultFunc
  23.    * @param timeout
  24.    * @param capacity
  25.    * @param prop
  26.    * @tparam R
  27.    * @return
  28.    */
  29.   def requestUnordered[R: TypeInformation](queryFunc: T => HBaseQuery, resultFunc: (T, Result) => R, timeout: Long = 1000, capacity: Int = 10)(implicit prop: Properties): DataStream[R] = {
  30.     Utils.copyProperties(property, prop)
  31.     val async = new HBaseAsyncFunction[T, R](prop, queryFunc, resultFunc, capacity)
  32.     AsyncDataStream.unorderedWait(stream, async, timeout, TimeUnit.MILLISECONDS, capacity)
  33.   }
  35. }

StreamPark 支持两种方式写入数据：1.addSink() 2. writeUsingOutputFormat 样例如下：

  1.     //1）插入方式1
  2.     HBaseSink().sink[TestEntity](source, "order")
  3.     //2) 插入方式2
  4.     //1.指定HBase 配置文件
  5.     implicit val prop = ConfigUtils.getHBaseConfig(context.parameter.toMap)
  6.     //2.插入...
  7.     source.writeUsingOutputFormat(new HBaseOutputFormat[TestEntity]("order", entry2Put))

其他配置

其他的所有的配置都必须遵守 StreamPark 配置,具体可配置项和各个参数的作用请参考项目配置