데이터 엔지니어링

1. library 설치

• 커버로스 인증을 타고 하이브에 접근을 할 경우 아래 패키지들이 설치되어 있어야 한다.

--os레벨 설치 패키지

$ yum install libsasl2-dev
$ yum install cyrus-sasl-devel
$ yum install cyrus-sasl-gssapi


--python 라이브러리 패키지
impyla==0.16.2
PyHive==0.6.1
PyMySQL==0.9.3

sasl==0.2.1
thrift==0.10.0  -> issue 시 pip install thrift==0.9.3
thrift-sasl==0.4.1
thriftpy==0.3.9
thriftpy2==0.4.10

2. pyhive 연동 예제 코드

from pyhive import hive

conn = hive.Connection(host='hadoop.host.io', port=10000, database='jssvs_tmp',auth='KERBEROS',kerberos_service_name='hive')

    cursor=conn.cursor()
    cursor.execute('s select * from hive_tbl')
    for r in cursor.fetchall():
        print(r)

3. ph2 연동 예제 코드

import pyhs2
 
with pyhs2.connect (host='hadoop.host.io',
                port=10000,
                authMechanism='KERBEROS') as conn:
        with conn.cursor() as cur:
        #Show databases
                print cur.getDatabases()
 
        #Execute query
                cur.execute("select * from table")
 
        #Return column info from query
                print cur.getSchema()
 
        #Fetch table results
        for i in cur.fetch():
                print i
  
  
---- or --------------
  
import pyhs2
conn = pyhs2.connect(host='hadoop.host.io', port=10000, authMechanism='KERBEROS' )
cursor = conn.cursor()
print cursor.getDatabases()

1. aws cli 설치 ( virtualenv 환경 )

$ curl "https://s3.amazonaws.com/aws-cli/awscli-bundle.zip" -o "awscli-bundle.zip"
$ unzip awscli-bundle.zip
$ sudo ./awscli-bundle/install -i /usr/local/aws -b /usr/local/bin/aws

2. aws 계정 configure 설정

$ pip install --user virtualenv

$ virtualenv ~/cli-ve

$ virtualenv -p /usr/bin/python37 ~/cli-ve

$ pip install --upgrade awscli

$ aws configure

3. s3 관련 기본 명령어들

버킷 생성
$ aws s3 mb s3://fruits
# mb = makebucket

버킷 조회
$aws s3 ls

파일 복사하기 (단일 파일)
# aws s3 cp s3://[source] [target]
$ aws s3 cp ./jayden_sample.csv s3://jayden_test/jayden/sample/

파일 복사하기 (멀티플 파일)
$ aws s3 cp s3://[source] [target] --recursive



파일 목록 조회하기
$ aws s3 ls s3://jayden_test

--recursive 옵션
: 하위 디렉토리 까지 적용
$ aws cp . s3://jayden_test --recursive

$ aws s3 cp [source] [target]

용량 확인하기
aws s3 ls s3://[버켓]/디렉토리 --summarize --human-readable

1. 문제 소개

• 이분탐색을 이용하여 문제를 풀어야 한다

2.코드 ( python )

TOTAL_COMPUTERS=0
MIN_COMPUTERS=0
MAX_COMPUTERS=0

def getMinMinute(servers):
    global TOTAL_COMPUTERS, MIN_COMPUTERS,MAX_COMPUTERS
    start=0
    end=MAX_COMPUTERS

    while(start<end):
        mid=(start+end)//2
        sumOfComputer=0

        for i in servers:
            if i>mid:
               sumOfComputer+=mid
            else:
                sumOfComputer+=i
        #if sumOfComputer==MIN_COMPUTERS:
         #   start=end
        if sumOfComputer<MIN_COMPUTERS:
            start=mid+1
        else:
            end=mid
        
    return end
    

def solution():
    global TOTAL_COMPUTERS,MIN_COMPUTERS,MAX_COMPUTERS
    arr=[]
    N=int(input())
    for i in range (0,N):
        in_line=input().split(' ')
        tmp=[int(x) for x in in_line]
        if max(tmp)>=MAX_COMPUTERS:
            MAX_COMPUTERS=max(tmp)
        TOTAL_COMPUTERS+=sum(tmp)
        arr+=tmp


    if TOTAL_COMPUTERS%2==0:
        MIN_COMPUTERS= (TOTAL_COMPUTERS//2)
    else:
        MIN_COMPUTERS= (TOTAL_COMPUTERS//2)+1
    print(getMinMinute(arr))

if __name__=='__main__':
    solution()

3.코멘트

1. 문제 소개

• 그리디 문제로 분류 된다.
• 체인의 개수와 길이로, 하나의 긴 체인을 묶기 위한 최소한의 고리 수를 찾는 문제이다.
• 체인의 길이는 고리의 개수와 같고, 체인을 분리하거나 연결할 수 있다는 내용을 이해해야 한다.

2.코드 ( python ) 

# 고리의 최소 개수 리턴 
def getMinimumChainRing(chains,v_chains):
    # 마지막 고리 위치
    endRingPos = chains-1
    # 시작 고리 위치
    startRingPos = 0
    while(True):
    	# 고리의 위치가 만나는 지점에서 break 
        if  startRingPos == endRingPos:
#            print("minimum chain Ring count : {} ".format(chains - 1 endRingPos))
            break
        elif v_chains[startRingPos]==0:
            startRingPos+=1
        else:
            v_chains[startRingPos]-=1
            endRingPos-=1

    return (chains - endRingPos -1)



def solution():
	# 입력값 전처리 
    chains = int(input())
    org_v_chains = input().split(" ")
    #chains = 5
    #org_v_chains=[1,1,1,2]
    #org_v_chains=[4,3,5,7,9]
    v_chains = [int(x) for x in org_v_chains]
    
    # 체인 개수와 정렬된 상태의 체인 길이 배열을 매개변수로 전달 
    minRings = getMinimumChainRing(chains,sorted(v_chains))
    
    # 정답 출력
    print(minRings)


if __name__=="__main__":
    solution()

3.코멘트

• 예제 입출력이 이해가 잘 안갔지만, 함께 스터디한 찬의님의 도움으로 이해할 수 있었다.
  • ex) 5 / 4,3,5,7,8.  
  • 3짜리를 하나 빼서 4,5,7,8 사이에 하나씩 연결하면, 최소 3개의 고리가 필요하게 된다.
• 아이디어
  • 체인을 정렬한다
  • 최소 길이의 체인부터 고리를 하나씩 빼서, 마지막 위치에서부터 차례로 연결한다. 
  • 시작 포인터에서 고리를 빼고, 마지막 포인터에서 고리를 연결하여 둘이 만나는 지점에서 break 후 정답을 출력한다.

1. 레디스(Redis) 란?

• Redis = Re(remote) + di(Dictionary) + s (Server )
• 메모리를 사용하는 db
• NoSQL, Replication, Master/Slave 구조로 구성이 가능하다.
• K/V 형태로 데이터 저장.
• 파일 싱크를 제공
• 다양한 데이터 타입 제공
• NoSQL 이기 때문에 Table, db, function, row, column 이 존재 하지 않는다
• 데이터 Interaction 이 Command 를 통해서 이뤄진다.

2. 왜 레디스?

• Simple , Flexible
• Durable
• 깃헙, 스냅챗, 트위터, 스택오버플로우 등 다양한 서비스에서 사용중

3. Redis 성능 관련

• 100만개의 키들을 저장할때 70MB 정도 공간이 소모 된다.
• 100만개 - 70 MB. 1억개 - 약 7GB,
• HashType은 매우 효과적으로 인코딩된 사전구조로 되어 있어, Redis 코어 개발자중 한명인 Pieter Noordhuis가 추천함
• HashType 을 이용하여 버킷으로 저장 하는경우 100만 키 - 16MB
• 데이터를 durable 할 수 있게 파일에 쓰는 옵션이 다양한데, 안전한 옵션일 수록 redis의 속도를 느리게 한다.

4. 레디스  구성 ( on Docker)

#네트워크 생성
$sudo docker network create redis-net

#레디스 서버 
$sudo docker run --name my-redis --network redis-net -p 6379:6379 -v /home/deploy/redis_data:/data -d redis:alpine redis-server --appendonly yes


$sudo docker run --name my-redis --network redis-net \
-p 6379:6379 \
-v /Users/jssvs/WorkSpace/docker/redis_data:/data \
-v /Users/jssvs/WorkSpace/docker/redis_data/redis.conf:/usr/local/etc/redis/redis.conf \
-d redis:alpine redis-server /usr/local/etc/redis/redis.conf


#레디스 CLI
$sudo docker run -it --network redis-net --rm redis:alpine redis-cli -h my-redis

sudo docker run -it --rm redis:alpine redis-cli -h my-redis
$vi redis.conf
requirepass 1111  


# redis. conf
#인증 패스워드 설정
requirepass 1111

#스냅샷 파일 저장 디렉토리 경로 설정
dir /data

#스냅샷 db 파일 이름 설정
dbfilename dump.rdb

#AOF 설정 (Append-Only File) OS 레벨에서 맛탱이 갔을 때를 대비해 
appendonly yes

# 60초마다 1000번 이상 데이터 변경 시 저장
save 60 1

5. 레디스 Client 커맨드 (기본)

인증
>AUTH 1111[PASS]

키 가져오기
>GET [keyname]

키 넣기
>set [keyname] [data]

키 삭제
>del [keyname]

키 스페이스 정보
>info keyspace

키 삭제(플러시)
> flushall
> flushall async

6. 레디스 hashset 예제 코드 ( python )

from redis import StrictRedis

r=StrictRedis(host='localhost',password='1111')

user = {"name":"jayden", "Company":"home", "Address":"Suwon","score":60}

r.hmset("jayden", user)

r.hgetall("jayden")

Docker 이용하여 mysql, redis 컨테이너 올리기

$ mkdir /usr/local/db_data
$ mkdir /usr/local/redis_data

$ docker pull mysql:8.0
$ sudo docker run -d -p 3306:3306 -e MYSQL_ROOT_PASSWORD=test --volume /usr/local/db_data/:/var/lib/mysql --name mysql_bi mysql:8  --default-authentication-plugin=mysql_native_password 

$ docker pull redis:5.0
$ sudo docker run -d --name redis_bi -p 6379:6379 --volume /usr/local/redis_data/:/data -e REDIS_PASSWORD=airflow redis:5.0

$ sudo docker ps

virtualenv 설치 및 가상 환경 만들기

# Python 을 이용하여 virtualenv 설치
$ python3 -m pip install virtualenv virtualenvwrapper --user

# 가상 환경 만들기
$ mkdir myenv
$ virtualenv --python=python3.5 myenv

#아래 명령어로도 가능하다
$ virtualenv -p `which python3` venv 

# 가상환경 진입
$ source myenv/bin/activate

#가상환경 빠져 나오기
$ deactivate

airflow 설치 및 서버 동작 스크립트

$ source myenv/bin/activate
$ pip install --upgrade pip
$ pip install pymysql

# airflow
pip install apache-airflow[mysql,redis,celery]==1.10.7

# 에어플로우 홈 설정
$ mkdir myenv/airflow_home
$ export AIRFLOW_HOME=`pwd`/airflow_home # virtualenv 바깥에서도 동일하게 설정

# 에어플로우 데이터베이스 초기화
$ airflow initdb
$ airflow resetdb #load_examples False 가 적용이 안될때 다시 db 리셋

# 웹서버 구동
$ airflow webserver -p 5000
$ airflow scheduler
$ airflow worker

$ airflow flower

airflow.cfg 설정 파일 수정

$ vi airflow.cfg

dags_folder = /home/deploy/work/airflow/dags

# The folder where airflow should store its log files
# This path must be absolute
base_log_folder = /home/deploy/work/airflow/logs

# executor = SequentialExecutor
executor = CeleryExecutor

# sql_alchemy_conn = sqlite:////home/airflow/airflow/airflow.db
sql_alchemy_conn =  mysql+pymysql://airflow:airflow@127.0.0.1:3306/airflow

# catchup_by_default = True
catchup_by_default = False

# broker_url = sqla+mysql://airflow:airflow@127.0.0.1:3306/airflow
broker_url = redis://airflow@127.0.0.1:6379/0

# result_backend = db+mysql://airflow:airflow@localhost:3306/airflow
result_backend = db+mysql://airflow:airflow@127.0.0.1:3306/airflow

# load_examples = True
load_examples = False

airflow 자주 쓰는 커맨드

# print the list of active DAGs
airflow list_dags

# prints the list of tasks the "tutorial" dag_id
airflow list_tasks jayden_tutorial

# prints the hierarchy of tasks in the tutorial DAG
airflow list_tasks jayden_tutorial --tree


# command layout: command subcommand dag_id task_id date
$ airflow test jayden_tutorial print_date 2020-03-02

다음에는 Hive Operator 연동을 올리겠다. :D 

• RDD의 연산 종류는 transformation 과 action 이 있다.
• 액션은 데이터를 드라이버로 되돌려주든지(count, collect 등) 데이터를 외부 저장 시스템에 쓰는것(CopyToHadoop) 등의 일이다.
• 액션은 스케쥴러를 시작하게 하며, RDD 트랜스포메이션 간의 종속성에 바탕을 둔 DAG 를 생성한다.
• 트랜스포메이션의 로직 에러가 발생할 때, 지연 평가 때문에, 액션이 수행된 지점에서 실패한 것으로 나타나는 경우에 유의하자
• word count 프로그램에서 null pointer exception 이 발생한다고 가정할때, 코드가 contains를 체크하는 시점이 아니라 collect 단계에서 예외가 발생한다.

• 맵리듀스와 비교해 스파크 성능상 이점은 반복 연산이 들어있는 사례이다.
• 스파크가 처리하는 데이터를 디스크에 기록하지 않고 익스큐터 메모리에 데이터를 로드해 놓는 것이다.
• 메모리 관리의 3가지 옵션
1. 메모리에 직렬화되지 않은 자바 객체 - RDD 객체를 직렬화 하지 않고 그대로 저장한다. ( 직렬화 비용이 안드는 대신 메모리 공간 사용이 비 효율이다.)
2. 메모리에 직렬화된 데이터 - RDD 객체를 네트워크 전송이 가능한 바이트 스트림으로 변환한다. ( 데이터를 읽는데 CPU 가 더 많이 사용되므로 접근방식은 더 느리지만 메모리 공간 사용 측면에서 효율적이다. 크리오(Kryo) 직렬화를 쓰면 공간 측변에서도 효과적이다.)
3. 디스크 - 익스큐터 램에 담기에 파티션이 큰 RDD 라면 디스크에 데이터를 쓸 수 있다. ( 반복 연산시 속도 면에서 불리하지만 장애에 안전하다.)
• persist() 의 기본 옵션은 RDD를 메모리에 직렬화되지 않은 상태로 저장한다.

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.
• RDD 생성 방식
• 기존 RDD에 Transformation 호출
• SparkContext 객체로부터 생성
• DataSet이나 DataFrame을 변환 ( 이것들은 SparkSession 으로 부터 만들어짐)
• SparkContext는 스파크 클러스터와 실행중인 스파크 애플리케이션 하나와의 연결을 나타낸다.
• RDD 속성을 알 수 있는 함수
• partitions() - 분산 데이터 셋의 부분들을 구성하는 파티션 객체들의 배열을 리턴한다. getPartition()의 결괏값과 같다
• iterator(p,parentIters) - 각각의 부모 파티션을 순회하는 반복자가 주어지면 파티션 p의 구성요소들을 계산해낸다.
• dependencies() - 종속성 객체의 목록을 리턴한다. 스케쥴러가 현재의 RDD가 어떤식으로 다른 RDD에 종속될지 알려준다.
• partitioner() - element 와 partition 사이에 연관되는 함수를 갖고 있는 RDD라면 스칼라의 option 타입으로 partitioner 객체를 리턴한다. 
• perferredLocations(p) - 파티션 p의 데이터 지역성에 대한 정보를 리턴한다. p가 저장된 각 노드의 정보를 문자열로 표현한 시퀀스를 리턴한다

• RDD는 정적인 타입인 데다 불변한 성격을 가지고 있어, Transformation 을 호출하는 것이 새롭게 정의한 속성들을 가진 새로운 RDD를 리턴하는 행위다.

• 종속성이 넓으냐 좁으냐는 트랜스포메이션 평가에 중요한 영향을 끼치며 성능에도 크게 작용한다.
• 좁은 종속성
• 자식 RDD 의 각 파티션이 부모 RDD의 파티션들에 대해 단순하고 한정적인 종속성을 갖는다
• 부모가 최대 하나의 자식파티션을 갖는 경우
• map, filter, mapPartitions 등의 연산이 이 패턴을 따른다.
• 넓은 종속성
• 자식 RDD가 다수의 부모 RDD의 파티션들과 관계를 맺고 있는 경우
• groupbykey, sort, reducebykey 등과 같이 Shuffle 을 일으키는 함수가 이 패턴을 따른다.
• 셔플 비용이 가장 크다.
• map 은 파티션 간 이동이 없는 연산, coalesce 는 파티션을 합치는 연산으로 파티션 개수를 줄이는 목적의 함수이다.
• join 함수는 두개의 부모 RDD 가 어떻게 파티션되었는지에 따라 좁거나 넓은 종속성을 가질 수 있다.

• 잡 실행 과정
• 스파크 프로그램 자체는 드라이버 노드에서 실행되며 일련의 명령들을 익스큐터에게 보낸다.
• 애플리케이션들은 클러스터 매니저에 의해 스케쥴링되고, 각각 하나의 SparkContext를 가진다.
• 스파크 애플리케이션들은 공존하는 여러 개의 잡을 차례로 실행할 수 있다.
• Job들은 애플리케이션의 한 RDD가 호출하는 각 액션에 대응한다.
• 자원할당
• 정적할당과 동적 할당이 가능
• 스파크 애플리케이션
• 잡들은 드라이버 프로그램의 SparkContext에 정의되어 있다.
• SparkContext가 생기면 스파크 애플리케이션이 구동한다.
• SparkContext를 실행하면 드라이버와 익스큐터들이 각 작업 노드에서 구동된다.
• 각 익스큐터는 JVM을 가지며 한 노드에 여러 개의 익스큐터가 존재할 수 있다.
• 스파크 잡이 실행될 때 익스큐터는 RDD를 계산할 태스크 실행을 위한 슬롯을 가진다.
• 스파크 잡 해부
• 각 액션마다 스파크 스케쥴러는 실행 그래프를 만들고 스파크 잡을 실행한다. 
• 각 잡은 최종 RDD를 만들어 내는데 필요한 데이터 변환의 각 단계를 의미하는 스테이지(Stage)들로 구성된다.
• 각 스테이지는 각 병렬 연산의 한 단위를 의미하며 익스큐터들 위에서 실행되는 다수의 태스크(Task)들로 구성된다.
• Spark Aplication -> 잡 -> 스테이지1 & 스테이지2 ... -> 태스크 1& 태스크 2... 
• SparkContext / SparkSession -> 액션 -> 넓은 포메이션 -> 하나의 파티션 평가를 위한 연산 
• 잡
• 스파크 실행 구성에서 가장 높은 단계
• 하나의 잡 = 하나의 액션에 대응하며, 액션은 스파크 애플리케이션의 드라이버의 프로그램에서 호출한다.
• 스테이지
• 잡의 부분들을 스테이지로 정의한다.
• 넓은 트랜스포메이션에 의해 생성되는 셔플 의존성에 대응한다.
• 하나의 스테이지는 다른 익스큐터나 드라이버와의 통신 없이 계산 가능한 태스크들의 집합으로 생각할 수 있다.
• 태스크
• 하나의 스테이지는 여러 개의 태스크로 이루어진다.
• 실행 계층에서 가장 작은 단위이며, 익스큐터가 태스크 실행을 위해 동적으로 할당된 여러개의 슬롯을 가진다.
• 스테이지당 태스크의 개수는 해당 스테이지의 결과 RDD의 파티션 개수에 대응된다.
• 익스큐터 코어 개수의 총합 = 익스큐터당 코어의 개수 X 익스큐터 개수를 쓰면 sparkConf로 부터 동시 실행의 태스크 개수를 구할 수 있다.
• 태스크 분산 과정은 TaskScheduler 가 담당하는데, 페어 스케쥴러인지 FIFO 스케쥴러인지 등에 따라 다르다.