태그 Archives: bigdata

python – apache pyarrow 를 이용한 parquet 생성 및 테스트

apache 재단에서 진행 되는 프로젝트 이다. python, java, R 등등 많은 언어를 지원 한다.

CSV (Comma-Separated Values)의 가로열 방식의 데이터 기록이 아닌 세로열 기록 방식으로 기존 가로열 방식에서 불가능한 영역을 처리가 가능하도록 한다.

보이는가 선조의 지혜가 -3-)b

이미지 출처: 훈민정음 나무위키

차이점을 그림으로 표현하자면 아래와 같다.

문서를 모두 읽는다 에서는 큰 차이가 발생하지 않지만 구조적으로 모든 행이 색인(index) 처리가 된 것처럼 파일을 읽을 수 있다.

sql 문으로 가정으로 “(SELECT * FROM 테이블 WHERE 재질 = ‘철’)” 을 찾게 될 경우 index 가 둘다 없다는 가정하에서

CSV 는 9개의 칸을 읽어야 하지만 (재질->무게->산화->나무->가벼워->탄다->철->무거워->안탄다->return)

parquet 의 경우 5개의 칸만 읽으면 된다. (재질->나무->철->무거워->안탄다->return)

PS. 물론 색인(index) 는 이런 구조가 아닌 hash 처리에 따른 협차법 으로 찾아서 빨리 찾을 수 있어 차이가 있다.

압축을 하더라도 컬럼별 압축이 되기 때문에 필요한 내용만 읽어서 압축해제 하여 데이터를 리턴 한다.

적당한 TSV (Tab-Separated Values)데이터를 준비 한다.

python 을 이용하여 TSV 파일을 읽고 python 의 pyarrow를 이용하여 parquet 파일을 생성 하고 읽는 테스트를 한다. (pyarrow, pandas 는 pip install pyarrow pandas 으로 설치할 수 있다.)

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import os

import time

import pandas as pd

import pyarrow as pa

import pyarrow.parquet as pq

from pyarrow import csv

def tsv2parquet(filename, skiphead, column_length, toformat):

if toformat in ('none', 'snappy', 'gzip', 'lzo', 'brotil', 'lz4', 'zstd'):

if skiphead == 0:

skiphead = None

table_columns = [str(i) for i in range(0, column_length)]

r_opt = csv.ReadOptions(skip_rows=skiphead, column_names=table_columns, use_threads=False)

p_opt = csv.ParseOptions(delimiter='\t')

pyarrow_table = csv.read_csv(fname, read_options=r_opt, parse_options=p_opt)

outname = os.path.splitext(fname)[0]+'.'+toformat+'.parquet'

pq.write_table(pyarrow_table, outname, compression=toformat)

else:

print('didn\'t support format: '+ toformat)

exit(1)

return outname

print('pyarrow version:', pa.__version__) # print pyarrow Version

fname = "sample/shjang_Genome_20191011.txt" # Target file (TSV)

sh = 4 # file header line.

cc = 10 # column count

out_format = 'gzip' # pyarrow 0.16 support: 'none', 'snappy', 'gzip', 'lz4', 'zstd'

print('File size: ' + str(os.path.getsize(fname)))

ts = time.time()

outfile = tsv2parquet(fname, sh, cc, out_format) # make parquet file.

print('make parquet(' + out_format + ') file: ' + str(round(time.time() - ts, 2)) + ' sec')

ts = time.time()

dataframe = pd.read_parquet(outfile, engine='pyarrow')

print('parquet -> pandas -> dataframe: ' + str(round(time.time() - ts, 2)) + ' sec')

ts = time.time()

dataframe = pq.read_table(outfile).to_pandas()

print('parquet -> pyarrow -> dataframe: ' + str(round(time.time() - ts, 2)) + ' sec')

exit(0)

TSV -> parquet 압축률(높을수록 좋음) 및 처리 시간(낮을수록 좋음)

	def	ext	MB	compress ratio	processing time python 2.7	processing time python 3.6
txt		.txt	58.8 MB
gzip		.txt.gz	16.3 MB	72%	3.24 sec
pyarrow	write_table, compression='none'	.parquet	40.1 MB	32%	0.74 sec	0.93 sec
	write_table, compression='snappy'		24.8 MB	58%	1.31 sec	0.95 sec
	write_table, compression='lz4'		24.7 MB	58%	0.79 sec	0.94 sec
	write_table, compression='zstd'		19.3 MB	67%	1.00 sec	0.98 sec
	write_table, compression='gzip'		18.8 MB	68%	5.07 sec	1.18 sec

읽기/쓰기 테스트 모두 AWS – EC2(m5.large-centos7) – gp2(100GB) 에서 진행 하였다.

parquet 을 생성한 이유는 파일을 읽을때 모든 컬럼인 index가 걸려있는것과 같이 빠르게 읽기 위함이니 읽기 테스트도 해본다.

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import os

import time

import pandas as pd

import pyarrow as pa

import pyarrow.parquet as pq

from pyarrow import csv

def tsv2table2dataframe(filename, skiphead, column_length):

table_columns = [str(i) for i in range(0, column_length)]

r_opt = csv.ReadOptions(skip_rows=skiphead, column_names=table_columns, use_threads=False)

p_opt = csv.ParseOptions(delimiter='\t')

pyarrow_table = csv.read_csv(fname, read_options=r_opt, parse_options=p_opt)

t1 = str(round(time.time() - ts, 2))

ts2 = time.time()

pyarrow_df = pyarrow_table.to_pandas()

t2 = str(round(time.time() - ts2, 2))

return pyarrow_df, t1, t2

print('pyarrow version:', pa.__version__) # print pyarrow Version

fname = "sample/shjang_Genome_20191011.txt" # Target file (TSV)

sh = 4 # file header line.

cc = 10 # column count

out_format = 'gzip' # pyarrow 0.16 support: 'none', 'snappy', 'gzip', 'lz4', 'zstd'

print('File size: ' + str(os.path.getsize(fname)))

ts = time.time()

dataframe = pd.read_csv(fname, skiprows=sh, sep='\t', quotechar='"', header=None, index_col=None, error_bad_lines=False)

print('text TSV file read with pandas to dataframe: ' + str(round(time.time() - ts, 2)) + ' sec')

ts = time.time()

dataframe = pd.read_csv(fname+'.gz', compression='gzip', skiprows=sh, sep='\t', quotechar='"', header=None, index_col=None, error_bad_lines=False)

print('gzip TSV file read with pandas to dataframe: ' + str(round(time.time() - ts, 2)) + ' sec')

ts = time.time()

dataframe, t1, t2 = tsv2table2dataframe(fname, sh, cc)

print('text TSV read(' + t1 + ' sec) with pyarrow to dataframe(' + t2 + ' sec): ' + str(round(time.time() - ts, 2)) + ' sec')

ts = time.time()

dataframe, t1, t2 = tsv2table2dataframe(fname+'.gz', sh, cc)

print('gzip TSV read(' + t1 + ' sec) with pyarrow to dataframe(' + t2 + ' sec): ' + str(round(time.time() - ts, 2)) + ' sec')

exit(0)

TSV, parquet 파일 읽기 테스트 (pandas, pyarrow)

	def	ext	MB	processing time python 2.7	processing time python 3.6
pandas	read_csv	.txt	58.8 MB	1.39 sec	1.56 sec
	read_csv, compression='gzip'	.txt.gz	16.3 MB	1.68 sec	2.06 sec
	read_parquet	.parquet (none)	40.1 MB	0.72 sec	0.93 sec
		.parquet (snappy)	24.8 MB	1.03 sec	0.95 sec
		.parquet (lz4)	24.7 MB	0.73 sec	0.94 sec
		.parquet (zstd)	19.3 MB	0.76 sec	0.95 sec
		.parquet (gzip)	18.8 MB	0.96 sec	1.18 sec
pyarrow	read_csv, to_pandas	.txt	58.8 MB	1.01 sec	1.30 sec
	read_csv, to_pandas	.txt.gz	16.3 MB	1.41 sec	1.37 sec
	read_table, to_pandas	.parquet (none)	40.1 MB	0.69 sec	0.90 sec
		.parquet (snappy)	24.8 MB	0.99 sec	0.89 sec
		.parquet (lz4)	24.7 MB	0.69 sec	0.92 sec
		.parquet (zstd)	19.3 MB	0.75 sec	0.95 sec
		.parquet (gzip)	18.8 MB	0.95 sec	1.22sec

이 문서 처음에 언급 했다 시피 대용량 파일을 처리 하기 위함. 즉 “빅데이터”(HIVE, Presto, Spark, AWS-athena)환경을 위한 포멧이다.

모두 테스트 해보면 좋겠지만 아직 실력이 부족해서 AWS athena 만 테스트를 진행 한다.

구조적으로 S3 버킷에 parquet 파일을 넣어 두고 athena 에서 테이블을(S3 디렉토리 연결) 생성 하여 SQL 문으로 검색을 하는데 사용 한다.

TSV, parquet 파일 읽기 테스트 (AWS – athena)

	ROW FORMAT SERDE	ext	Searched MB	processing time (select target 2)	processing time (select target 50)
athena	org.apache.hadoop.hive. serde2.lazy. LazySimpleSerDe	.txt	58.8 MB	1.17 ~ 3.35 sec	1.86 ~ 2.68 sec
	org.apache.hadoop.hive. serde2.lazy. LazySimpleSerDe	.txt.gz	16.3 MB	1.37 ~ 1.49 sec	1.44 ~ 2.69 sec
	org.apache.hadoop.hive. ql.io.parquet.serde. ParquetHiveSerDe	.txt.parquet	10.48 MB	1.11 ~ 1.49 sec	1.00 ~ 1.38 sec
		.snappy.parquet	4.71 MB	0.90 ~ 2.36 sec	0.90 ~ 1.00 sec
	지원 불가	.lz4.parquet	지원 불가
	지원 불가	.zstd.parquet	지원 불가
	org.apache.hadoop.hive. ql.io.parquet.serde. ParquetHiveSerDe	.gzip.parquet	2.76 MB	0.89 ~ 1.17 sec	0.90 ~ 1.85 sec

읽는 속도가 향상되었고 스캔 크기가 적게 나온다. (parquet 의 강점을 보여주는 테스트-스캔비용의 절감이 가능.)

athena 테이블 생성에 사용된 DDL 쿼리문 (TSV, parquet)

CREATE EXTERNAL TABLE IF NOT EXISTS [데이터베이스명].[테이블명] (

`rsid` string,

`chr` string,

`pos` int,

`gt` string

)

ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe'

STORED AS INPUTFORMAT 'org.apache.hadoop.mapred.TextInputFormat'

OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'

LOCATION 's3://[S3-URL]/[TSV폴더]';

CREATE EXTERNAL TABLE IF NOT EXISTS [데이터베이스명].[테이블명] (

`rsid` string,

`chr` string,

`pos` int,

`gt` string

)

ROW FORMAT SERDE 'org.apache.hadoop.hive.ql.io.parquet.serde.ParquetHiveSerDe'

WITH SERDEPROPERTIES ('serialization.format' = '1', 'parquet.column.index.access'='true')

LOCATION 's3://[S3-URL]/[parquet폴더]'

TBLPROPERTIES ('has_encrypted_data'='true');

PS. 이건 저도 어려 웠어요…..

graylog 를 통한 syslog 통합 감사(audit)

graylog 는 syslog 처리를 위한 빅데이터 분석 솔루션이다. 기본적으로야 syslog 를 한곳에 적재하여 로그 감사를 하는 목적으로 사용한다.

물론 syslog 뿐만 아니라 웹로그등을 수집하여 분석하는것 역시 가능하다.

Mongodb + elasticsearch + java 를 사용하여 운영 되며 노드를 추가하여 다중 노드에서의 분석 처리가 가능하다.

위와 같이 대쉬보드를 만들어 한눈에 볼수 있으며 alert 설정을 통해 중요 특정한 이벤트 발생시 손쉬운 확인이 가능 하다.

graylog 의 권장 사항은 4GB의 램을 필요로 하나 구축테스트를 해본 결과 1GB 에 2GB 수준의 swap 으로 정상적인 운영이 가능했다.

오픈소스이기 때문에 유료 프로그램인 kiwi syslog의 라이선스 비용 대비 40% 수준이면 클라우드에 구축하여 운영할 수 있다.
(t2.micro / 1 core – 1GB RAM) CentOS7 에서 graylog 를 설치 하는 방법을 설명한다.

1. 가상 서버를 준비하고 기본적인 업데이트 및 hostname 지정 등을 한 뒤에 리부팅 한다.

~]# yum update yum

~]# yum update

~]# hostname Graylog-01

~]# echo "Graylog-01" > /etc/hostname

~]# yum install epel-release

~]# yum install wget pwgen java perl-Digest-SHA

~]# sync

~]# reboot

2. 시스템 메모리 및 네트워크 튜닝값을 적용 한다.

~]# echo "fs.aio-max-nr = 1048576

fs.file-max = 6815744

kernel.sem = 250 32000 100 128

kernel.shmall = 2097152

kernel.shmmax = $(free -b|awk '/^Mem:/{print int($2*0.9)}')

net.core.rmem_default = 262144

net.core.rmem_max = 16777216

net.core.wmem_default = 262144

net.core.wmem_max = 16777216

net.ipv4.tcp_rmem = 262144 262144 16777216

net.ipv4.tcp_wmem = 262144 262144 16777216

net.ipv4.ip_local_port_range = 9000 65500" >> /etc/sysctl.conf

~]# sysctl -p

3. swap 을 생성한다.(aws-centos7 의경우 swap 설정되어 있지 않다.)

프로그램 초기 설치시 selinux 충돌을 방지 하기 위해 임의 setenforce 0 명령어로 selinux 를 permissive 모드로 변경 후 진행 한다.

~]# setenforce 0

~]# cd /

~]# dd if=/dev/zero of=/swapfile bs=1M count=2048

~]# chown root:root /swapfile

~]# chmod 600 /swapfile

~]# mkswap /swapfile

~]# swapon /swapfile

재부팅 후에 swap 이 자동 적용 될수 있게 fstab 을 수정한다. ~]# vi /etc/fstab

1	/swapfile swap swap defaults,noatime 0 0

4. elasticsearch 설치

~]# echo '[elasticsearch-2.x]

name=Elasticsearch repository for 2.x packages

baseurl=http://packages.elastic.co/elasticsearch/2.x/centos

gpgcheck=1

gpgkey=http://packages.elastic.co/GPG-KEY-elasticsearch

enabled=1

' | sudo tee /etc/yum.repos.d/elasticsearch.repo

~]# yum install elasticsearch

~]# systemctl enable elasticsearch.service

~]# systemctl start elasticsearch.service

5. mongodb 설치

~]# echo '[mongodb-org-3.4]

name=MongoDB Repository

baseurl=https://repo.mongodb.org/yum/redhat/$releasever/mongodb-org/3.4/x86_64/

gpgcheck=1

enabled=1

gpgkey=https://www.mongodb.org/static/pgp/server-3.4.asc

' | sudo tee /etc/yum.repos.d/mongodb-org.repo

~]# yum install mongodb-org

6. mongodb 설정

mongdb 의 data 파일을 /free/mongo_data 에 적재할 예정 이기 때문에 conf 파일을 수정 하고 selinux 설정을 추가 한다.

~]# sed -i 's=#^ dbPath: /var/lib/mongo$= dbPath: /free/mongo_data=g' /etc/mongod.conf

~]# systemctl enable mongod.service

~]# systemctl start mongod.service

7. graylog 설치

~]# rpm -Uvh https://packages.graylog2.org/repo/packages/graylog-2.4-repository_latest.rpm

~]# sudo yum install graylog-server

8. graylog 설정

conf 파일을 수정하기 전에 pwgen 을 이용하여 secret 값을 생성 하고 shasum 을 이용하여 관리자 패스워드 값을 생성한다.

]# pwgen -N 1 -s 72

LVTBitCSsOFx0aJkvAPNHBjTJVLTB0Z5l8vy4X5AltgMhIYCyV7Pwo2sKAX3iYEPyTyeK7l9

~]# echo 'P@$$W0RD' | shasum -a 256 | cut -d" " -f1

eec206d73eb43b63a9c381aadcc43864a0fe881c79e4d716f08bda25ef30d7f4

~]# vi /etc/graylog/server/server.conf ## 각 값을 찾아서 수정한다.

password_secret = LVTBitCSsOFx0aJkvAPNHBjTJVLTB0Z5l8vy4X5AltgMhIYCyV7Pwo2sKAX3iYEPyTyeK7l9

root_password_sha2 = eec206d73eb43b63a9c381aadcc43864a0fe881c79e4d716f08bda25ef30d7f4

root_email = 관리자@이메일.com

root_timezone = Asia/Seoul

web_listen_uri = http://127.0.0.1:9000/

9. graylog 시작 및 systemd 등록

1 2	~]# systemctl enable graylog-server.service ~]# systemctl start graylog-server.service

10. nginx 설치

여기서는 손쉬운 http/2 구축과 brotli 이미지 압축 등을 위해 codeit – nginx 을 설치 한다.

~]# cd /etc/yum.repos.d

~]# wget https://repo.codeit.guru/codeit.el`rpm -q --qf "%{VERSION}" $(rpm -q --whatprovides redhat-release)`.repo

~]# wget https://repo.codeit.guru/codeit.mainline.el`rpm -q --qf "%{VERSION}" $(rpm -q --whatprovides redhat-release)`.repo

~]# yum -y install nginx

11. nginx 설정

DH 파라메터파일을 생성 한다.

1	~]# openssl dhparam -dsaparam -out /etc/nginx/dhparam.pem 4096

/etc/nginx/nginx.conf 파일을 수정한다.

user nginx;

worker_processes auto;

error_log /var/log/nginx/error.log warn;

pid /var/run/nginx.pid;

load_module modules/ngx_http_geoip_module.so;

load_module modules/ngx_stream_geoip_module.so;

events {

worker_connections 1024;

}

http {

include /etc/nginx/mime.types;

default_type application/octet-stream;

geoip_country /usr/share/GeoIP/GeoIP.dat;

log_format main '$remote_addr - $remote_user [$time_local] "$request" '

'$status $body_bytes_sent "$http_referer" '

'"$http_user_agent" "$http_x_forwarded_for"';

access_log /var/log/nginx/access.log main;

server_tokens off;

sendfile on;

keepalive_timeout 65;

gzip on;

gzip_types text/html text/css text/x-component application/x-javascript application/javascript text/javascript text/x-js text/richtext image/svg+xml text/plain text/xsd text/xsl text/xml image/x-icon application/octet-stream;

brotli on;

brotli_types text/html text/css text/x-component application/x-javascript application/javascript text/javascript text/x-js text/richtext image/svg+xml text/plain text/xsd text/xsl text/xml image/x-icon application/octet-stream;

server_names_hash_bucket_size 64;

include /etc/nginx/conf.d/*.conf;

ssl_dhparam /etc/nginx/dhparam.pem;

ssl_session_timeout 24h;

ssl_session_cache shared:SSL:30m;

ssl_protocols TLSv1 TLSv1.1 TLSv1.2;

ssl_ciphers ECDH+AESGCM:DH+AESGCM:ECDH+AES256:DH+AES256:ECDH+AES128:DH+AES:ECDH+3DES:DH+3DES:RSA+AESGCM:RSA+AES:RSA+3DES:!aNULL:!MD5:!DSS;

ssl_buffer_size 8k;

ssl_stapling on;

resolver 8.8.8.8 8.8.4.4 valid=300s;

resolver_timeout 5s;

ssl_trusted_certificate "/etc/pki/tls/certs/ca-bundle.crt";

ssl_stapling_verify on;

ssl_prefer_server_ciphers on;

client_max_body_size 100m;

}

/etc/nginx/conf.d/default.conf 파일을 수정한다.

server {

listen *:80;

server_name syslog.enteroa.com;

return 301 https://$host$request_uri;

}

server {

listen 443 ssl http2;

server_name syslog.enteroa.com;

ssl on;

ssl_certificate /etc/letsencrypt/live/syslog.enteroa.com/fullchain.pem;

ssl_certificate_key /etc/letsencrypt/live/syslog.enteroa.com/privkey.pem;

location ^~ /.well-known/acme-challenge/ {

root /var/www/;

}

location / {

expires off;

proxy_set_header Host $http_host;

proxy_set_header X-Forwarded-Host $host;

proxy_set_header X-Forwarded-Server $host;

proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;

proxy_set_header X-Graylog-Server-URL https://syslog.enteroa.com/api;

proxy_pass http://127.0.0.1:9000;

}

12. let’s encrypt 를 이용하여 인증서 발급을 진행하였음. ( https://www.enteroa.com/2016/03/12/lets-encrypt-설치-및-운용centos )

nginx 에서 location 으로 let’s encrypt 설정을 /var/www 으로 했기 때문에 아래와 같은 명령어로 인증서 발급이 가능 하다.

~]# ./certbot-auto certonly --server https://acme-v02.api.letsencrypt.org/directory \

--rsa-key-size 4096 --agree-tos \

--email 관리자@이메일.com \

--webroot -w /var/www \

-d syslog.enteroa.com

~]# nginx -t

nginx: the configuration file /etc/nginx/nginx.conf syntax is ok

nginx: configuration file /etc/nginx/nginx.conf test is successful

~]# systemctl start nginx

13. selinux 설정

~]# yum -y install setools-console policycoreutils-python setroubleshoot-server

~]# setsebool -P httpd_can_network_connect 1

~]# setsebool -P nis_enabled 1

~]# semanage port -a -t http_port_t -p tcp 9000

~]# semanage port -a -t http_port_t -p tcp 9200

~]# semanage port -a -t mongod_port_t -p tcp 27017

~]# semanage fcontext -a -t mongod_var_lib_t "/free/mongo_data(/.*)?"

~]# restorecon -Rv /free/mongo_data

~]# setenforce 1

14. graylog 설정

graylog 서비스가 모두 정상적이라면 웹접근을 통해서 로그인을 할 수 있다.
로그인 한 뒤에 먼저 대시보드를 생성한다.
이후 input 생성을 하기위해 아래와 같이 설정 진행을 한다.
기본적으로 아래 다섯 항목을 설정 한다.
설정을 하게 되면 input 이 생성되고 곧 running 상태가 된다.

15. linux – rsyslog 설정 (UDP 설정)

~]# /etc/rsyslog.conf 파일의 맨밑에 삽입 한다. ( @=UDP / @@=TCP )

1 2	. @127.0.0.1:30511;RSYSLOG_SyslogProtocol23Format

rsyslog 을 재시작 한다.

1	~]# systemctl restart rsyslog

16. 생성된 input 의 show recived messages 버튼으로 이동하면 아래와 같은 화면을 볼 수 있다. (검색 화면과 같다.)

검색 일자를 선택하고 대쉬보드 추가를 할 수 있다.

17. L3 – syslog 설정 (cisco)

cisco 스위치의 경우 input 생성을 할때 RAW/Plaintest UDP 를 선택해서 진행해야 한다. ( RFC3414 표준이 아님 )
아울러 스위치는 일반적으로 분배의 목적으로 한개의 gateway 밑에 존재하므로 각 스위치마다 별도의 input 을 만든다
input 을 만들때 override source 옵션을 통해 특정된 port 에 들어온 메세지의 소스 호스트 네임을 알기 쉽도록 Office-L3 와 같이 설정한다.
일반적인 switch 장비의 log 설정 방법은 아래와 같다.

L3# conf t

L3(config)# logging on

L3(config)# logging traps information

L3(config)# logging host 123.123.123.123 transport udp port 12345

L3(config)# end

L3# wr

L3# show log

Trap logging: level informational, 141 message lines logged

Logging to 123.123.123.123 (udp port 12345, audit disabled,

link up),

32 message lines logged,

0 message lines rate-limited,

0 message lines dropped-by-MD,

xml disabled, sequence number disabled

filtering disabled

Logging Source-Interface: VRF Name: