Apache HBase is a highly reliable, high-performance, column-oriented, and scalable distributed storage system. Using HBase technology, large-scale structured storage clusters can be built on cheap PC Servers. Unlike general relational databases, Apache HBase is a database suitable for unstructured data storage because HBase storage is based on a column rather than a row-based schema.

Apache Flink does not officially provide a connector for HBase DataStream. Apache StreamPark encapsulates HBaseSource and HBaseSink based on HBase-client. It supports automatic connection creation based on configuration and simplifies development. StreamPark reading Apache HBase can record the latest status of the read data when the checkpoint is enabled, and the offset corresponding to the source can be restored through the data itself. Implement source-side AT_LEAST_ONCE.

HBaseSource implements Flink Async I/O to improve streaming throughput. The sink side supports AT_LEAST_ONCE by default. EXACTLY_ONCE is supported when checkpointing is enabled.

StreamPark reading Apache HBase can record the latest state of the read data when checkpoint is enabled. Whether the previous state can be restored after the job is resumed depends entirely on whether the data itself has an offset identifier, which needs to be manually specified in the code. The recovery logic needs to be specified in the func parameter of the getDataStream method of HBaseSource.

Dependency of Apache HBase writing

Apache HBase Maven Dependency:

  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>

Regular way to write and read Apache HBase

1.Create database and table

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

2.Write demo and Read demo

read data

  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. }

data input

  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: Read stream data, then write to Apache 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. Write to Apache HBase
  52.   Inherit RichSinkFunction to override the parent class method
  53.   <p>
  54.   When writing to Apache HBase, 500 items are flushed once, inserted in batches, using 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. }

Reading and writing Apache HBase in this way is cumbersome and inconvenient. StreamPark follows the concept of convention over configuration and automatic configuration. Users only need to configure Apache HBase connection parameters and Flink operating parameters. StreamPark will automatically assemble source and sink, which greatly simplifies development logic and improves development efficiency and maintainability.

write and read Apache HBase with Apache StreamPark™

1. Configure policies and connection information

  1. # apache 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. Read and write Apache HBase

Writing to Apache HBase with StreamPark is very simple, the code is as follows:

read Apache 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.       //The following methods determine the logic for restoring offsets from checkpoints
  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. }

write Apache 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）INSERT WAY 1
  29.     HBaseSink().sink[TestEntity](source, "order")
  30.     //2) 插入方式2
  31.     //1.Specify the HBase configuration file
  32.     implicit val prop = ConfigUtils.getHBaseConfig(context.parameter.toMap)
  33.     //2.break in...
  34.     source.writeUsingOutputFormat(new HBaseOutputFormat[TestEntity]("order", entry2Put))
  35.   }
  36. }

When StreamPark writes to Apache HBase, you need to create the method of HBaseQuery, specify the method to convert the query result into the required object, identify whether it is running, and pass in the running parameters. details as follows

  1. /**
  2.  * @param ctx
  3.  * @param property
  4.  */
  5. class HBaseSource(@(transient@param) val ctx: StreamingContext, property: Properties = new Properties()) {
  7.   /**
  8.    * @param query   Specify the method to create H Base Query
  9.    * @param func    The query results are converted into the expected counterparty method
  10.    * @param running runID
  11.    * @param prop    Job parameters
  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 implements flink Async I/O, which is used to improve the throughput of Streaming: first create a DataStream, then create an HBaseRequest and call requestOrdered() or requestUnordered() to create an asynchronous stream, as follows:

  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 supports two ways to write data: 1. addSink() 2. writeUsingOutputFormat Examples are as follows:

  1.     //1）Insert way 1
  2.     HBaseSink().sink[TestEntity](source, "order")
  3.     //2) insert way 2
  4.     //1.Specify the HBase configuration file
  5.     implicit val prop = ConfigUtils.getHBaseConfig(context.parameter.toMap)
  6.     //
  7.     source.writeUsingOutputFormat(new HBaseOutputFormat[TestEntity]("order", entry2Put))

Other configuration

All other configurations must comply with the StreamPark configuration. For specific configurable items and the role of each parameter, please refer to the project configuration