订单交易数据统计分析_自定义Spark SQL多数据源

Spark内置很多数据源，却没有HBase的数据源，需要调用rdd的api，如果能有下面这种方式就非常完美了。

xxxxxxxxxx
 frame.write.format("hbase")
      .mode(SaveMode.Append)
      .option(ZK_HOST_HBASE, "itbaizhan.cn")
      .option(ZK_PORT_HBASE, 2181)
      .option(HBASE_TABLE, "users")
      .option(HBASE_FAMILY, "detail")
      .option(HBASE_ROW_KEY, "id")
      .option(HBASE_SELECT_WHERE_CONDITIONS,"id[ge]50")
      .save()

参考spark sql内置的数据源和第三方提供的数据源(如Kudu、Elaticsearch)，自定义HBase数据源。

思路

• HBaseRelation（数据源）：负责实现读写HBase的逻辑，继承两个接口，TableScan(读)，InsertableRelation(写)。
• DefaultSource： 工厂类，创建HBaseRelation提供给框架调用，实现两个接口，RelationProvider(创建TableScan对象)，CreatableRelationProvider(创建InsertableRelation对象)。如果要用简称format("hbase")，还需要实现DataSourceRegister。

默认数据源DefaultSource

xxxxxxxxxx
package com.itbaizhan.hbase

import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.sql.sources.{BaseRelation, CreatableRelationProvider, DataSourceRegister, RelationProvider}
import org.apache.spark.sql.types.{StringType, StructField, StructType}
/**
 * 默认数据源提供Relation对象，分别为加载数据和保存提供Relation对象
 */
class DefaultSource extends RelationProvider with CreatableRelationProvider with DataSourceRegister{

  val SPERATOR: String = ","
  val HBASE_TABLE_SELECT_FIELDS: String = "selectFields"

  // 使用简称
  override def shortName(): String = {
    "hbase"
  }

  /**
   * 返回BaseRelation实例对象，提供加载数据功能
   * @param sqlContext SQLContext实例对象
   * @param parameters 参数信息
   * @return
   */
  override def createRelation(sqlContext: SQLContext,
                               parameters: Map[String, String]
                             ): BaseRelation = {
    // 1. 定义Schema信息
    val schema: StructType = StructType(
      parameters(HBASE_TABLE_SELECT_FIELDS)
        .split(SPERATOR)
        .map{
          field => StructField(field, StringType, nullable = true)
        })

    // 2. 创建HBaseRelation对象
    val relation = new HBaseRelation(sqlContext, parameters, schema)

    // 3. 返回对象
    relation
  }

  /*** 返回BaseRelation实例对象，提供保存数据功能
   * @param sqlContext SQLContext实例对象
   * @param mode 保存模式
   * @param parameters 参数
   * @param data 数据集
   * @return
   */
  override def createRelation(
                               sqlContext: SQLContext,
                               mode: SaveMode,
                               parameters: Map[String, String],
                               data: DataFrame
                             ): BaseRelation = {
    // 1. 创建HBaseRelation对象
    val relation = new HBaseRelation(sqlContext, parameters, data.schema)
    // 2.保存数据
    relation.insert(data,true)
    // 3.返回
    relation
  }
}