Thrift API ..

Apache HBas..

HBase中文参考指南 3.0

HBase 与 Spark

贡献者：TsingJyujing

Apache Spark 是一个分布式的、用于在内存中处理数据的软件框架，在许多场景中用于代替 MapReduce。

Spark 本身已经超出了本文档的范围，请参考 Spark 的项目及子项目的网站来获取更多信息。本文档将会集中在 4 个主要的 HBase 和 Spark 交互的要点上，这四点分别是：

基础 Spark

这可以在 Spark DAG 中的任意一点使用 HBase Connection。

Spark Streaming

这可以在 Spark Streaming 应用中的任意一点使用 HBase Connection。

Spark 批量加载

这可以允许在批量插入 HBase 的时候直接写 HBase 的 HFiles。

SparkSQL/DataFrames

这将提供为 HBase 中定义的表提供写 SparkSQL 的能力。

下面的部分将会用几个例子来说明上面几点交互。

104. 基础 Spark

这一部分将会在最底层和最简单的等级讨论 HBase 与 Spark 的整合。其他交互的要点都是基于这些操作构建的，我们会在这里完整描述。

一切 HBase 和 Spark 整合的基础都是 HBaseContext，HBaseContext 接受 HBase 配置并且会将其推送到 Spark 执行器（executor）中。这允许我们在每个 Spark 执行器（executor）中有一个静态的 HBase 连接。

作为参考，Spark 执行器（executor）既可以和 Region Server 在同一个节点，也可以在不同的节点，他们不存在共存的依赖关系。

可以认为每个 Spark 执行器（executor）都是一个多线程的客户端程序，这允许运行在不同的执行器上的 Spark 任务访问共享的连接对象。

例 31. HBaseContext 使用例程

这个例子展现了如何使用 Scala 语言在 RDD 的foreachPartition方法中使用 HBaseContext。

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

...

val hbaseContext = new HBaseContext(sc, config)

rdd.hbaseForeachPartition(hbaseContext, (it, conn) => {
 val bufferedMutator = conn.getBufferedMutator(TableName.valueOf("t1"))
 it.foreach((putRecord) => {
. val put = new Put(putRecord._1)
. putRecord._2.foreach((putValue) => put.addColumn(putValue._1, putValue._2, putValue._3))
. bufferedMutator.mutate(put)
 })
 bufferedMutator.flush()
 bufferedMutator.close()
})

这里是使用 Java 编写的同样的例子。

JavaSparkContext jsc = new JavaSparkContext(sparkConf);

try {
  List<byte[]> list = new ArrayList<>();
  list.add(Bytes.toBytes("1"));
  ...
  list.add(Bytes.toBytes("5"));

  JavaRDD<byte[]> rdd = jsc.parallelize(list);
  Configuration conf = HBaseConfiguration.create();

  JavaHBaseContext hbaseContext = new JavaHBaseContext(jsc, conf);

  hbaseContext.foreachPartition(rdd,
      new VoidFunction<Tuple2<Iterator<byte[]>, Connection>>() {
   public void call(Tuple2<Iterator<byte[]>, Connection> t)
        throws Exception {
    Table table = t._2().getTable(TableName.valueOf(tableName));
    BufferedMutator mutator = t._2().getBufferedMutator(TableName.valueOf(tableName));
    while (t._1().hasNext()) {
      byte[] b = t._1().next();
      Result r = table.get(new Get(b));
      if (r.getExists()) {
       mutator.mutate(new Put(b));
      }
    }

    mutator.flush();
    mutator.close();
    table.close();
   }
  });
} finally {
  jsc.stop();
}

所有的函数式都同时在 Spark 和 HBase 中，并且都支持用 Scala 或者 Java 开发。除了 SparkSQL 以外，所有 Spark 支持的语言在这里也都支持。目前在余下的文档中，我们将会重点关注 Scala 的例程。

上面的例程阐释了如何在 foreachPartition 操作中使用连接。除此之外，许多 Spark 的基础函数都是支持的：

bulkPut

并行的写入大量数据到 HBase

bulkDelete

并行的删除 HBase 中大量数据

bulkGet

并行的从 HBase 中获取大量的数据，并且创建一个新的 RDD

mapPartition

在 Spark 的 Map 函数中使用连接对象，并且允许使用完整的 HBase 访问

hBaseRDD

简单的创建一个用于分布式扫描数据的 RDD

想要参看所有机能的例程，参见 HBase-Spark 模块。

105. Spark Streaming

Spark Streaming 是一个基于 Spark 构建的微批流处理框架。 HBase 和 Spark Streaming 的良好配合使得 HBase 可以提供一下益处：

可以动态的获取参考或者描述性数据
基于 Spark Streaming 提供的恰好一次处理，可以存储计数或者聚合结果

HBase-Spark 模块整合的和 Spark Streaming 的相关的点与 Spark 整合的点非常相似，以下的指令可以在 Spark Streaming DStream 中立刻使用：

bulkPut

并行的写入大量数据到 HBase

bulkDelete

并行的删除 HBase 中大量数据

bulkGet

并行的从 HBase 中获取大量的数据，并且创建一个新的 RDD

mapPartition

在 Spark 的 Map 函数中使用连接对象，并且允许使用完整的 HBase 访问

hBaseRDD

简单的创建一个用于分布式扫描数据的 RDD。

例 32. bulkPut在 DStreams 中使用的例程

以下是 bulkPut 在 DStreams 中的使用例程，感觉上与 RDD 批量插入非常接近。

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

val hbaseContext = new HBaseContext(sc, config)
val ssc = new StreamingContext(sc, Milliseconds(200))

val rdd1 = ...
val rdd2 = ...

val queue = mutable.Queue[RDD[(Array[Byte], Array[(Array[Byte],
    Array[Byte], Array[Byte])])]]()

queue += rdd1
queue += rdd2

val dStream = ssc.queueStream(queue)

dStream.hbaseBulkPut(
  hbaseContext,
  TableName.valueOf(tableName),
  (putRecord) => {
   val put = new Put(putRecord._1)
   putRecord._2.foreach((putValue) => put.addColumn(putValue._1, putValue._2, putValue._3))
   put
  })

这里到hbaseBulkPut函数有三个输入，hbaseContext 携带了配置广播信息，来帮助我们连接到执行器中的 HBase Connections。表名用于指明我们要往哪个表放数据。一个函数将 DStream 中的记录转换为 HBase Put 对象。

106. 批量加载

使用 Spark 加载大量的数据到 HBase 有两个选项。基本的大量数据加载功能适用于你的行有数百万列数据，以及在 Spark 批量加载之前的 Map 操作列没有合并和分组的情况。

Spark 中还有一个轻量批量加载选项，这个第二选项设计给每一行少于一万的情况。第二个选项的优势在于更高的吞吐量，以及 Spark 的 shuffle 操作中更轻的负载。

两种实现都或多或少的类似 MapReduce 批量加载过程，因为分区器基于 Region 划分对行键进行分区。并且行键被顺序的发送到 Reducer 所以 HFile 可以在 reduce 阶段被直接写出。

依照 Spark 的术语来说，批量加载将会基于repartitionAndSortWithinPartitions实现，并且之后是 Spark 的foreachPartition。

让我们首先看一下使用批量加载功能的例子

例 33. 批量加载例程

下面的例子展现了 Spark 中的批量加载。

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

val hbaseContext = new HBaseContext(sc, config)

val stagingFolder = ...
val rdd = sc.parallelize(Array(
      (Bytes.toBytes("1"),
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
      (Bytes.toBytes("3"),
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...

rdd.hbaseBulkLoad(TableName.valueOf(tableName),
  t => {
   val rowKey = t._1
   val family:Array[Byte] = t._2(0)._1
   val qualifier = t._2(0)._2
   val value = t._2(0)._3

   val keyFamilyQualifier= new KeyFamilyQualifier(rowKey, family, qualifier)

   Seq((keyFamilyQualifier, value)).iterator
  },
  stagingFolder.getPath)

val load = new LoadIncrementalHFiles(config)
load.doBulkLoad(new Path(stagingFolder.getPath),
  conn.getAdmin, table, conn.getRegionLocator(TableName.valueOf(tableName)))

hbaseBulkLoad 函数需要三个必要参数：

我们需要从之加载数据的表名
一个函数用于将 RDD 中的某个记录转化为一个元组形式的键值对。其中键值是一个 KeyFamilyQualifer 对象，值是 cell value。 KeyFamilyQualifer 将会保存行键，列族和列标识位。针对行键的随机操作会根据这三个值来排序。
写出 HFile 的临时路径

接下来的 Spark 批量加载指令，使用 HBase 的 LoadIncrementalHFiles 对象来加载 HBase 中新创建的 HFiles。

使用 Spark 批量加载的附加参数

你可以在 hbaseBulkLoad 中用附加参数设置以下属性：

HFile 的最大文件大小
从压缩中排除 HFile 的标志
列族设置，包含 compression（压缩）, bloomType（布隆（过滤器）类型）, blockSize（块大小）, and dataBlockEncoding（数据块编码）

例 34. 使用附加参数

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

val hbaseContext = new HBaseContext(sc, config)

val stagingFolder = ...
val rdd = sc.parallelize(Array(
      (Bytes.toBytes("1"),
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
      (Bytes.toBytes("3"),
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...

val familyHBaseWriterOptions = new java.util.HashMap[Array[Byte], FamilyHFileWriteOptions]
val f1Options = new FamilyHFileWriteOptions("GZ", "ROW", 128, "PREFIX")

familyHBaseWriterOptions.put(Bytes.toBytes("columnFamily1"), f1Options)

rdd.hbaseBulkLoad(TableName.valueOf(tableName),
  t => {
   val rowKey = t._1
   val family:Array[Byte] = t._2(0)._1
   val qualifier = t._2(0)._2
   val value = t._2(0)._3

   val keyFamilyQualifier= new KeyFamilyQualifier(rowKey, family, qualifier)

   Seq((keyFamilyQualifier, value)).iterator
  },
  stagingFolder.getPath,
  familyHBaseWriterOptions,
  compactionExclude = false,
  HConstants.DEFAULT_MAX_FILE_SIZE)

val load = new LoadIncrementalHFiles(config)
load.doBulkLoad(new Path(stagingFolder.getPath),
  conn.getAdmin, table, conn.getRegionLocator(TableName.valueOf(tableName)))

现在让我们来看一下如何调用轻量化对象批量加载的实现：

例 35. 使用轻量批量加载

val sc = new SparkContext("local", "test")
val config = new HBaseConfiguration()

val hbaseContext = new HBaseContext(sc, config)

val stagingFolder = ...
val rdd = sc.parallelize(Array(
      ("1",
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("a"), Bytes.toBytes("foo1"))),
      ("3",
        (Bytes.toBytes(columnFamily1), Bytes.toBytes("b"), Bytes.toBytes("foo2.b"))), ...

rdd.hbaseBulkLoadThinRows(hbaseContext,
      TableName.valueOf(tableName),
      t => {
        val rowKey = t._1

        val familyQualifiersValues = new FamiliesQualifiersValues
        t._2.foreach(f => {
          val family:Array[Byte] = f._1
          val qualifier = f._2
          val value:Array[Byte] = f._3

          familyQualifiersValues +=(family, qualifier, value)
        })
        (new ByteArrayWrapper(Bytes.toBytes(rowKey)), familyQualifiersValues)
      },
      stagingFolder.getPath,
      new java.util.HashMap[Array[Byte], FamilyHFileWriteOptions],
      compactionExclude = false,
      20)

val load = new LoadIncrementalHFiles(config)
load.doBulkLoad(new Path(stagingFolder.getPath),
  conn.getAdmin, table, conn.getRegionLocator(TableName.valueOf(tableName)))

注意在使用轻量行批量加载的时候函数返回的元组中：第一个元素是行键，第二个元素是 FamiliesQualifiersValues，这个对象中含有这一行里所有的数值，并且包含了所有的列族。

107. SparkSQL/DataFrames

（HBase-Spark 中的）HBase-Spark 连接器提供了： DataSource API (SPARK-3247) 在 Spark 1.2.0 的时候被引入，连接了简单的 HBase 的键值存储与复杂的关系型 SQL 查询，并且使得用户可以使用 Spark 在 HBase 上施展复杂的数据分析工作。 HBase Dataframe 是 Spark Dataframe 的一个标准，并且它允许和其他任何数据源——例如 Hive, Orc, Parquet, JSON 之类。 HBase-Spark Connector 使用的关键技术例如分区修剪，列修剪，推断后置以及数据本地化。

为了使用 HBase-Spark connector，用户需要定义 HBase 到 Spark 表中的映射目录。准备数据并且填充 HBase 的表，然后将其加载到 HBase DataFrame 中去。在此之后，用户可以使用 SQL 查询语句整合查询与数据获取。接下来的例子说明了最基本的过程

107.1. 定义目录

def catalog = s"""{
       |"table":{"namespace":"default", "name":"table1"},
       |"rowkey":"key",
       |"columns":{
         |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
         |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
         |"col2":{"cf":"cf2", "col":"col2", "type":"double"},
         |"col3":{"cf":"cf3", "col":"col3", "type":"float"},
         |"col4":{"cf":"cf4", "col":"col4", "type":"int"},
         |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
         |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
         |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
         |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
       |}
     |}""".stripMargin

目录定义了从 HBase 到 Spark 表的一个映射。这个目录中有两个关键部分。第一个是行键的定义，另一个是将 Spark 表中的列映射到 HBase 的列族和列标识位。上面的 schema 定义了一个 HBase 中的表，名为 Table1，行键作为键与一些列(col1 - col8)。注意行键也需要被定义为一个列（col0），该列具有特定的列族（rowkey）。

107.2. 保存 DataFrame

case class HBaseRecord(
   col0: String,
   col1: Boolean,
   col2: Double,
   col3: Float,
   col4: Int,
   col5: Long,
   col6: Short,
   col7: String,
   col8: Byte)

object HBaseRecord
{
   def apply(i: Int, t: String): HBaseRecord = {
      val s = s"""row${"%03d".format(i)}"""
      HBaseRecord(s,
      i % 2 == 0,
      i.toDouble,
      i.toFloat,
      i,
      i.toLong,
      i.toShort,
      s"String$i: $t",
      i.toByte)
  }
}

val data = (0 to 255).map { i =>  HBaseRecord(i, "extra")}

sc.parallelize(data).toDF.write.options(
 Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "5"))
 .format("org.apache.hadoop.hbase.spark ")
 .save()

用户准备的数据（data）是一个本地的 Scala 集合，含有 256 个 HBaseRecord 对象。 sc.parallelize(data) 函数分发了从 RDD 中来的数据。toDF返回了一个 DataFrame。 write 函数返回了一个 DataFrameWriter 来将 DataFrame 中的数据到外部存储（例如这里是 HBase）。 save 函数将会创建一个具有 5 个 Region 的 HBase 表来在内部保存 DataFrame。

107.3. 加载 DataFrame

def withCatalog(cat: String): DataFrame = {
  sqlContext
  .read
  .options(Map(HBaseTableCatalog.tableCatalog->cat))
  .format("org.apache.hadoop.hbase.spark")
  .load()
}
val df = withCatalog(catalog)

在 withCatalog 函数中，sqlContext 是一个 SQLContext 的变量，是一个用于与 Spark 中结构化（行与列）的数据一起工作的一个入口点。 read 返回一个 DataFrameReader，他可以用于从 DataFrame 中读取数据。option函数为输出到 DataFrameReader 的底层的数据源增加了输入选项。以及，format函数表示了 DataFrameReader 的输入数据源的格式。 load() 函数将其加载为一个 DataFrame，数据帧 df将由withCatalog函数返回，用于访问 HBase 表，例如 4.4 与 4.5.

107.4. Language Integrated Query

val s = df.filter(($"col0" <= "row050" && $"col0" > "row040") ||
  $"col0" === "row005" ||
  $"col0" <= "row005")
  .select("col0", "col1", "col4")
s.show

DataFrame 可以做很多操作，例如 join, sort, select, filter, orderBy 等等等等。df.filter 通过指定的 SQL 表达式提供过滤器，select选择一系列的列：col0, col1 和 col4。

107.5. SQL 查询

df.registerTempTable("table1")
sqlContext.sql("select count(col1) from table1").show

registerTempTable 注册了一个名为 df 的 DataFrame 作为临时表，表名为table1，临时表的生命周期和 SQLContext 有关，用于创建df。 sqlContext.sql函数允许用户执行 SQL 查询。

107.6. Others

例 36. 查询不同的时间戳

在 HBaseSparkConf 中，可以设置 4 个和时间戳有关的参数，它们分别表示为 TIMESTAMP, MIN_TIMESTAMP, MAX_TIMESTAMP 和 MAX_VERSIONS。用户可以使用不同的时间戳或者利用 MIN_TIMESTAMP 和 MAX_TIMESTAMP 查询时间范围内的记录。同时，下面的例子使用了具体的数值取代了 tsSpecified 和 oldMs。

下例展示了如何使用不同的时间戳加载 df DataFrame。tsSpecified 由用户定义，HBaseTableCatalog 定义了 HBase 和 Relation 关系的 schema。writeCatalog 定义了 schema 映射的目录。

val df = sqlContext.read
      .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.TIMESTAMP -> tsSpecified.toString))
      .format("org.apache.hadoop.hbase.spark")
      .load()

下例展示了如何使用不同的时间范围加载 df DataFrame。oldMs 由用户定义。

val df = sqlContext.read
      .options(Map(HBaseTableCatalog.tableCatalog -> writeCatalog, HBaseSparkConf.MIN_TIMESTAMP -> "0",
        HBaseSparkConf.MAX_TIMESTAMP -> oldMs.toString))
      .format("org.apache.hadoop.hbase.spark")
      .load()

在加载 DataFrame 之后，用户就可以查询数据。

df.registerTempTable("table")
sqlContext.sql("select count(col1) from table").show

例 37. 原生 Avro 支持

Example 37. Native Avro support

HBase-Spark Connector 支持不同类型的数据格式例如 Avro, Jason 等等。下面的用例展示了 Spark 如何支持 Avro。用户可以将 Avro 记录直接持久化进 HBase。在内部，Avro schema 自动的转换为原生的 Spark Catalyst 数据类型。注意，HBase 表中无论是键或者值的部分都可以在 Avro 格式定义。

def catalog = s"""{
                     |"table":{"namespace":"default", "name":"Avrotable"},
                      |"rowkey":"key",
                      |"columns":{
                      |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
                      |"col1":{"cf":"cf1", "col":"col1", "type":"binary"}
                      |}
                      |}""".stripMargin

catalog是一个 HBase 表的 schema，命名为 Avrotable。行键作为键，并且有一个列 col1。行键也被定义为详细的一列（col0），并且指定列族（rowkey）。

准备数据：

 object AvroHBaseRecord {
   val schemaString =
     s"""{"namespace": "example.avro",
         |   "type": "record",      "name": "User",
         |    "fields": [
         |        {"name": "name", "type": "string"},
         |        {"name": "favorite_number",  "type": ["int", "null"]},
         |        {"name": "favorite_color", "type": ["string", "null"]},
         |        {"name": "favorite_array", "type": {"type": "array", "items": "string"}},
         |        {"name": "favorite_map", "type": {"type": "map", "values": "int"}}
         |      ]    }""".stripMargin

   val avroSchema: Schema = {
     val p = new Schema.Parser
     p.parse(schemaString)
   }

   def apply(i: Int): AvroHBaseRecord = {
     val user = new GenericData.Record(avroSchema);
     user.put("name", s"name${"%03d".format(i)}")
     user.put("favorite_number", i)
     user.put("favorite_color", s"color${"%03d".format(i)}")
     val favoriteArray = new GenericData.Array[String](2, avroSchema.getField("favorite_array").schema())
     favoriteArray.add(s"number${i}")
     favoriteArray.add(s"number${i+1}")
     user.put("favorite_array", favoriteArray)
     import collection.JavaConverters._
     val favoriteMap = Map[String, Int](("key1" -> i), ("key2" -> (i+1))).asJava
     user.put("favorite_map", favoriteMap)
     val avroByte = AvroSedes.serialize(user, avroSchema)
     AvroHBaseRecord(s"name${"%03d".format(i)}", avroByte)
   }
 }

 val data = (0 to 255).map { i =>
    AvroHBaseRecord(i)
 }

首先定义 schemaString，然后它被解析来获取avroSchema，avroSchema是用来生成 AvroHBaseRecord的。data 由用户准备，是一个有 256 个AvroHBaseRecord对象的原生 Scala 集合。

保存 DataFrame:

 sc.parallelize(data).toDF.write.options(
     Map(HBaseTableCatalog.tableCatalog -> catalog, HBaseTableCatalog.newTable -> "5"))
     .format("org.apache.spark.sql.execution.datasources.hbase")
     .save()

对于由 schema catalog提供的已有的数据帧，上述语句将会创建一个具有 5 个分区的 HBase 表，并且将数据存进去。

加载 DataFrame

def avroCatalog = s"""{
            |"table":{"namespace":"default", "name":"avrotable"},
            |"rowkey":"key",
            |"columns":{
              |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
              |"col1":{"cf":"cf1", "col":"col1", "avro":"avroSchema"}
            |}
          |}""".stripMargin

 def withCatalog(cat: String): DataFrame = {
     sqlContext
         .read
         .options(Map("avroSchema" -> AvroHBaseRecord.schemaString, HBaseTableCatalog.tableCatalog -> avroCatalog))
         .format("org.apache.spark.sql.execution.datasources.hbase")
         .load()
 }
 val df = withCatalog(catalog)

在 withCatalog 函数中，read 会返回一个可以将数据读取成 DataFrame 格式的 DataFrameReader。 option 函数追加输入选项来指定 DataFrameReader 使用的底层数据源。这里有两个选项，一个是设置avroSchema为AvroHBaseRecord.schemaString，另一个是设置HBaseTableCatalog.tableCatalog 为 avroCatalog。load() 函数加载所有的数据为 DataFrame。数据帧 df 由withCatalog 函数返回，可用于访问 HBase 表中的数据。

SQL 查询

 df.registerTempTable("avrotable")
 val c = sqlContext.sql("select count(1) from avrotable").

在加载 df DataFrame 之后，用户可以查询数据。registerTempTable 将 df DataFrame 注册为一个临时表，表名为 avrotable。 sqlContext.sql函数允许用户执行 SQL 查询。

HBase中文参考指南 3.0

HBase 与 Spark

104. 基础 Spark

105. Spark Streaming

106. 批量加载

107. SparkSQL/DataFrames

107.1. 定义目录

107.2. 保存 DataFrame

107.3. 加载 DataFrame

107.4. Language Integrated Query

107.5. SQL 查询

107.6. Others

全栈开发指南2021

程序员的自我修养

简明Excel VBA

Git 菜单-高质量的 Git 中文教程

Git的奇技淫巧

Office 365 开发入门指南