Hive partition table로 DW를 구축할 때 고려할 점(upsert)

과거에 팀장님께서도 한번 주문했던 내용인데 하둡에 저장된 과거 데이터의 update 시나리오를 고민했던 적이 있다. 당시 결국 만족할만한 방법이 없어서 drop했던 내용인데 그 기억을 살려 hive를 기준으로 다시 포스팅을 해본다.

대부분 HDFS에 적재된 데이터는 Hive로 접근한다. HDFS 데이터를 가지고 DW를 구축할 때 Hive가 가장 편하긴 하지만 HDFS 특징으로 인해 Hive에서의 약간의 제약이 있다. 파일시스템이기 때문에 일반적인 DB처럼 row 1개를 delete, update를 하기 어렵기 때문이다.

대부분의 경우 하이브 테이블은 파티셔닝을 한다. 데이터의 양이 많고 매일 적재되기 때문에 파티셔닝은 필수다. 하지만 이런 경우 소급이 문제가 된다.

가령 과거 데이터를 새로 만들어서 insert overwrite table target_tbl partition(col='yyyymmdd') select * from source_tbl 형태로 데이터를 적재해보자. 이런 경우 upsert로 동작하지 않는다. 해당 파티션의 데이터를 delete 후 새로 엎어치는 작업을 한다. 즉 delete&copy 형태로 동작한다. 

이런 경우 그럼 지워지면 안되는 데이터도 날아가기 때문에 결과적으로는 통으로 데이터를 만들어줘야한다. 이런 경우 old 데이터와 new 데이터를 full outer join을 해서 통으로 새로 다시 만들거나 new_tbl left join old_tbl+ union all old_tbl not exists new_tbl 형태로 통으로 다시 만들어줘야한다.

insert는 해당 파티션에 단순 append이기 때문에 단순 데이터를 적재하는데 쓸수는 있지만 insert overwrite로 밀어넣는 형태라면 해당 파티션이 새로 적재된다는 사실이다.

파일시스템이라서 당연한 결과이기는 하지만 hive를 어떠한 형태로 사용하더라도 hdfs를 사용하기 때문에 방법이 없다. 또한 인덱스의 효과도 일반 DB와 비교해서 크지 않다.

그럼 결과적으로 통으로 새로 만들어야하는가? "그렇다."

그럼 어느 부분에서 효율을 낼 수 있나? 현재로서는 spark를 사용해서 연산 속도를 높이는 것 외에는 없어보인다.

다른 방법이기는 하지만 파티션을 일자 외에도 다른 컬럼을 추가해서 아주 잘(?) 사용한다면 방법이 있을 수도 있다. 하지만 권장하지는 않는다!

6) 빅데이터 플랫폼 아키텍처에 대하여.. 다른 팀과 협업 시 구성하면 좋은 프레임워크(hive, hue)

무슨 내용을 쓸까하다가 보안적인 부분에 대해서 포스팅을 안했기 때문에 이번에는 이 부분에 대해서 다뤄보려고 한다. 최근 깃랩에 사이드프로젝트를 만들어보니 토큰 발급이 필수로 바뀌어서 문득 다음 포스팅 주제도 보안적인 부분을 다루면 좋겠다고 생각했다.

선택사항이고 개인적으로 보안적인 부분은 최전방에서 최대한 막고 내부 동료들끼리는 개발 효율을 위해 최대한 풀어줘야한다는 생각이지만 이는 업에 따라 법적인 문제(생각보다 고려할 부분이 많다)도 있기 때문에 잘 고려해서 해야한다. 단지 이 포스팅은 한가지 시나리오일 뿐이다.

상황은 하둡 데이터를 접근해야한다. 당연히 관리자는 모두 접근이 가능하겠지만 만약 다른 부서 사람들이 접근을 해야하는 경우는 어떻게 해야할까? 데이터를 직접 뽑아줄 수도 있지만 분석가는 데이터를 불러와서 모델링을 해야하기 때문에 결국 open을 해야하는 이슈가 있다.

하둡 자체에 kerberos 인증이 있기는 하지만 우리가 로그인할 때의 수준이지 데이터 접근에 대한 개념이 아니다.

그럼 어떻게 할까.

1차적으로 하둡 페이지들에 대한 개인의 접근은 모두 막는다. 즉 모든 방화벽을 열어주지 않는다. 굳이 열어준다면 리소스 매니저만 열어준다.

그럼 사람들이 데이터는 어떻게 접근하는가?

hue만 열어준다. (hue는 절대 스케쥴러 용도가 아닌 단순 파티션이나 데이터 확인용!!) 

그럼 단지 쿼리만 날릴 수 있겠지만 불필요한 접근이 없게 될 것이다.

그리고 개발이나 배치는 hive를 통해서만 접근하도록 한다. 즉 개발서버에서 hive만 접근가능하게 열어주고 hive 쿼리만 날릴 수 있게 한다. 어차피 데이터는 쿼리로 접근할테니.. 중요한 것은 hdfs에 direct로 접근하지 못하게 하는 것이다.

체감되는 트랜드 순서가 MR -> Tez -> Spark인 것 같은데 hive on spark를 구성하면 좋겠지만 MR이든 Tez든 Spark든 모두 Spark로 처리할 필요는 없고 데이터의 사이즈에 따라서 MR이나 Tez를 써야하는 경우도 있다.

갑자기 다른데로 샌것같은데 아무튼 이번 포스팅은 보안(?)이라기보다는 어쨌든 아무나 접근 못하게 하는 방법(?)을 제안해보았다.

결론은 개인 컴퓨터에선 hue만 열어줘도 된다.

Hive Error) Cannot insert into target table because column number/types are different

Hive에 Insert를 할 때 이런 에러를 볼 수 있다. 

target table은 partition table이고 source 테이블보다 컬럼이 한개(div_col)가 더 많다. 

대충 Insert 구문은 이런식으로 작성을 했다. 

INSERT OVERWRITE TABLE TARGET_TBL_NMPARTITION(partition_col='yyyymm') SELECT '0' as div_col, * FROM SOURCE_TBL_NM

"Cannot insert into target table because column number/types are different : Table insclause-0 has N columns, and the N+1 columns are partitioned and we not required any filters we have to dump/store from non partitioned table to partitioned table."

넣고자 하는 데이터에 SELECT * FROM 구문이 문제가 되는 것으로 partition 컬럼이 return 되면서 컬럼이 한개가 더 +1 되는 상황이다.

그래서 이런 경우에는 타겟 테이블 컬럼 이름을 Select 절에 다 써줘서 갯수를 맞춰줘야한다.

즉 INSERT OVERWRITE TABLE TARGET_TBL_NMPARTITION(partition_col='yyyymm') SELECT '0' as div_col, Col1, Col2, Col3... FROM SOURCE_TBL_NM

컬럼명이 천개가 넘어가면 설계를 바꾸는것을 권장하고싶다. 굳이 통으로 한개를 들고다닐 필요가 없다면 말이다. 

교훈 : 테이블의 컬럼 갯수는 적당히 쓰자.. 테이블당 컬럼이 1000개가 넘어가면 골때린다.. hdfs를 사용하는 Hive가 Column store가 된다는 일은 앞으로도 없을테니까...

MSSQL(PDW)에서 ORC 타입으로 하둡에 데이터를 내리고 하이브 테이블 생성하기

MSSQL및 PDW(parallel data warehouse)에는 하둡과 연결고리가 잘 되어있다.
이들이 지원하는 PolyBase라는 기능은 RDBMS의 데이터와 하둡에 저장된 데이터를 조인할 수 있도록 하는데 예를 들어 하둡에 저장되어 있는 데이터를 RDBMS로 가져오고 둘을 조인해서 원하는 측정값을 뽑아내기에 매우 유용하다.

위 처럼 HDFS -> MSSQL(PDW) 로의 데이터 이동이 아닌 MSSQL(PDW) -> HDFS로 데이터를 내리는 것도 가능하며 예제는 다음과 같다. (PDW 기준으로 예제를 작성하였다.)

다음 예제는 ORC 타입으로 데이터를 내리는 방법이고 파케이(parquet)나 텍스트 등 다양한 타입 및 압축(Gzip, Snappy 등)으로 데이터 포맷을 생성할 수 있다.

먼저 데이터 소스랑 파일 포맷을 생성한다. (각 DB별로 생성할 수 있다.)


USE TEST_DB;
-- 데이터 소스 생성
IF EXISTS (SELECT * FROM sys.external_data_sources WHERE name = 'HDFS_DS_SNAPSHOT')
   DROP EXTERNAL DATA SOURCE HDFS_DS_SNAPSHOT

CREATE EXTERNAL DATA SOURCE HDFS_DS_SNAPSHOT
WITH (TYPE = HADOOP
, LOCATION = N'hdfs://XXX.XXX.XXX.XXX:8020'
, RESOURCE_MANAGER_LOCATION = N'XXX.XXX.XXX.XXX:8050')
GO

-- 파일 포맷 생성
IF EXISTS (SELECT * FROM sys.external_file_formats WHERE name = 'HDFS_FF_ORC')
   DROP EXTERNAL FILE FORMAT HDFS_FF_ORC

CREATE EXTERNAL FILE FORMAT HDFS_FF_ORC
WITH (
    FORMAT_TYPE = ORC,
    DATA_COMPRESSION = 'org.apache.hadoop.io.compress.SnappyCodec'
);

-- 생성된 데이터 소스 및 파일 포맷 확인

SELECT * FROM sys.external_data_sources

SELECT * FROM sys.external_file_formats

-- 하둡에 월 파티션으로 데이터 내리기

DECLARE @DT VARCHAR(8) = '20170101'
DECLARE @SQL NVARCHAR(4000);

WHILE(1=1)
BEGIN
      SET @SQL  = N' IF  OBJECT_ID(''TEST_DB..HDFS_TEST_TABLE_'+@DT+''') IS NOT NULL
                       DROP EXTERNAL TABLE HDFS_TEST_TABLE_' + @DT + '

      CREATE EXTERNAL TABLE HDFS_TEST_TABLE_' + @DT + '
      WITH (LOCATION=' + '''/apps/hive/xxxxxxxx/yyyyyyyy/zzzzzzzzz/TEST_TABLE/yyyymmdd=' + @DT + ''',
           DATA_SOURCE = HDFS_DS_SNAPSHOT,
           FILE_FORMAT = HDFS_FF_ORC,
           REJECT_TYPE = PERCENTAGE,
           REJECT_SAMPLE_VALUE = 10000,
           REJECT_VALUE = 0.1
      ) AS
      SELECT *
      FROM TEST_TABLE
      WHERE ' + N'DT' + ' >= ''' + @DT +  ''' AND ' + N'DT' +  ' < CONVERT(VARCHAR(8),DATEADD(MM,1,''' + @DT + '''),112) '

      --PRINT @SQL
      SET @DT = CONVERT(VARCHAR(8),DATEADD(MM,1,@DT),112)
      IF(@DT > '20180101')
        BREAK;
END

다음부터는 하이브에서 진행하며 위에서 생성한 HDFS를 사용한다.

-- 하이브에서 테이블 만들기

 CREATE EXTERNAL TABLE HDFS_TEST_TABLE

(

    `COL1` bigint ,

    `COL2` bigint ,

    `COL3` bigint ,

)

PARTITIONED BY (YYYYMMDD STRING)

ROW FORMAT DELIMITED

FIELDS TERMINATED BY ','

STORED AS ORC

LOCATION '/apps/hive/xxxxxxxx/yyyyyyyy/zzzzzzzzz/TEST_TABLE';

-- 파티션 적용하기
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170101');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170201');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170301');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170401');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170501');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170601');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170701');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170801');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20170901');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20171001');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20171101');
ALTER TABLE database_name.HDFS_TEST_TABLE ADD PARTITION (YYYYMMDD='20171201');


정리하면 MSSQL(PDW)에서 데이터 소스와 파일 포맷을 만들어주고 이를 활용해서 HDFS로 데이터를 내린다. 그리고 하이브에서 내린 데이터를 가지고 외부 테이블을 생성하면 된다.