서론

Spark SQL에서 외부 데이터 소스를 사용할 때, 특히 HBase와 같은 NoSQL 저장소를 연동하는 경우 스키마 정의와 데이터 매핑이 핵심 과제가 된다. 개발자는 Spark의 DataFrame API를 통해 HBase 테이블을 마치 관계형 테이블처럼 쿼리하고 싶어 하지만, 두 시스템 간의 데이터 모델 차이로 인해 예기치 않은 결과가 발생할 수 있다.

Spark SQL 커스텀 데이터 소스 기본 구조

Spark는 DataSourceV1 인터페이스를 기반으로 사용자 정의 데이터 소스를 지원한다. 주요 구성 요소는 다음과 같다:

• RelationProvider: 파라미터 기반으로 데이터 소스를 생성
• BaseRelation: 실제 데이터 관계를 표현
• TableScan: 전체 테이블 스캔을 수행하여 RDD[Row] 반환

필요 시 PrunedScan 또는 PrunedFilteredScan을 구현해 컬럼 추출 및 필터링 최적화도 가능하다.

HBase 연동 구현 예시

HBase 테이블을 Spark SQL에서 접근하기 위해 아래와 같은 커스텀 소스 클래스를 작성할 수 있다:

class DefaultSource extends RelationProvider {
  override def createRelation(
    sqlContext: SQLContext,
    parameters: Map[String, String]
  ): BaseRelation = {
    HBaseRelation(parameters)(sqlContext)
  }
}

데이터 매핑을 위한 핵심 클래스는 다음과 같이 정의된다:

case class HBaseRelation(
  hbaseProps: Map[String, String]
)(@transient val sqlContext: SQLContext
) extends BaseRelation with TableScan with Serializable {

  private val tableName = hbaseProps("hbase_table_name")
  private val sparkSchemaDef = hbaseProps("sparksql_table_schema")
  private val hbaseSchemaDef = hbaseProps("hbase_table_schema")

  // 스키마 파싱
  private val sparkFields = parseSparkSchema(sparkSchemaDef)
  private val hbaseFields = parseHBaseSchema(hbaseSchemaDef)

  // 순서 보장이 필요한 경우 LinkedHashMap 사용
  private val fieldMapping: java.util.LinkedHashMap[HBaseColumn, SparkField] =
    buildOrderedMapping(hbaseFields, sparkFields)

  override def schema: StructType = {
    StructType(fieldMapping.entrySet().asScala.map { entry =>
      val hbaseCol = entry.getKey
      val sparkField = entry.getValue
      StructField(sparkField.name, toDataType(sparkField.dataType), nullable = false)
    }.toArray)
  }

  override def buildScan(): RDD[Row] = {
    val conf = HBaseConfiguration.create()
    conf.set(TableInputFormat.INPUT_TABLE, tableName)

    val rdd = sqlContext.sparkContext.newAPIHadoopRDD(
      conf,
      classOf[TableInputFormat],
      classOf[ImmutableBytesWritable],
      classOf[Result]
    )

    rdd.map(_._2).map { result =>
      val values = fieldMapping.keySet().asScala.map { col =>
        col.resolveValue(result)
      }.toArray
      Row.fromSeq(values.toSeq)
    }
  }
}

문제 원인: 무순서 맵핑

초기 구현에서 컬럼 순서 불일치 문제가 발생한 근본 원인은 Scala의 기본 Map이 삽입 순서를 보장하지 않기 때문이다. 예를 들어 다음 코드는 의도한 순서와 다른 결과를 낼 수 있다:

val zipResult = Array(("a", 1), ("b", 2), ("c", 3))
val map = zipResult.toMap // 순서 보장 안 됨

해결책: 순서 보장형 맵 사용

JVM 생태계에서 삽입 순서를 유지하는 대표적인 자료구조는 LinkedHashMap이다. 이를 활용해 스키마 매핑 로직을 수정해야 한다:

def buildOrderedMapping(
  hbaseCols: Array[HBaseColumn],
  sparkFields: Array[SparkField]
): java.util.LinkedHashMap[HBaseColumn, SparkField] = {
  require(hbaseCols.length == sparkFields.length, "컬럼 수 불일치")

  val linkedMap = new java.util.LinkedHashMap[HBaseColumn, SparkField]()
  hbaseCols.zip(sparkFields).foreach { case (hbaseCol, sparkField) =>
    linkedMap.put(hbaseCol, sparkField)
  }
  linkedMap
}

스키마 정의 시 주의사항

다음과 같은 조건을 반드시 지켜야 올바른 매핑이 가능하다:

• Spark SQL 스키마와 HBase 스키마의 컬럼 순서가 동일해야 함
• 두 스키마의 컬럼 개수는 일치해야 함
• 타입 변환 규칙은 명확히 정의되어야 함 (예: String → StringType)

테스트 코드 예시

정상 동작을 검증하기 위한 테스트 코드:

val spark = SparkSession.builder()
  .appName("HBase Integration Test")
  .master("local[4]")
  .getOrCreate()

val df = spark.read.format("com.example.HBaseSource")
  .option("hbase_table_name", "order_info")
  .option("sparksql_table_schema",
    "(id String, create_time String, open_lng String, open_lat String)")
  .option("hbase_table_schema",
    "(MM:id, MM:create_time, MM:open_lng, MM:open_lat)")
  .load()

df.createOrReplaceTempView("orders")
spark.sql("SELECT id, create_time FROM orders LIMIT 5").show()

결론

Spark와 HBase 통합 시 스키마 매핑의 순서 문제는 subtle하지만 치명적인 버그를 유발할 수 있다. 기본 Map 구조 대신 LinkedHashMap과 같은 순서 보장 자료구조를 사용함으로써 이 문제를 확실하게 방지할 수 있다. 또한, 스키마 정의 시 양쪽의 컬럼 순서 일치를 강제하는 검증 로직을 추가하면 런타임 오류를 사전에 차단할 수 있다.