Apache Sqoop [4] - 유저가이드3 [ sqoop-import 2 ]

1. 최신행 Import 

스쿱은 최신행만 가져올 수 있는 기능을 제공한다. 예를 들어 2020-10-14일 이후의 열들을 가져오고 싶을 경우 사용될 수 있다.

인자	설명
--check-column (col)	가져올 행을 결정할 기준 열을 지정 (CHAR/NCHAR/VARCHAR/VARNCHAR/ LONGVARCHAR/LONGNVARCHAR 타입은 기준열로 지정불가)
--incremental (mode)	가져올 행을 결정하는 방법을 지정
--last-value (value)	가져올 열의 값에 대한 최대값을 지정

--incremental 인자는 두가지 타입을 지정할 수 있는데 하나는 append와 lastmodified이다.

[append]

append사용시 --check-column을 지정하고(예시에선 ID로 지정) --last-value를 500으로 지정한다. 즉 id가 500보다 큰 행을 가져오는 결과 값을 가진다.

sqoop import --connect jdbc:mysql://localhost:3306/dbname --table tt1 --username root -P --check-column id --incremental append --last-value 500

[lastmodified]

lastmodified사용시 수정된 열의 값이 --last-value로 지정된 타임스탬프보다 최근인 열을 가져온다. 

sqoop import --connect jdbc:mysql://localhost:3306/dbname --table tt1 --username root -P --check-column update_date --incremental lastmodified --last-value '2020-08-24 22:04:56.0'

2. 파일포맷

일반적으로 'delimited text'와 'sequenceFiles' 포맷을 지원한다. 

[ Delimited text ]

Delimited text형식이 default이며 --as-textfile 인자로 명시적으로 지정할 수 도 있다. 'ROW FORMAT' 옵션을 활용, delimited를 설정해 HIVE에서도 활용 할 수 있다. 

1,here is a message,2010-05-01
2,happy new year!,2010-01-01
3,another message,2009-11-12

[ SequenceFiles ]

SequenceFiles는 이진 형식 저장타입이다. 모든 데이터의 정확한 저장을 지원하며 Java클래스로 표현할 수 있어 MapReduce프로그램에서 사용되는 데이터를 저장하기에 적합하다. 그 중 대표적인게 다른 프로그래밍 언어로 작성된 데이터도 확장가능하여 효율적인 Avro데이터타입이다. 

[ 압축 ]

default는 비압축형식인데 압축을 위해선 --compress인수를 활용할 수 있으며, 하둡압축코덱을 위한 --compression-codec도 지원한다.

3. Large Data 처리

스쿱은 큰 데이터(BLOB, CLOB)를 처리하기 적합하다. BLOB는 바이너리 데이터로 RDB외부에 저장하기 위한 데이터타입이다. CLOB는 문자열 데이터를 저장하기 위한 타입.

스쿱은 이를 처리하기 위해 메모리에 전부 올려놓치 않고 스트리밍 방식으로 처리할 수 있게 인라인으로 저장한다. 인라인 데이터를 통해 모든 데이터를 액세스 할 수 있다. 

사용할 수 인자는 아래와 같다.

인자	설명
--enclosed-by <char>	필수 포함 필드 문자 설정
--escaped-by <char>	이스케이프 문자 설정
--fields-terminated-by <char>	필드 구분 문자 설정
--lines-terminated-by <char>	줄 끝 문자(개행) 설정
--mysql-delimiters	mysql 기본 구분자 사용 ( 줄:\n, 이스케이프:\, 선택적 포함:' )
--optionally-enclosed-by <char>	필드 동봉 문자 설정

escape character지원 문자: \b, \n, \r, \t, \", \\' \\

예제 데이터셋

Some string, with a comma.
Another "string with quotes"

스쿱명령어

$ sqoop import --fields-terminated-by , --escaped-by \\ --enclosed-by '\"' ...

결과 값

"Some string, with a comma.","1","2","3"...
"Another \"string with quotes\"","4","5","6"...

SQL - Oracle[1] SELECT문

1. 개요

• sql문은 대소문자를 구분하지 않는다.
• sql문은 한 줄 또는 여러 줄에 입력할 수 있다.
• sql문은 선택적으로 세미콜론(;)으로 끝날 수 있습니다. 세미콜론은 여러 sql문을 실행하는 경우에 필요하다.

2. SELECT

[ 산술연산자 사용 ]

SELECT last_name, salary, salary + 300

FROM employees;

• 곱하기와 나누기는 더하기와 빼기보다 먼저 수행
• 동일한 우선 순위를 갖는 연산자는 왼쪾에서 오른쪽으로 평가됨
• 괄호는 기본 우선 순위를 재정의하거나 명령문을 명확히 하기 위해 사용

[ Null 값 ] 

SELECT last_name, job_id, salary, commission_pct

FROM employees;

• Null은 사용할 수 없거나, 할당되지 않았거나, 알 수 없거나,적용할 수 없는 값이다.
• Null은 0이나 공백과는 다릅니다. 0은 숫자이고 공백은 문자이다.
  
• 모든 데이터 유형의 열은 null을 포함 할 수 있으나 primary key열은 null을 사용할 수 없다.
• null값을 포함하는 산술식은 null이다. null*500 = null

[ 연결 연산자 ]

SELECT last_name || job_id AS "Employees"

FROM employees;

last_name=kim이고 job_id는 123이라면 kim123 으로 출력된다.

[ 리터널 문자열 사용 ]

SELECT last_name || ' is a ' || job_id AS "Employee Details"

FROM employees;

출력 : kim is a 123


[ 중복 행 ]

SELECT DISTINCT department_id

FROM employees;

[ 테이블 구조 표시 ]

DESCRIBE employees;

DESCRIBE 명령을 사용하여 테이블의 구조를 표시합니다.

HADOOP HDFS [2] - CDH을 통한 리밸런싱 (Rebalancing)

1. HDFS Balancers

기존 클러스테어 새로운 DataNode를 지속적으로 추가하기 떄문에 HDFS데이터가 항상 균일하게 배포되지 않을 수 있다. 따라서 HDFS는 블록 배치를 분석하고 데이터균형을 조정하는 밸런서 유틸리티를 제공한다. 밸런서는 단일 DataNode의 개별 볼륨 간에 균형을 유지하지는 않는다.

2. CDH를 이용한 HDFS Balancer Configuration

[ 밸런서 임계값 ]

HDFS Balancer는 일반적으로 CDH서비스가 설치될 때 추가된다. 밸런서의 기본 임계값은 10%로, 각 데이터노드의 디스크 사용량이 클러스터의 전체 사용량과 10% 이하로 차이가 남을 의미한다. 

예를 들어 데이터노드의 전체 사용량이 40%라면 각 클러스터는 30~50% 스토리지 용량을 사용하고 있으면 된다. 

임계값을 변경하려면

1. HDFS서비스의 구성탭 선택

2. 범위->Balancer 선택

3. 범주->기본 선택

4. 재조정임계값 설정값 변경

5. 변경 후 저장

[ Concurrent Move 구성 ]

dfs.datanode.balance.max.concurrent.moves속성은 데이터 이동시 데이터노드 밸런서가 사용하는 최대 스레드 수를 설정한다.

값을 올리면 밸런싱 프로세스가 더 빨리완료될 수 있다. 반대로 값을 줄이면 재조정이 더디지만, 데이터노드의 다른 작업에 리소스를 더 많이 할당 할 수 있다.

값 변경은 아래와 같다.

1. HDFS서비스의 구성탭 선택

2. hdfs-site.xml에 대한 DataNode 고급 구성 스니펫(안전 밸브) 검색

3. Value값 변경

<property>
  <name>dfs.datanode.balance.max.concurrent.moves</name>
  <value>50</value>
</property>

4. 변경 후 저장

[ 밸런서에 대한 권장 구성 ]

HDFS밸런서를 백그라운드에서 실행할지 최대속도는 어느정도일지 권장되는 값은 아래와 같다.

Property	Values for Running the Balancer in the Background	Value for Running the Balancer at Maximum Speed
DataNode
dfs.datanode.balance.bandwidthPerSec	10 MB	10 GB
Balancer
dfs.balancer.moverThreads	1000	20000
dfs.balancer.max-size-to-move	10 GB	100 GB
dfs.balancer.getBlocks.min-block-size	10 MB	100 MB

* Ref : https://docs.cloudera.com/documentation/enterprise/latest/topics/admin_hdfs_balancer.html#cmug_topic_5_14

3. 밸런서 실행

1. HDFS서비스 선택

2. 재조정 역할이 있는지 확인

3. 작업->재조정 실행

4. 블록 크기 구성

아래와 같은 설정을 통한 블록크기 조정으로 밸런싱 시 작업 효율을 늘릴 수 있다.

• dfs.balancer.getBlocks.size : Block Metadata Batch Size
• dfs.balancer.getBlocks.min-block-size : Minimum Block Size

Apache Sqoop [3] - 유저가이드2 [ sqoop-import ]

 

[ Sqoop-Import ]

1. 일반인수 (Common arguments)

• --connect <jdbc-uri> : JDBC 접속URI 지정
• --connect-manager <class-name> : 사용할 연결 관리자
• --driver <class-name> : JDBC driver class 지정
• --hadoop-mapred-home <dir> : Override $HADOOP_MAPRED_HOME
• --help : 도움말
• --password-file : 인증관련 파일 경로
• -P : 콘솔에서 비밀번호를 읽을 때
• --password <password> : 인증패스워드 SET
• --username <username> : 유저이름 SET
• --verbose : working시 더 많은 정보 출력
• --relaxed-isolation : 매퍼에 대해 커밋되지 않은 읽기 데이터를 가져온다. HDFS에 저장되기전 데이터를 컨트롤하기 위한 인수

2. 데이터베이스 서버 접속

접속예시

$ sqoop import --connect jdbc:mysql://database.example.com/employees \
    --username aaron --password 12345

비밀번호가 저장된 파일을 이용하여 접속

$ sqoop import --connect jdbc:mysql://database.example.com/employees \
    --username venkatesh --password-file ${user.home}/.password

또는 -P 인수를 통해 비밀번호를 콘솔에서 입력할 수 있다.

$ sqoop import --connect jdbc:mysql://database.example.com/employees \
    --username aaron -P

password: *****

Sqoop은 몇몇 데이터베이스(Mysql)들은 기본적으로 지원한다. 지원하는 데이터베이스들의 JDBC가 설치되어 있기 때문이다. 

그러나 다른 데이터베이스를 사용하기위해 JDBC를 설치해야 하는 경우 해당 .jar파일을 $SQOOP_HOME/lib 경로에 추가시켜야한다. (데비안-/usr/lib/sqoop/lib) 

3. 유효성관련 인수 (Validation arguments)

인수	인수
--validate	복사된 데이터의 유효성 검사 사용, 단일 테이블 복사본만 지원
--validator <class-name>	특정 vaildator클래스 지정
--validation-threshold <class-name>	사용할 유효성 검사 임계값 클래스 지정
--validation-failurehandler <class-name>	사용할 유효성 검사 오류 처리기 클래스 지정

4. Import 관련 인수 (Import control arguments)

인수	인수
--append	기존 HDFS의 데이터세트에 데이터 추가
--as-avrodatafile	Avro 데이터파일로 가져오기
--as-sequencefile	SequenceFiles 데이터파일로 가져오기
--as-textfile	text파일로 가져오기 (default)
--as-parquetfile	Parquet 데이터파일로 가져오기
--boundary-query <statement>	분할 작성에 사용한 경계쿼리
--columns <col,col,col…>	테이블에서 가져올 Columns
--delete-target-dir	target directory가 있을 경우 삭제
--direct	데이터베이스에 대한 직접 컨넥터 사용
--fetch-size <n>	데이터베이스에서 한 번에 읽을 항목 수
--inline-lob-limit <n>	inline LOB의 최대크기 설정
-m,--num-mappers <n>	병렬로 작업 할 map tasks 수 설정
-e,--query <statement>	쿼리로 가져올 데이터 지정
--split-by <column-name>	컬럼을 구분하는데 사용되는 유닛 ex) ","
--split-limit <n>	정수 및 날짜 열에 대한 분할크기지정
--autoreset-to-one-mapper	primary key나 분할 기준열이 없는 경우 자동적으로 매퍼가 조정
--table <table-name>	읽을 테이블 지정
--target-dir <dir>	HDFS 디렉토리 경로
--temporary-rootdir <dir>	import될 동안 생성할 임시 HDFS 디렉토리 경로
--warehouse-dir <dir>	지정된 테이블의 상위 HDFS 디렉토리
--where <where clause>	작성할 WHERE절
-z,--compress	압축사용
--compression-codec <c>	사용할 하둡 아축 코덱 (default gzip)
--null-string <null-string>	문자열컬럼이 null일 떄 쓸 문자열
--null-non-string <null-string>	비문자열컬럼이 null일 때 쓸 문자열

예시1)

$ sqoop import \
  --table employ \ \
  --columns "id,name,age" \
  --where "id > 300" 

예시2)

$ sqoop import \
  --query 'SELECT a.*, b.* FROM a JOIN b on (a.id == b.id) WHERE $CONDITIONS' \
  --split-by a.id --target-dir /user/foo/joinresults

--query 사용시 WHERE절 뒤에 "AND \$CONDITIONS" 문구를 추가하여 이 문장이 조건절을 나타내고 있다는 표시를 해줘야 한다.

Ex) "SELECT * FROM x WHERE a='foo' AND \$CONDITIONS"

또한 복잡한 조인쿼리는 오류가 생길 수 있다.

5. 병렬 제어 (Contorlling Parallelism)

-m 또는 --num-mappers 인수로 제어 할 수 있다. 병렬시 분할 기준은 기본키 또는 지정한 키값이다. 예를 들어 기본키가 0-1000이고, -m 4(맵리듀스 4개)로 설정하면 각 맵태스크는 250개씩 나누어 작업을 한다.

기본 키값이 균일하게 분포되지 않으면 작업 불균형이 생긴다. --split-by인수를 사용해 명시적인 컬럼을 선택한다.

--split-limit옵션을 사용하여 -m 또는 --num-mapers를 재정의 할 수 있다. --split-limit옵션은 분할된 태스크의 크기를 제한한다. 같은 예시로 위의 1-1000개의 기본키가 있는 테스크를 똑같이 -m 4, 그리고 --split-limit옵션은 task 200개로 지정하면 맵퍼수가 하나 증가하며 5개로 재조정된다.

6. 저장위치

 /shared/foo 에 저장된다

$ sqoop import --connnect <connect-str> --table foo --warehouse-dir /shared \
    ...

$ sqoop import --connnect <connect-str> --table foo --target-dir /shared/foo\
    ...

--target-dir와 --warehouse-dir은 같이 쓸 수 없다. 또한 대상 디렉토리가 HDFS에 이미 존재하는 경우 작동하지 않는다. 그럴 경우 대상 디렉토리 이름을 변경하던가 --delete-target-dir인수를 사용하여 기존 파일을 지운후 가져오는 방식을 사용해야 한다.

Apache Sqoop [2] - 유저가이드1 [ Sqoop Tools ]

[ 1. Sqoop Tools ]

스쿱은 도구모음이다. 커맨드라인에 커맨드와 인자를 입력하여 사용한다. 다른 프로그램없이 자체 스쿱소스로 컴파일 될 경우, bin/sqoop 프로그램을 실행하여 sqoop을 사용할 수 있다. 스쿱 패키지 배포(apache bigtop과 함께 제공된 RPM등) 사용자는 프로그램을 /usr/bin/sqoop으로 설치하게 된다. 

Sqoop의 사용설명서는 'help'로 볼 수 있다.

$sqoop help

Available commands:
  codegen            Generate code to interact with database records
  create-hive-table  Import a table definition into Hive
  eval               Evaluate a SQL statement and display the results
  export             Export an HDFS directory to a database table
  help               List available commands
  import             Import a table from a database to HDFS
  import-all-tables  Import tables from a database to HDFS
  import-mainframe   Import datasets from a mainframe server to HDFS
  job                Work with saved jobs
  list-databases     List available databases on a server
  list-tables        List available tables in a database
  merge              Merge results of incremental imports
  metastore          Run a standalone Sqoop metastore
  version            Display version information

특정 인자에 대한 설명은 아래와 같이 메뉴얼을 확인할 수 있다.

$ sqoop help import

1-1. 커맨드별칭(Using Command Aliases)

sqoop import를 sqoop-import로, sqoop export를 sqoop-export 별칭으로 사용가능하다.

1-2. 하둡 설치 제어 (Controlling the hadoop Installation)

스쿱을 통해 하둡 bin/hadoop 스크립트를 실행 할 수 있다. 여러 개의 하툽이 설치되어 있는 경우 $HADOOP_CONMANY_HOME 또는 $HADOOP_MAPRED_HOME 환경변수를 이용해 하둡을 선택하여 설치한다.

예시 :

$ HADOOP_COMMON_HOME=/path/to/some/hadoop \
  HADOOP_MAPRED_HOME=/path/to/some/hadoop-mapreduce \
  sqoop import --arguments...

또는 :

$ export HADOOP_COMMON_HOME=/some/path/to/hadoop
$ export HADOOP_MAPRED_HOME=/some/path/to/hadoop-mapreduce
$ sqoop import --arguments...

1-3. 인수사용 ( Using Generic and Specific Arguments )

* 일반 인수

• --connect <jdbc-uri> : 접속할 JDBC 주소
• --connect-manager <class-name> : 사용할 연결 관리자
• --driver <class-name> : JDBC driver class 지정
• --hadoop-mapred-home <dir> : Override $HADOOP_MAPRED_HOME
• --help : 도움말
• --password-file : 인증관련 파일 경로
• -P : 콘솔에서 비밀번호를 읽을 때
• --password <password> : 인증패스워드 SET
• --username <username> : 유저이름 SET
• --hadoop-home <dir> : Override $HADOOP_HOME

* 하둡 관련 인수

• -conf : 특성 어플리케이션 설정파일 지정
• -D <property=value> : 주어진 property-value를 사용한다.
• -fs <local | namenode:port> : 네임노드 지정
• -files <comma separated list of files> :  지정할 맵리듀스 클러스터 파일
• -libjars <comma separated list of jars> : 지정할 jar파일 경로
• -archives <comma separated list of archives> : 지정할 아카이브

-conf, -D와 같은 인수는 --connect와 같은 인수 앞에 같이 사용 할 수 있다. 하둡인수는 단일 대쉬문자(-)로 표현되고 도구별 인수는 (--)더블 대쉬로 표현된다.

-files, -libjars, -archives 인수는 sqoop과 함께 사용되지 않지만, 하둡 내부 인수로 시스템 일부에 포함되어있다. 

1.4 옵션 파일을 이용한 명령어

$sqoop import --connect jdbc:mysql://localhost/db --username john --table TT

$sqoop --options-file /user/home/work/import.txt --table TT

*import.txt 내용

import
--connect
jdbc:mysql://localhost/db
--username
foo

Apache Hive [1] - 하이브 쿼리 퍼포먼스(속도)를 높이는 방법

[ 1. Tez Engine을 사용 ]

Apache Tez는 데이터 처리 작업의 복합 비순환 방향 그래프(DAG)를 생성하는 프레임 워크이다. 하둡의 YARN에 의해 매니징되면, 맵리듀스의 능력을 유지함과 동시에 대용량데이터 처리 속도를 높힌다.

[ Tez와 MR아키텍처 비교 ]

하이브 엔진으로 Tez를 사용하여 쿼리속도를 높힌다.

* set hive.execution.engine=tez;

[ 2. 백터화 (Vectorization) ]

백터화는 작업시 단일 행을 가져오지 않고 한번의 작업으로 1,024개의 행을 가져온다. filter, join, aggregation과 같은 작업의 성능을 향상시킨다.

아래 커맨드로 환경구성을 한다.

* set hive.vectorized.execution.enabled=true;

* set hive.vectorized.execution.reduce.enabled=true;

[ 3. ORC파일, ORC SerDe ]

ORC(Optimized Row Columnar)파일 형식은 데이터원본 크기의 78%까지 줄여 저장시 매우 효율적인 방법을 제공한다. 

[ https://blog.cloudera.com/orcfile-in-hdp-2-better-compression-better-performance ]

초기 컬럼(row)단위로 저장되는 RC파일형식은 읽기과정에서 조인의 수행시간이 길었다. 그래서 컬럼과 함께 인덱스까지 저장하는 ORC파일형식의 등장으로 데이터의 압축과 함께 읽고 쓰는데 향상된 퍼포먼스를 제공할 수 있게 되었다.

* 
Create Table Emp ( empId int, empName varchar(100), age int)

STORED AS ORC tblproperties("compress.mode"="SNAPPY") 

[ 4. 파티셔닝( Partitioning ) ]

하이브는 RDBMS와 다르게 디렉토리로 파일을 분리하여 저장한다. 분리되는 기준은 어떻게 테이블을 파티션을 했느냐에 따라 달라지며 쿼리성능을 좌우한다. 

* 파티션 생성예제 :

CREATE TABLE tb1( col1 STRING ) PARTITIONED BY ( yymmdd STRING );

[ 5. 버켓팅(Bucketing) ]

버켓팅은 데이터를 파일별로 나누어 저장한다. 지정된 컬럼의 값이 해쉬처리되어 조인시 쿼리속도가 더 빠르다. 파티션을 버켓으로 더욱 세분화하여 성능을 향상시킬 수 있다.

* 버켓 생성예제 : 

CREATE TABLE tb1 (col1 STRING ) CLUSTERED BY (col1) INTO 50 BUCKETS;

[ 6. CBO(Cost-Based Optimizer) ]

하이브의 CBO는 쿼리처리의 핵심 컴포넌트이다. Apache Calcite의 구동되며 쿼리질의에 대한 비용을 최적화하고 계산한다. 

Calcite는 쿼리재작성, Join수정 및 제거 등으로 다양한 최적화 작업을 한다. Calcite를 거친 논리작업은 Hive에의해 tree로 변환되어 물리적으로 최적화되고 Tez작업으로 바뀐 후 Hadoop클러스터에서 실행된다.

비용기반 최적화(CBO)를 사용하려면 아래 매개변수를 설정한다.

* set hive.cbo.enable=true;

* set hive.compute.query.using.stats=true;

* set hive.stats.fetch.column.stats=true;

* set hive.stats.fetch.partition.stats=true;

zeppelin [ 4 ] - crontab

[ zeppelin-site.xml 설정 ]

sudo vi /usr/local/zeppelin/conf/zeppelin-site.xml

-- 추가 -- 

<property>

  <name>zeppelin.notebook.cron.enable</name>

  <value>true</value>

  <description>Notebook enable cron scheduler feature</description>

</property>

[ web UI crontab 설정 ]

Cron Expressions

seconds - minutes - hours - day of month - month - day of week - year

예시

0 0/5 * * * ? : 매 5분

10 0/5 * * * ? : 매 5분 10초 ( 10:00:10, 10:05:10 )

0 30 10-12 ? * WED,FRI : 10:30, 11:30, 12:30 매주 수요일,금요일

0 30 8 5,20 * ? : 5일~20일 매일 8시 30분

* 재플린Doc : https://zeppelin.apache.org/docs/0.8.2/usage/other_features/cron_scheduler.html#setting-up-a-cron-scheduler-on-a-notebook

* 크론 표현 : https://www.quartz-scheduler.org/documentation/quartz-2.3.0/tutorials/tutorial-lesson-06.html