Spark SQL로, Hive parquet 접근 시 필드명이 parquet schema와 다르게 잡힐 때..

문제

며칠 전 spark SQL로 조회하던 parquet 파일 베이스의 hive external table의 필드 schema의 대소문자가 꼬여 문제가 생겼던 적이 있다. 어떤 field는 대소문자를 구분했고, 어떤 필드는 소문자로만 보였다. case-sensitive와 case-insesitive가 섞인 mixed-case 의 형태를 보인 것이다.

spark.sql('FROM tbl').printSchema()

root
 |-- isValid: string (nullable = true) # 대소문자를 구분한다
 |-- requestid: string (nullable = true) # parquet schema상으로는 reqeustId이지만, 소문자로만 조회가 된다.
 ..

개발 및 테스트 과정 중 hive table을 삭제하고 다시 만드는 과정 중에 뭔가 꼬인 것 같았다. 아직 서비스가 오픈하지 않았기 때문에 데이터와 설정을 모두 삭제하고 다시 hive tbl생성 및 적재를 통해 해결했지만, 본질적으로 어떻게 동작하는지 관련된 설정을 알아보기로 했다.

Spark SQL은 어떻게 대소문자를 구분해 Hive 쿼리할 수 있을까?

기본적으로 hive는 table, field명은 소문자만 허용한다. parquet은 field명의 대소문자를 구분한다.
그런데 external table로 parquet파일을 지정하고, Spark SQL에서 쿼리를 날리면, Hive 쿼리 임에도 parquet의 schema대로 대소문자를 구분한 쿼리를 날릴 수 있도록 지원한다.
이것이 가능한 이유는 Spark SQL이 필드명을 추론하기 때문이다. 이와 관련된 설정은 아래 2개이다.

spark.sql.hive.convertMetastoreParquet=true (default)
spark.sql.hive.caseSensitiveInferenceMode=INFER_AND_SAVE (default)

아래와 같이 동작한다는 의미이다.

• spark sql 이 hive 스키마와 parquet스키마를 이용해 조정된 스키마(reconciliation)를 만들고, 이를 통해서 대소문자를 구분한 schema를 사용할 수 있게 해 줌
• 이렇게 추론된 스키마를 hive table properties 에 다시 적는 방식(INFER_AND_SAVE - 추론과 저장)으로 동작

한번 추론 된 스키마가 hive 테이블 속성에 저장이 되고 나면, 아래와 같이 추론하지 않도록 설정을 주어도 대소문자를 구분하는 형태로만 조회된다. 아래와 같이 추론하지 않도록 설정을 해도 테이블 속성에 추론된 방식을 사용하도록 되어있다. (spark 2.3.2 기준)

spark.sql.hive.convertMetastoreParquet=false
spark.sql.hive.caseSensitiveInferenceMode=NEVER_INFER

그렇다면 hive 테이블 속성에는 어떻게 저장되고 있는지는 아래 명령어로 확인해 볼 수 있다. spark.sql.sources.schema.part.0 에 아래와 같은 값이 escaped 처리된 string으로 들어있다.

describe formatted `tablename`;

...
| spark.sql.sources.schema.numParts                  | 1                                                  
| spark.sql.sources.schema.part.0                    | \n{\"type\":\"struct\",\"fields\":[{\"name\":\"isValid\",\"type\":\"string\",\"nullable\":true,\"metadata\":{\"com ...

escaped된 json을 표현해보면 아래와 같다. fields[].name 에 있는 값을 spark SQL 상에서 field로 조회할 수 있는 이름이다.

{
  "type": "struct",
  "fields": [
    {
      "name": "isValid", "type": "string""nullable": true", metadata": {
        "comment": "유효 여부"
      }
    },
    {
      "name": "requestId", "type": "string", "nullable": true", metadata": {
        "comment": "요청Id"
      }
    }
  ]
}

그렇다면 혹시 다시 스키마가 꼬이게 되거나, 이미 적용된 schema를 사용하고 싶지 않다면 어떻게 해야할까

• 강제로 spark sql을 추론하고 싶은 경우, 'spark.sql.sources.schema.numParts' 설정만 지우고 다시 spark SQL에서 추론하도록 설정을 주면 변경되는 걸 확인할 수 있었다. 이 부분 처리하는 코드도 찾아볼 수 있었다.
  https://github.com/apache/spark/blob/master/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveExternalCatalog.scala#L1390-L1391

  # 아래 설정만 지우면 schema 안 먹는다.
  ALTER TABLE sg_cardad_clk UNSET TBLPROPERTIES ('spark.sql.sources.schema.numParts');

• field 명 변경은 위의 schema json을 원하는 대로 수정한 후에 아래와 같은 명령어를 통해 hive table 속성을 바꿀 수 있다.

  ALTER TABLE `tablename` SET TBLPROPERTIES ('spark.sql.sources.schema.part.0'="escaped처리된 json string";

이 옵션을 default 그대로 쓰는 것이 좋은 것일까?

어떻게 추론하는 지에 관한 내용은 아래 spark.sql.hive.convertMetastoreParquet에서 간략하게 살펴볼 수 있다. 더 자세한 내용은 위 코드 내용을 좀 더 보면 될 것 같다.

• inferSchema 코드: ParquetUtils.scala

parquet이나 hive table에 변경이 없다면 추론 schema가 정확하게 동작한다. 그러나 실제로 추론된 필드 스키마를 살펴보니 hive field: parquet field명으로 1대 1매핑이 되어있지 않는 걸 확인할 수 있었다. 추론 로직을 더 살펴봐야겠지만 hive/parquet schema가 변경되는 경우 의도치 않은 동작을 할 확률이 높아보인다. schema 변경이 있다면 이렇게 사용하는 케이스는 가능하면 피하는 것이 좋을 것 같다는 생각을 하며 위에서 소개했던 config document 번역으로 글을 마친다.

Config details from docs

spark.sql.hive.convertMetastoreParquet

Hive metastore parquet 테이블을 읽거나 쓸 때, Spark SQL는 더 나은 퍼포먼스와, 추가적인 조정(reconciliation)을 위해서 HiveSerDe 대신 자신의 parquet support를 쓰도록 시도하게 된다.

테이블 스키마 처리 관점에서 Hive와 Parquet에는 두 가지 주요 차이점이 있다.

1. 하이브는 대소문자를 구분하지 않고, parquet은 구분한다
2. hive는 모든 컬럼을 nullable로 생각하지만, parquet에서 null 허용 여부는 중요하다. 이 때문에, Hive metastore parquet table을 spark sql parquet table로 변환할 때 우리는 조정이 반드시 필요하다. 조정 규칙은 다음과 같다.
   1. 두 스키마에서 이름이 같은 필드는 null 허용 여부에 관계 없이 동일한 데이터 유형을 가져야 한다. 조정된 필드의 데이터 유형은 Parquet 의 데이터 타입을 가져야 하므로 nullable이 고려된다.
   2. 조정(reconciliation)된 스키마는 hive metastore에서 정의된 필드가 정확하게 포함된다.
      • parquet 에서만 나타난 필드는 조정된 스키마에서 삭제(drop)됨
      • hive metastore 스키마에 있는 필드는 조정된 스키마에서 nullable필드로 추가됨

spark.sql.hive.caseSensitiveInferenceMode

spark.sql.hive.caseSensitiveInferenceMode (INFER_AND_SAVE — default) 

위의 속성은 추론된 schema를 사용할 지 말지, 혹은 미리 저장해두고 쓸 지에 관한 옵션이다.

옵션:

• INFER_AND_SAVE (default - 기본 데이터 파일에서 대소문자를 구분하는 스키마를 추론하고 테이블 속성에 다시 기록)
• INFER_ONLY (스키마를 추론하지만 테이블 속성에 쓰려고 시도하지 않음)
• NEVER_INFER (추론 대신 대소문자를 구분하지 않는 메타스토어 스키마를 사용)

이미 저장된 schema가 있는 경우 NEVER_INFER를 사용하더라도 테이블 속성의 추론된 스키마를 사용한다.

Ref

• https://spark.apache.org/docs/2.3.2/sql-programming-guide.html#hive-metastore-parquet-table-conversion
• https://spark.apache.org/docs/2.2.1/sql-programming-guide.html#upgrading-from-spark-sql-21-to-22 이 문서는 migration 문서지만 내용은 spark.sql.hive.caseSensitiveInferenceMode 에 관한 내용임