Apache Hive

Apache Hive

Apache Hive는 데이터 웨어하우스 생태계의 중심으로 자리 잡았어요. 빅데이터 분석과 ETL을 위한 SQL 엔진일 뿐만 아니라, 데이터가 발견·정의·진화되는 데이터 관리 플랫폼이기도 해요. Flink는 Hive와의 두 가지 통합을 제공해요.

출처: 문서

본문

첫 번째 통합은 Hive의 Metastore를 Flink의 HiveCatalog와 함께 영구 카탈로그로 활용해, 세션 간에 Flink 특정 메타데이터를 저장하는 방식이에요. 예를 들어 사용자는 HiveCatalog를 사용해 Kafka나 Elasticsearch 테이블을 Hive Metastore에 저장하고, 나중에 SQL 쿼리에서 재사용할 수 있어요. 두 번째는 Flink를 Hive 테이블을 읽고 쓰는 대체 엔진으로 제공하는 것이에요.

HiveCatalog는 기존 Hive 설치와 "out of the box"로 호환되도록 설계되었어요. 기존 Hive Metastore를 수정하거나 테이블의 데이터 배치·파티셔닝을 변경할 필요가 없어요.

지원되는 Hive 버전

Flink는 다음 Hive 버전을 지원해요.

• 2.3.0, 2.3.1, 2.3.2, 2.3.3, 2.3.4, 2.3.5, 2.3.6, 2.3.7, 2.3.8, 2.3.9, 2.3.10
• 3.1.0, 3.1.1, 3.1.2, 3.1.3

Hive 자체도 버전별로 사용 가능한 기능이 다르며, 다음 사항은 Flink 때문이 아니라는 점에 주의하세요.

• Hive 내장 함수는 1.2.0 이상에서 지원돼요.
• PRIMARY KEY와 NOT NULL 같은 컬럼 제약은 3.1.0 이상에서 지원돼요.
• 테이블 통계 변경은 1.2.0 이상에서 지원돼요.
• DATE 컬럼 통계는 1.2.0 이상에서 지원돼요.
• ORC 테이블에 쓰기는 2.0.x에서 지원되지 않아요.

의존성 (Dependencies)

Flink 2.3 버전에 사용할 수 있는 커넥터는 (아직) 없어요. Hive 커넥터는 바이너리 배포에 포함되지 않아요. 클러스터 실행을 위해 어떻게 링크하는지는 여기를 참고하세요.

Hive와 통합하려면, Table API 프로그램이나 SQL Client의 SQL에서 통합이 동작하도록 Flink 배포의 /lib/ 디렉터리에 몇 가지 추가 의존성을 넣어야 해요. 또는 이 의존성들을 전용 폴더에 두고, Table API 프로그램이나 SQL Client에 각각 -C 또는 -l 옵션으로 클래스패스에 추가할 수 있어요.

Apache Hive는 Hadoop 위에 구축되므로, HADOOP_CLASSPATH 환경 변수를 설정해 Hadoop 의존성을 제공해야 해요.

export HADOOP_CLASSPATH=`hadoop classpath`

Hive 의존성을 추가하는 방법은 두 가지예요. 첫 번째는 Flink에 번들된 Hive jar를 사용하는 것이에요. 사용 중인 metastore의 버전에 따라 번들된 Hive jar를 선택할 수 있어요. 두 번째는 필요한 각 jar를 별도로 추가하는 것이에요. 두 번째 방법은 사용 중인 Hive 버전이 여기에 나열되지 않은 경우 유용할 수 있어요.

참고: 의존성 추가의 권장 방식은 번들된 jar를 사용하는 것이에요. 별도의 jar는 번들된 jar가 요구 사항을 충족하지 못할 때만 사용해야 해요.

번들된 hive jar 사용하기

다음 표는 사용 가능한 모든 번들된 hive jar를 나열해요. Flink 배포의 /lib/ 디렉터리에 하나를 선택해 넣으면 돼요.

사용자 정의 의존성 (User defined dependencies)

다른 Hive 주요 버전에 필요한 의존성은 아래를 참고하세요.

/flink-2.3.0
   /lib

       // Flink's Hive connector.Contains flink-hadoop-compatibility and flink-orc jars
       flink-connector-hive_2.12-2.3.0.jar

       // Hive dependencies
       hive-exec-2.3.4.jar

       // add antlr-runtime if you need to use hive dialect
       antlr-runtime-3.5.2.jar

/flink-2.3.0
   /lib

       // Flink's Hive connector
       flink-connector-hive_2.12-2.3.0.jar

       // Hive dependencies
       hive-exec-3.1.0.jar
       libfb303-0.9.3.jar // libfb303 is not packed into hive-exec in some versions, need to add it separately

       // add antlr-runtime if you need to use hive dialect
       antlr-runtime-3.5.2.jar

프로그램 maven

자체 프로그램을 구축한다면 mvn 파일에 다음 의존성이 필요해요. 결과 jar 파일에 이 의존성들을 포함하지 않는 것이 권장돼요. 위에서 설명한 것처럼 런타임에 의존성을 추가해야 해요.

  org.apache.flink
  flink-connector-hive_2.12
  2.3.0
  provided


  org.apache.flink
  flink-table-api-java-bridge_2.12
  2.3.0
  provided


    org.apache.hive
    hive-exec
    ${hive.version}
    provided

Hive에 연결하기 (Connecting To Hive)

카탈로그 인터페이스와 HiveCatalog를 통해 테이블 환경 또는 YAML 구성으로 기존 Hive 설치에 연결해요. 다음은 Hive에 연결하는 예시예요.

EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
TableEnvironment tableEnv = TableEnvironment.create(settings);

String name            = "myhive";
String defaultDatabase = "mydatabase";
String hiveConfDir     = "/opt/hive-conf";

HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir);
tableEnv.registerCatalog("myhive", hive);

// set the HiveCatalog as the current catalog of the session
tableEnv.useCatalog("myhive");

val settings = EnvironmentSettings.inStreamingMode()
val tableEnv = TableEnvironment.create(settings)

val name            = "myhive"
val defaultDatabase = "mydatabase"
val hiveConfDir     = "/opt/hive-conf"

val hive = new HiveCatalog(name, defaultDatabase, hiveConfDir)
tableEnv.registerCatalog("myhive", hive)

// set the HiveCatalog as the current catalog of the session
tableEnv.useCatalog("myhive")

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog

settings = EnvironmentSettings.in_batch_mode()
t_env = TableEnvironment.create(settings)

catalog_name = "myhive"
default_database = "mydatabase"
hive_conf_dir = "/opt/hive-conf"

hive_catalog = HiveCatalog(catalog_name, default_database, hive_conf_dir)
t_env.register_catalog("myhive", hive_catalog)

# set the HiveCatalog as the current catalog of the session
tableEnv.use_catalog("myhive")

execution:
    ...
    current-catalog: myhive  # set the HiveCatalog as the current catalog of the session
    current-database: mydatabase
    
catalogs:
   - name: myhive
     type: hive
     hive-conf-dir: /opt/hive-conf

CREATE CATALOG myhive WITH (
    'type' = 'hive',
    'default-database' = 'mydatabase',
    'hive-conf-dir' = '/opt/hive-conf'
);
-- set the HiveCatalog as the current catalog of the session
USE CATALOG myhive;

YAML 파일 또는 DDL로 HiveCatalog 인스턴스를 생성할 때 지원되는 옵션은 아래와 같아요.

DDL

Hive 테이블, 뷰, 파티션, 함수를 Flink 안에서 생성하려면 Hive dialect를 사용해 DDL을 실행하는 것을 권장해요.

DML

Flink는 Hive 테이블에 DML 쓰기를 지원해요. 자세한 내용은 Reading & Writing Hive Tables를 참고하세요.

더 알아보기 (Learn more)

• Hive dialect
• Reading & Writing Hive Tables
• Table API 개념