databaser.net: Hadoop

레이블이 Hadoop인 게시물을 표시합니다. 모든 게시물 표시

2025년 6월 21일 토요일

Hive 서버와 Hive Python 클라이언트 사용하기

1 Abstraction
2 Hive
3 Apache Thrift
4 Hive Server
5 Hive Python 클라이언트
6 References

원문: http://mixellaneous.tistory.com/826

[edit]

1 Abstraction #

Hive Thrift 서버와 Python 클라이언트를 사용하는 방법에 대하여 알아본다.

[edit]

2 Hive #

Hive is a data warehouse infrastructure built on top of Hadoop that provides tools to enable easy data summarization, adhoc querying and analysis of large datasets data stored in Hadoop files.
http://hadoop.apache.org/hive/

[edit]

3 Apache Thrift #

Thrift is a software framework for scalable cross-language services development. It combines a software stack with a code generation engine to build services that work efficiently and seamlessly between C++, Java, Python, PHP, Ruby, Erlang, Perl, Haskell, C#, Cocoa, Smalltalk, and OCaml.
http://incubator.apache.org/thrift/

[edit]

4 Hive Server #

Hive 서버는 Thrift 서버로 동작한다.

서버 시작


$ hive --service hiveserver 
[1] 9818

$ Starting Hive Thrift Server

09/12/17 16:59:39 INFO service.HiveServer: Starting hive server on port 10000



.

.



$

[edit]

5 Hive Python 클라이언트 #

Hadoop & Hive 설치 및 확인 (Hadoop 0.20.1 & Hive 0.4.0)

$ rpm -qa | grep hadoop-0.20

hadoop-0.20-jobtracker-0.20.1+133-1

hadoop-0.20-libhdfs-0.20.1+133-1

hadoop-0.20-tasktracker-0.20.1+133-1

hadoop-0.20-0.20.1+133-1

hadoop-0.20-datanode-0.20.1+133-1

hadoop-0.20-secondarynamenode-0.20.1+133-1

hadoop-0.20-conf-pseudo-0.20.1+133-1

hadoop-0.20-pipes-0.20.1+133-1

hadoop-0.20-namenode-0.20.1+133-1

hadoop-0.20-native-0.20.1+133-1

hadoop-0.20-docs-0.20.1+133-1

$

$ rpm -qa | grep hive

hadoop-hive-webinterface-0.4.0+14-1

hadoop-hive-0.4.0+14-1

샘플 데이터

$ cat /tmp/r.txt

a       1       1.0

b       2       2.0

c       3       3.0

$

PYTHONPATH 설정 (Hive Python 라이브러리)

$ export PYTHONPATH="/usr/lib/hive/lib/py"

$ env | grep PYTHONPATH

PYTHONPATH=/usr/lib/hive/lib/py

코드

import sys



from hive_service import ThriftHive

from hive_service.ttypes import HiveServerException

from thrift import Thrift

from thrift.transport import TSocket

from thrift.transport import TTransport

from thrift.protocol import TBinaryProtocol



try:

    transport = TSocket.TSocket('localhost', 10000)

    transport = TTransport.TBufferedTransport(transport)

    protocol = TBinaryProtocol.TBinaryProtocol(transport)



    client = ThriftHive.Client(protocol)

    transport.open()



    client.execute("CREATE TABLE r(a STRING, b INT, c DOUBLE) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t' STORED AS TEXTFILE")

    client.execute("LOAD DATA LOCAL INPATH '/tmp/r.txt' OVERWRITE INTO TABLE r")

    client.execute("SELECT * FROM r")

    for row in client.fetchAll():

      print row



    transport.close()



except Thrift.TException, tx:

    print '%s' % (tx.message)

실행
{{{
$ python hive_py.py

a       1       1.0

b       2       2.0

c       3       3.0

[edit]

6 References #

Hive Wiki
Apache Thrift

카산드라 우분투에 설치하기


wget http://mirror.apache-kr.org/cassandra/1.2.8/apache-cassandra-1.2.8-bin.tar.gz
tar xzf apache-cassandra-1.2.8-bin.tar.gz

su
mv apache-cassandra-1.2.8 /usr/local/cassandra
chown -R huser:hadoop /usr/local/cassandra
mkdir /var/lib/cassandra
mkdir /var/log/cassandra
chown -R huser:hadoop /var/lib/cassandra
chown -R huser:hadoop /var/log/cassandra
exit

vi /usr/local/cassandra/conf/cassandra-env.sh
JVM_OPTS="$JVM_OPTS -Xss180k -> JVM_OPTS="$JVM_OPTS -Xss256k 으로 변경

cassandra -f
pkill -f 'java.*cassandra'

vi .profile
export CASSANDRA_PATH=/usr/local/cassandra/bin

vi /usr/local/cassandra/conf/cassandra.yaml
rpc_address, listen_address 를 0.0.0.0으로 세팅

Hadoop에 관한 10가지 미신 깨기

Hadoop에 관한 10가지 미신 깨기.
전문: http://tdwi.org/Articles/2012/03/20/Busting-10-Hadoop-Myths.aspx

#01. Hadoop은 다수의 제품으로 구성된다.
#02. Hadoop은 오픈 소스지만, 벤더들도 공급한다.
#03. Hadoop은 단일 제품이 아니라, 일종의 생태계다.
#04. HDFS는 파일 시스템이지, 데이터 베이스 관리 시스템이 아니다.
#05. Hive는 SQL을 닮았지만, 표준 SQL은 아니다.
#06. Hadoop과 MapReduce는 관계가 있지만, 서로간에 반드시 필요한 것은 아니다.
#07. MapReduce는 분석에 대한 제어를 할 수 있지, 분석 자체가 아니다.
#08. Hadoop은 데이터 다양성에 대한 것이지, 데이터 크기 자체는 아니다.
#09. Hadoop은 DW를 보완하지, 대체하진 않는다.
#10. Hadoop은 웹 분석 뿐 아니라 다양한 분석을 가능케 한다.

- 번역: 윤주형

WebHDFS

[edit]

업로드 #

Windows의 파일을 HDFS로 업로드

curl -i -X PUT -L -T "c:\temp\list.txt" "http://webhdfs:50070/webhdfs/v1/user/hive/warehouse/gamelog/list.txt?op=CREATE&user.name=hdfs&overwrite=true"

[edit]

상황1 #

cluster1에는 1시간마다 파일명을 바꾸어 가면서 데이터가 쌓인다.
webhdfs를 이용하여 10분마다 추가된 데이터를 가져오고 싶다.

hdfs에 /tmp/test.txt의 t1 시점의 데이터

{"id":1,"nm":"aaa"}
{"id":2,"nm":"aaaa"}
{"id":3,"nm":"aaaaa"}
{"id":4,"nm":"aaaa"}
{"id":5,"nm":"aaaaa"}
{"id":6,"nm":"aaa"}
{"id":7,"nm":"aaa"}
{"id":8,"nm":"aa"}
{"id":9,"nm":"a"}
{"id":10,"nm":"aa"}

t1 시점에는 203 byte였다.

hdfs에 /tmp/test.txt의 t2 시점의 데이터

{"id":1,"nm":"aaa"}
{"id":2,"nm":"aaaa"}
{"id":3,"nm":"aaaaa"}
{"id":4,"nm":"aaaa"}
{"id":5,"nm":"aaaaa"}
{"id":6,"nm":"aaa"}
{"id":7,"nm":"aaa"}
{"id":8,"nm":"aa"}
{"id":9,"nm":"a"}
{"id":10,"nm":"aa"}
{"id":11,"nm":"aa"} ---> 여기부터 추가되었다. 
{"id":12,"nm":"aa"}
{"id":13,"nm":"aa"}

solution
offset을 사용하면 된다.

t1 시점

wget "http://192.168.0.10:50070/webhdfs/v1/tmp/test.txt?op=OPEN&offset=0" -O test1.txt

t2 시점

wget "http://192.168.0.10:50070/webhdfs/v1/tmp/test.txt?op=OPEN&offset=204" -O test2.txt

hadoop metadata가 정확해야 한다.

[Hive] Hive 조인방식

셔플 조인
- 맵/리듀스를 이용하여 키를 기준으로 셔플(재분배)하여 조인
- 조인 방식 중 자원을을 가장 많이 사용(=느림)
브로드캐스트 조인
- 작은 테이블을 메모리에 올리고, 큰 테이블을 읽어가면서(맵퍼가 읽음) 조인(RDBMS에서 해쉬 조인시 작은 테이블을 메모리에 올리는 것과 비슷)
- 빠르지만, 메모리에 올리려는 테이블이 메모리에 올라갈만큼 작아야 함
정렬-병합-버킷(SMB) 조인
- 키가 인접한 특성을 이용한 조인
- 빠르지만, 사전에 자료가 정렬되어 있고, 버킷화되어 있어야 함

[Hive] Hive Python

파이썬 프로그램을 짠다.

import sys
import string
from pykospacing import spacing

for line in sys.stdin:
    line = line.strip()
    pc_id, msg = line.split("\t")
    print ("\t".join([pc_id, spacing(msg)]))

hdfs에 넣는다.

hadoop fs -put -f /home/hdfs/py/chat_spacing.py /user/hive/udfs

transform 함수를 써서 사용한다.

set hive.execution.engine=mr;
add file hdfs:///user/hive/udfs/chat_spacing.py;

select transform(msg) using 'python3.4 chat_spacing.py' as (pc_id bigint, msg string)
from (
    select concat(cast(id as string), "\t", sss) msg
    from sample
) t;

[edit]

pyhive #

from pyhive import hive
conn = hive.Connection(host="192.168.0.100", port=10000, username="hdfs")
 
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
for table in cursor.fetchall():
    print(table)

export http_proxy='http://xxx.com'
export https_proxy='https://xxx.com'

pip3.4 install pip3.4 install tensorflow --trusted-host pypi.python.org --index-url=https://pypi.python.org/simple/

[Hive] Hive 테이블 만들기

1 테이블 만들기:기본

[edit]

1 테이블 만들기:기본 #

[Hive] Hive 로드밸런싱 - nginx

1 로드밸런싱
2 nginx
3 nginx.conf 편집
4 방화벽
5 서비스 등록
6 참고

windows 서버에 nginx 설치

[edit]

1 로드밸런싱 #

[edit]

2 nginx #

http://nginx.org/en/download.html 에 들어가서, Stable version을 다운로드 하여, 압축풀고, c:에 위치시킨다.

[edit]

3 nginx.conf 편집 #

events {
	worker_connections 1024;
}

stream {
	upstream 10.0.0.100 {
		server 10.0.0.200:10000;
		server 10.0.0.201:10000;
		server 10.0.0.202:10000;
	}

	server {
		listen 10001;
		proxy_server 10.0.0.100
	}
}

로드밸런싱 방식은 기본이 라운드로빈 방식이다. 바꾸려면 맨 아래 참고 url

[edit]

4 방화벽 #

윈도우즈 방화벽에서 인바운드 규칙에 TCP 10001 추가

[edit]

5 서비스 등록 #

NSSM 다운로드 (windows service로 만들어줄 프로그램) 하여 압축 풀고
https://nssm.cc/download

cmd에서 압축푼 경로로 이동하여 다음을 입력

nssm install nginx

이렇게 하면 ui가 뜨는데, path에서 nginx.exe 을 선택해주고, install service 버튼을 눌러주면 끝난다.
서비스로 이동해서 시작시켜주면 된다.

[edit]

6 참고 #

https://kscory.com/dev/nginx/loadbalancer

[Hive] Hive 트러블슈팅

1 Name node is in safe mode
2 상면 공간 이동 후 느려진 현상

[edit]

1 Name node is in safe mode #

hadoop 이 비정상 종료를 해서 그렇다. 정상모드로 바꾸어야 한다.
hadoop dfsadmin -safemode leave

[edit]

2 상면 공간 이동 후 느려진 현상 #

서버 상면 공간 이동 후 쿼리가 완전 느려졌다.
10시간 정도 찾은 결과 광케이블 문제였다. 스위치에서 핑을 때리니 1%가 빠졌다.
문제 서버를 제외하니 모든게 정상으로 돌아왔다.

[Hive] Hive개요

Hive는..

~~인덱스가 없다. -> 다차원 파티션과 버켓(or 정렬된 버켓)으로 해결한다.~~
인덱스가 있긴 있다. -> 하지만, 제한된 기능으로 역시 다차원 파티션과 버켓(or 정렬된 버켓)으로 해결한다.
갱신, 삭제도 없다. -> 해당 파티션에 데이터를 overwrite함으로써 해결한다.
어떤 버전부터 지원되는지 모르겠지만, 0.8.0 버전은 create schema, create database 를 지원한다. 물론 use <database_name> 도 된다.
응답시간이 느리다. 적절한 곳에 적절히 써먹어야 한다.
HiveQL이라고 ANSI-92와 닮은 쿼리언어를 지원한다. 생각보다 유연한듯..
인프라 운영자(hdfs)가 꼭 있어야 할 듯... 전기세 기타등등을 따지면 그리 싼비용도 아닌거 같음..

[Hive] Hive Tuning

1 SQL Server에서 ODBC연결하여 Hive에서 데이터 가져올때
2 파티션 컬럼 읽을 때
3 자원 쓰기

[edit]

1 SQL Server에서 ODBC연결하여 Hive에서 데이터 가져올때 #

sql1방식보다 sql2 방식으로 가져오는게 4~10배 빠르다. (경험적으로..)

sql1

drop table if exists #temp1
select *
into #temp1
from openquery(hive, '
select
    col1, col2, col3
from gamelog
where 1=1
and a.date_key between 20190619 and 20190821
'
)

sql2

exec('drop table if exists etl_fcc1d2fd1da44550bbde0661f25b631a;') at hive
 
exec('
create table etl_fcc1d2fd1da44550bbde0661f25b631a
as
select
    col1, col2, col3
from gamelog
where 1=1
and a.date_key between 20190619 and 20190821
') at hive
go
 
drop table if exists #1
select *
into #1
from openquery(hive, '
select
    col1, col2, col3
from etl_fcc1d2fd1da44550bbde0661f25b631a')
--4:30

[edit]

2 파티션 컬럼 읽을 때 #

현상:

gamelog 테이블은 date_key, log_no로 파티션된 테이블이다.
아래와 같이 파티션 키만 읽는 쿼리를 날렸다.
그런데 1시간이 걸리는 것이다.

hive> set hive.execution.engine=mr;
hive> select log_no
    > from gamelog
    > where 1=1
    > and date_key=20181228
    > group by log_no;

해결:
영문을 모르겠으나 메시지를 잘 살펴보니 알겠더라.
이유는 hive가 input data size를 졸라 작제 추정한 것에 있다.

그냥 쓸때없이 파티션 키가 아닌 다른 데이터도 조회하도록 해준다.

hive> select log_no, count(get_json_object(log_string, '$.1') cnt
    > from gamelog
    > where 1=1
    > and date_key=20181228
    > group by log_no;

[edit]

3 자원 쓰기 #

투기적 실행

set hive.mapred.reduce.tasks.speculative.execution =true;

map수 늘리기(숫자를 작게 할 수록 map숫자가 늘어남)

set mapreduce.input.fileinputformat.split.maxsize=1048576;

[Hive] Hive UDF JSON Split

환경은 Hortonworks Sandbox다. HDP 2.6(참고: VirtualBox에서 Hortonworks Sandbox 시작하기)

아래의 json string을 hive로 주물러봤다.

[{"product_id":220003038067,"timestamps":"1340321132000"},{"product_id":300003861266,"timestamps":"1340271857000"}]

하아~. 정말 거지같다.
get_json_object, json_tuple이라는 함수가 있지만, 위와 같은 형태의 json은 다루기 힘들었다.
구글에서 찾아보니 UDF로 해결하고 있었으며, 다음의 UDF를 찾았다.

https://github.com/pythian/hive-json-split

jar 파일이 없다. 메이븐으로 빌드를 해야 하는데, 난 메이븐을 다뤄 본적이 한 번도 없다.
시도.

maven 설치

http://maven.apache.org/download.cgi 에서 다운로드
d:\apache-maven-3.5.0에 풀었다.
MAVEN_HOME, JAVA_HOME 환경변수 설정

MAVEN_HOME = d:\apache-maven-3.5.0
JAVA_HOME = C:\Program Files\Java\jdk1.8.0_111

udf 빌드

https://github.com/pythian/hive-json-split 에서 다운로드

d:\hive-json-split-master에 풀었다.

d:
cd hive-json-split-master
D:\apache-maven-3.5.0\bin\mvn package

-- 첫 번째 빌드에 실패했었다. 혹시 JAVA_HOME 설정을 해야 하나 싶어 환경변수 설정을 해줬더니 성공했다.
--

JsonSplit-1.3-SNAPSHOT.jar

빌드가 성공하고나니 target이라는 디렉토리가 생기고, 아래와 같이 jar 파일 2개가 생기더라.

WinSCP를 이용해 hive가 돌어가는 서버에 JAR파일 2개를 /usr/hdp/current/hive-client/lib/ 에 복사 후 hive CLI에서 다음의 명령 수행

add jar /usr/hdp/current/hive-client/lib/JsonSplit-1.3-SNAPSHOT.jar;
create function json_split as 'com.pythian.hive.udf.JsonSplitUDF';
create function json_map as 'com.pythian.hive.udf.JsonMapUDF';

json test

create table json_test(json_string string);
insert into json_test values('[{"product_id":220003038067,"timestamps":"1340321132000"},{"product_id":300003861266,"timestamps":"1340271857000"}]');

set hive.cli.print.header=true;
select 
    b.col.row_id
,   b.col.json_string
,   c.*
from json_test a
    lateral view explode(json_split(json_string)) b
    lateral view json_tuple(b.col.json_string, 'product_id', 'timestamps') c as product_id, timestamps;

[root@sandbox ~]# hive
log4j:WARN No such property [maxFileSize] in org.apache.log4j.DailyRollingFileAppender.

Logging initialized using configuration in file:/etc/hive/2.6.1.0-129/0/hive-log4j.properties
hive> set hive.cli.print.header=true;
hive> select
    >     b.col.row_id
    > ,   b.col.json_string
    > ,   c.*
    > from json_test a
    >     lateral view outer explode(json_split(json_string)) b
    >     lateral view json_tuple(b.col.json_string, 'product_id', 'timestamps') c as product_id, timestamps;
OK
row_id  json_string     c.product_id    c.timestamps
0       {"product_id":220003038067,"timestamps":"1340321132000"}        220003038067    1340321132000
1       {"product_id":300003861266,"timestamps":"1340271857000"}        300003861266    1340271857000
Time taken: 2.067 seconds, Fetched: 2 row(s)
hive>

이 UDF보다 더 좋은 솔루션을 찾아 보았으나 UDF만한 것이 없다는 결론을 얻는데 약 5시간 걸렸다.

세션 열때마다 add jar /usr/hdp/current/hive-client/lib/JsonSplit-1.3-SNAPSHOT.jar;를 해야하니까 permanent function을 만드는 것이 좋겠다.

hdfs에 jar 파일을 올리고

hadoop fs -put JsonSplit-1.3-SNAPSHOT.jar ;

hive 쉘에서..

create function json_split as 'com.pythian.hive.udf.JsonSplitUDF' using JAR 'hdfs://localhost:8020/user/cloudera/JsonSplit-1.3-SNAPSHOT.jar';
create function json_map as 'com.pythian.hive.udf.JsonMapUDF' using JAR 'hdfs://localhost:8020/user/cloudera/JsonSplit-1.3-SNAPSHOT.jar';

os 쉘에서..

sudo service hive-server2 restart

json_map 함수는 아래와 같이 써먹는다.

--이건 sql server의 openjson과 유사
select explode(json_map(json_string))
from json_test

select json_map(b.col.json_string)['product_id']
from json_test a
    lateral view outer explode(json_split(json_string)) b

[edit]

다른 방법 #

이게 빠르다. 하지만 귀찮다.

select b.*
from json_test a
    lateral view outer posexplode(split(regexp_replace(substr(json_string, 2, length(json_string)-2), '},\\{', '},,,,{'), ',,,,')) b

select b.*
from json_test a
    lateral view explode(split(regexp_replace(regexp_extract(a.json_string,'^\\[(.+)\\]$',1),'\\}\\,\\{', '\\}\\|\\|\\{'),'\\|\\|')) b

https://github.com/klout/brickhouse -- 이재학 2018-03-04 13:31:33

https://github.com/aaronshan/hive-third-functions -- 이재학 2018-03-04 15:14:53

https://github.com/klout/brickhouse -- 이재학 2018-06-08 13:11:58

Post List

2025년 6월 21일 토요일

Contents

1 Abstraction #

2 Hive #

3 Apache Thrift #

4 Hive Server #

5 Hive Python 클라이언트 #

6 References #

업로드 #

상황1 #

pyhive #

Contents

1 테이블 만들기:기본 #

Contents

1 로드밸런싱 #

2 nginx #

3 nginx.conf 편집 #

4 방화벽 #

5 서비스 등록 #

6 참고 #

Contents

1 Name node is in safe mode #

2 상면 공간 이동 후 느려진 현상 #

Contents

1 SQL Server에서 ODBC연결하여 Hive에서 데이터 가져올때 #

2 파티션 컬럼 읽을 때 #

3 자원 쓰기 #

다른 방법 #