SparkSQL on HBase - hbase.column.mapping에 사용가능한 타입

HBase 공식 문서의 SparkSQL 부분을 확인해보면 다음과 같이 Dataframe을 만드는 부분이 있다:

val sqlContext = new SQLContext(sc)

df = sqlContext.load("org.apache.hadoop.hbase.spark",
  Map("hbase.columns.mapping" ->
   "KEY_FIELD STRING :key, A_FIELD STRING c:a, B_FIELD STRING c:b",
   "hbase.table" -> "t1"))

df.registerTempTable("hbaseTmp")

*) 우선 최신버전에 비해 바뀐점은 SparkSQL의 SqlContext에서 load를 바로 사용하는것은 deprecated 됐다. 현재 버전 기준 올바른 사용법은 sqlContext.read.format("org.apache.hadoop.hbase.spark').options ( Map("hbase.columns.mapping" -> "[mapping rule]", "hbase.table" -> "table name")).load() 이다.

여기서 궁금한게 hbase.columns.mapping 에 관련된 부분이였다. 실제 HBase에서는 매우 큰 숫자값에 의해서 Long이나 Int의 byte를 그대로 저장하는 방식도 존재하기 때문이다.

hbase.column.mapping 의 경우 내부적으로는 SchemaQualifierDefinition 이라는 객체내부에 정의된 타입만 사용할 수 있도록 정해져 있으며 이는 hbase-spark project 코드에서 org/apache/hadoop/hbase/spark/DefaultSource.scala 파일에 정의돼 있다.

DefaultSource.scala의 529 line을 확인하면 다음과 같이 정의돼 있다:

/**

 * Construct to contains column data that spend SparkSQL and HBase

 *

 * @param columnName   SparkSQL column name

 * @param colType      SparkSQL column type

 * @param columnFamily HBase column family

 * @param qualifier    HBase qualifier name

 */

case class SchemaQualifierDefinition(columnName:String,

                          colType:String,

                          columnFamily:String,

                          qualifier:String) extends Serializable {

  val columnFamilyBytes = Bytes.toBytes(columnFamily)

  val qualifierBytes = Bytes.toBytes(qualifier)

  val columnSparkSqlType:DataType = if (colType.equals("BOOLEAN")) BooleanType

    else if (colType.equals("TINYINT")) IntegerType

    else if (colType.equals("INT")) IntegerType

    else if (colType.equals("BIGINT")) LongType

    else if (colType.equals("FLOAT")) FloatType

    else if (colType.equals("DOUBLE")) DoubleType

    else if (colType.equals("STRING")) StringType

    else if (colType.equals("TIMESTAMP")) TimestampType

    else if (colType.equals("DECIMAL")) StringType

    else throw new IllegalArgumentException("Unsupported column type :" + colType)

}

따라서 hbase.columns.mapping에서 사용가능한 타입은 BOOLEAN, TINYINT, INT, BIGINT, FLOAT, DOUBLE, STRING, TIMESTAMP, DECIMAL 이라는것을 확인할 수 있다.