태그 Archives: csv

GEOIP database 파일 업데이트

그렇다. 또 바뀌었다 ‘ㅅ’….

GeoIP database 파일을 다운 받는데 링크 주소가 변경 되었다

맥스마인드 사이트의 회원 가입 이후 라이선스 키를 받아야 다운로드를 받을 수 있도록 변경 되었다.

crontab 에 아래와 같이 geoipupdate 명령어를 주기적으로 실행 함으로서 업데이트를 실행할 수 있다.

~]# crontab -e

------------------------------------------------------

00 06 * * 5 /bin/geoipupdate # GeoIP database update.

------------------------------------------------------

단지 위 명령어를 실행 하기 위해 맥스마인드사의 회원 가입 과 라이선스 키를 발급 받아야 한다.

그리고 /etc/GeoIP.conf 파일에서 라이선스키 등록을 진행해야 한다.

1 2	AccountID 123456 LicenseKey ABCDEFGHIJKL

이후 geoipupdate명령어를 통해서 업데이트가 잘 되나 실행해보면 된다.

1	~]# geoipupdate

출처 : https://dev.maxmind.com/geoip/geoipupdate/

다른 방법으로는 기존 처럼 스크립트를 이용 하여 바뀐 URL 에 파라메터로 라이선스 키를 넣어서 호출하는 방법이 있다.

개인 적으로 남이 만든 프로그램 보단 직접 하는 스크립트를 선호하는 편이라.. (OS 의존성도 없는 편이고 해서 ‘ ㅅ’a)

스크립트 3 번째 줄에 위 방법을 따라 만든 maxmind 라이선스 키를 입력해야 한다.

#!/bin/bash

#################################

Maxmind_Licensekey="ABCDEFGHIJKL"

#################################

CITYDATA="N" ### config - DISABLE city it'll be need free memory 2GB

### geoip set

GEOIPDIR="/usr/share/GeoIP"

DATALINK="/usr/share/xt_geoip /var/lib/GeoIP"

### avoid overlap

lockfile=/var/lock/$(basename $0)

if [ -f $lockfile ];then

P=$(cat $lockfile)

if [ -n "$(ps --no-headers -f $P)" ];then

exit 1

echo $$ > $lockfile

trap 'rm -f "$lockfile"' EXIT

### define server are primary or secandary.

### install dependances

if [[ -z $(which git) ]];then sudo yum -y install git > /dev/null 2>&1 ;fi

if [[ -z $(which pip) ]];then sudo yum -y install python2-pip > /dev/null 2>&1;fi

if [[ -z $(pip list|grep pygeoip) ]];then sudo pip install pygeoip > /dev/null 2>&1 ;fi

if [[ -z $(pip list|grep ipaddr) ]];then sudo pip install ipaddr > /dev/null 2>&1 ;fi

### link path

if [[ ! -d $GEOIPDIR ]];then

mkdir -p $GEOIPDIR

for a in $DATALINK

if [[ ! -d $a ]];then if [[ $(readlink $a) != $GEOIPDIR ]];then

rm -rf $a;ln -s $GEOIPDIR $a

fi;fi

done

### https://github.com/sherpya/geolite2legacy - csv to data file.

cd $GEOIPDIR

if [ ! -e $GEOIPDIR/geolite2legacy/geolite2legacy.py ];then cd $GEOIPDIR

git clone https://github.com/sherpya/geolite2legacy.git

### make GeoIP.dat files from GeoLite2 CSV file.

if [ -d $GEOIPDIR/geolite2legacy ];then

cd $GEOIPDIR/geolite2legacy

EXT="zip"

CIF="GeoLite2-City-CSV"

COF="GeoLite2-Country-CSV"

BASEURL="https://download.maxmind.com/app/geoip_download?edition_id"

COF_URL="$BASEURL=$COF&license_key=$Maxmind_Licensekey&suffix=$EXT"

ORI_DATE=$(date +"%Y%m%d%H%M.%S" -d "$(curl -sI $COF_URL|grep -i ^Last-Modified:|cut -d, -f2)")

CSV_DATE=$(date +"%Y%m%d%H%M.%S" -d "$(stat -c %y $COF.$EXT)")

if [[ "$ORI_DATE" != "$CSV_DATE" ]];then

rm -f $COF.$EXT $CIF.$EXT

wget "$COF_URL" -O $COF.$EXT >/dev/null 2>&1

if [ -s $GEOIPDIR/geolite2legacy/$COF.$EXT ];then

python geolite2legacy.py --input-file $COF.$EXT --fips-file geoname2fips.csv --output-file GeoIP.dat

python geolite2legacy.py --input-file $COF.$EXT -6 --fips-file geoname2fips.csv --output-file GeoIPv6.dat

if [[ $CITYDATA == "Y" ]];then

CIF_URL="$BASEURL=$CIF&license_key=$Maxmind_Licensekey&suffix=$EXT"

wget "$CIF_URL" -O $CIF.$EXT >/dev/null 2>&1

python geolite2legacy.py --input-file $CIF.$EXT --fips-file geoname2fips.csv --output-file GeoLiteCity.dat

python geolite2legacy.py --input-file $CIF.$EXT -6 --fips-file geoname2fips.csv --output-file GeoLiteCityv6.dat

touch -t $ORI_DATE GeoLiteCity.dat GeoLiteCityv6.dat

mv -f Geo{LiteCity,LiteCityv6}.dat $GEOIPDIR/ >/dev/null 2>&1

touch -t $ORI_DATE GeoIP.dat GeoIPv6.dat $COF.$EXT

mv -f Geo{IP,IPv6}.dat $GEOIPDIR/ >/dev/null 2>&1

exit 0

기존 스크립트에서 추가 변경된 기능이 있는데 DB 업데이트가 이루어 졌는지 배포주소의 HEAD 체크를 통해서 Last-Modified 를 체크 한다.

쉽게 말하면 맥스마인드사에서 새로운 db를 업데이트 할때만 작동 한다는 소리다. ( 더 쉽게: 매일 돌리면 된다. )

때문에 위 스크립트는 /etc/cron.daily/ 폴더내에 넣어두면 필요할 때 자동 실행이 된다.

출처 : https://dev.maxmind.com/geoip/geoip-direct-downloads/#Direct_Downloads

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. 이건 저도 어려 웠어요…..

GEOIP data 파일 업데이트

아래 스크립트는 배포 url 이 변경되어 동작 되지 않아요.

새로운 업데이트 방법 관련 글 : GEOIP database 파일 업데이트

GeoIP.dat 파일을 배포하는 maxmind (https://www.maxmind.com/) 사에서 레거시 파일 형태인 GeoIP.dat 파일을 2019-01-02 일로 부터 배포를 중단하였다.

기존에 apache 혹은 nginx 그리고 iptables 등에서 사용중인 국가별 구분 및 제한을 을 지속적으로 사용 하기 위해서는

GeoIP의 지속적인 업데이트가 필요로 한다.

– 주기적으로 ip는 로테이션이 되기 때문에 dat 파일이 갱신되지 않은채 1년 정도가 지나면 신뢰도가 떨어진다.

다만 과거 형태 .dat 파일만 배포 중단이며 .mmdb 및 .csv 형태의 배포는 지속되고 있다.

때문에 배포되는 csv 파일을 이용 하여 dat 파일을 직접 생성해 주는 방향으로 진행할 수 있다.

이미 github 에서 csv를 이용한 dat 파일을 만드는 게 몇몇 올라와 있다. 이중 https://github.com/sherpya/geolite2legacy 을 이용하여 생성할 수 있는 스크립트를 작성 하였다 ‘ㅅ’a

#!/bin/bash

### config - DISABLE city it'll be need free memory 2GB

REPLACEDATA="30"

CITYDATA="N"

### avoid overlap

lockfile=/var/lock/$(basename $0)

if [ -f $lockfile ];then

P=$(cat $lockfile)

if [ -n "$(ps --no-headers -f $P)" ];then

exit 1

echo $$ > $lockfile

trap 'rm -f "$lockfile"' EXIT

### install dependances

if [[ -z $(which git) ]];then sudo yum -y install git > /dev/null 2>&1 ;fi

if [[ -z $(which pip) ]];then sudo yum -y install python2-pip > /dev/null 2>&1;fi

if [[ -z $(pip list|grep pygeoip) ]];then sudo pip install pygeoip > /dev/null 2>&1 ;fi

if [[ -z $(pip list|grep ipaddr) ]];then sudo pip install ipaddr > /dev/null 2>&1 ;fi

### link path

GEOIPDIR="/usr/share/GeoIP"

DATALINK="/usr/share/xt_geoip /var/lib/GeoIP"

if [[ ! -d $GEOIPDIR ]];then

mkdir -p $GEOIPDIR

for a in $DATALINK

if [[ ! -d $a ]];then if [[ $(readlink $a) != $GEOIPDIR ]];then

rm -rf $a;ln -s $GEOIPDIR $a

fi;fi

done

### https://github.com/sherpya/geolite2legacy

cd $GEOIPDIR

if [ ! -e $GEOIPDIR/geolite2legacy/geolite2legacy.py ];then cd $GEOIPDIR

git clone https://github.com/sherpya/geolite2legacy.git

### make GeoIP.dat files

if [ -d $GEOIPDIR/geolite2legacy ];then cd $GEOIPDIR/geolite2legacy

COUNTRY="GeoLite2-Country-CSV.zip"

CITY="GeoLite2-City-CSV.zip"

if [[ -z $(find $GEOIPDIR/geolite2legacy -maxdepth 1 $ -name $COUNTRY -o -name $CITY $ -mtime +$REPLACEDATA) ]];then

rm -f $COUNTRY $CITY

wget "https://geolite.maxmind.com/download/geoip/database/GeoLite2-Country-CSV.zip" > /dev/null 2>&1

if [[ $CITYDATA == "Y" ]];then

wget "https://geolite.maxmind.com/download/geoip/database/GeoLite2-City-CSV.zip" > /dev/null 2>&1

if [ -e $GEOIPDIR/geolite2legacy/$COUNTRY ];then

python geolite2legacy.py --input-file $COUNTRY --fips-file geoname2fips.csv --output-file GeoIP.dat

python geolite2legacy.py --input-file $COUNTRY -6 --fips-file geoname2fips.csv --output-file GeoIPv6.dat

if [[ $CITYDATA == "Y" ]];then

python geolite2legacy.py --input-file $CITY --fips-file geoname2fips.csv --output-file GeoLiteCity.dat

python geolite2legacy.py --input-file $CITY -6 --fips-file geoname2fips.csv --output-file GeoLiteCityv6.dat

if [ -e GeoIP.dat ];then

find $GEOIPDIR -maxdepth 1 -name "*.dat" -mtime +$REPLACEDATA -exec rm -f {} \;

if [[ $CITYDATA == "Y" ]];then

mv -f Geo{IP,IPv6,LiteCity,LiteCityv6}.dat $GEOIPDIR/ > /dev/null 2>&1

else

mv -f Geo{IP,IPv6}.dat $GEOIPDIR/ > /dev/null 2>&1

fi;fi;fi

unset GEOIPDIR DATALINK CITY COUNTRY CITYDATA REPLACEDATA

exit 0

GeoLiteCity.dat 는 최상단의 config CITYDATA 를 Y 로 지정하면 생성한다. (일반적으로 불필요 하며, 생성시 2GB 이상의 메모리를 사용 한다.)

cron 등록은 /etc/cron.daily/ 혹은 /etc/cron.weekly/ 에 파일을 만들어 두면 anacron 이 알아서 주 or 일에 돌려주겠지요..