[ AWS ] AWS를 통한 Data Lake 디자인 (1)

 

코세라의 "Introduction to Designing Data Lakes on AWS" 강의는

AWS개발자들이 생각하는 좋은 Data Lake 아키텍처를 설명하고 있다.

 

위 그림은 배치레이어, 스피드레이어(실시간)를 둘 다 구성할 수 있으며

키바나를 통해 집계그래프로 시각화한다.

[ Data Lake Related  Store Services ]

S3

실시간이든 배치든 저장, 인덱싱, 프로세싱 등 S3가 중심이 된다.

Data Lake의 저장소로 고려해야 될 사항은 네트워크 성능, 접근시간, 용량, 비용, 보안 등 

여러 요소가 있다. 

AWS S3는 위와 같은 요소들을 가지면서 개발자가 환경을 구축시간을 줄여준다.

S3가 가지는 장점은 아래와 같다. 

• Scalable( 확장성), no limit data ( 용량제한이 없음)
• Durable, 데이터내구성을 99% 보장한다.
• Any Type of Data, 데이터의 다형성

또한 Retrieval Timed에 따라 여러 S3의 제품을 선택하여 사용할 수 있다.

즉 자주 접근하는 데이터 일수록 더 비싼 S3를 사용해야 한다.

GLUE

S3과 함께 자주 사용되는 AWS Glue는 

Serverless한 데이터처리 및 카탈로깅 서비스이며, 중앙 메타데이터 저장소이다.

도서관으로 예를 들면, 도서관은 수많은 책(Data)들에게 장르를 부여(cataloging)하여

원하고자 하는 책을 빨리 찾을 수 있게한다.

더욱이 Glue는 다양한 Data소스에서 데이터를 받아올 수 있다.

참조

S3 :

• https://aws.amazon.com/products/storage/data-lake-storage/
• https://docs.aws.amazon.com/AmazonS3/latest/dev/storage-class-intro.html

Glue : 

• https://docs.aws.amazon.com/glue/latest/dg/add-classifier.html

[ Data Movement ]

참조 : https://aws.amazon.com/big-data/datalakes-and-analytics/https://aws.amazon.com/big-data/datalakes-and-analytics/what-is-a-data-lake/

API GateWay

AWS API Gateway는 보안 API를 쉽게생성하며, 유지 관리를 쉽게 지원하는 서비스이다.

자체 백엔드에서 구현해도 되지만 트래픽관리, 권한 및 제어, 모니터링들을 더 용이하게 해준다. 

API를 통해 이동되는 데이터들은 위의 이미지와 같이 EC2, 람다, 키네시스로 전달 할 수 있다.

API가 아니라 바로 파일로 저장되는 경우, Web Server-> Kinesis-> S3로 구성할 수 있다.

Kinesis

실시간 스트리밍 데이터를 쉽게 수집, 처리, 분석할 수 있도록 지원하는 서비스이다.

소규모든 대규모든 데이터가 도착하는대로 처리하고 분석을 즉시 할 수 있다.

데이터 유형과 목적에 따라 다양한 Kinesis 제품을 이용할 수 있다.

• Kinesis Video Streams
• Kinesis Data Streams
• Kinesis Data Firehose

참조 : https://aws.amazon.com/kinesis/  

[ Data Processing ]

데이터를 이동시켰으면 데이터처리를 해야하는데 

Apache Hadoop을 이용한 분산처리시스템을 이용한다.

https://aws.amazon.com/emr/details/hadoop/what-is-hadoop/

EMR

EMR은 Hadoop을 EC2에 쉽게 추가할 수 있는 환경을 만들고,

분산처리에 필요한 어플리케이션( Hbase, Hive, Spark, Phoenix 등 )들을 설치 할 수 있다.

EMR을 사용하지 않고 EC2자체에 Ambari나 CDH(클라우데라 매니저)를 설치해 

Hadoop을 사용할 경우 비용적으로 이득이 있지만, 

개인적인 경험으로 운영시에 EMR이 인적자원이 훨씬 덜 소모되었다. 

https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-what-is-emr.html  

EMR을 Serverless한 아키텍처로 Glue(+Lambda)를 사용할 수도 있다.

Glue Jobs : https://docs.aws.amazon.com/glue/latest/dg/author-job.html

Lambda : https://docs.aws.amazon.com/lambda/latest/dg/welcome.html

Athena Redshift

Sql을 이용하여 작업 후 저장할 때는 athena나 redshift는 좋은 선택지가 된다

참조

Athena : https://docs.aws.amazon.com/athena/latest/ug/what-is.html

RedShift : https://aws.amazon.com/redshift/?whats-new-cards.sort-by=item.additionalFields.postDateTime&whats-new-cards.sort-order=desc  

sdf

[ Algorithm ] 기능개발 - Queue

[ 문제 ]

프로그래머스 팀에서는 기능 개선 작업을 수행 중입니다. 각 기능은 진도가 100%일 때 서비스에 반영할 수 있습니다.

또, 각 기능의 개발속도는 모두 다르기 때문에 뒤에 있는 기능이 앞에 있는 기능보다 먼저 개발될 수 있고, 이때 뒤에 있는 기능은 앞에 있는 기능이 배포될 때 함께 배포됩니다.

먼저 배포되어야 하는 순서대로 작업의 진도가 적힌 정수 배열 progresses와 각 작업의 개발 속도가 적힌 정수 배열 speeds가 주어질 때 각 배포마다 몇 개의 기능이 배포되는지를 return 하도록 solution 함수를 완성하세요.

제한 사항

작업의 개수(progresses, speeds배열의 길이)는 100개 이하입니다.

작업 진도는 100 미만의 자연수입니다.

작업 속도는 100 이하의 자연수입니다.

배포는 하루에 한 번만 할 수 있으며, 하루의 끝에 이루어진다고 가정합니다. 예를 들어 진도율이 95%인 작업의 개발 속도가 하루에 4%라면 배포는 2일 뒤에 이루어집니다.

입출력 예

progresses	speeds	return
[93, 30, 55]	[1, 30, 5]	[2, 1]
[95, 90, 99, 99, 80, 99]	[1, 1, 1, 1, 1, 1]	[1, 3, 2]

[ 풀이 ]

큐를 이용하여 푸는 문제지만 그렇지 않은 코드도 첨부했다.

progresses의 각 배열의 완료일을 따로 저장한다음

Ex) [ 93, 30, 55 ] -> [ 7, 3, 9 ]

특정 배열 인덱스 값이 그 뒤의 배열 값과 비교, 큰 값이 나올 때까지 반복문을 돈다.

Ex) [ 7, 3, 9 ]  -> 7은 9보다 작으므로 3까지 반복문을 돈다.

7과 3, 2개의 값이 결과 값으로 2를 리턴배열에 input한다.

[ 제출1 ]

import java.util.*;

class Solution {

    public int[] solution(int[] progresses, int[] speeds) {

        int[] temp = new int[100];

        int day = -1;

        for(int i = 0; i < progresses.length; i++){

            while(progresses[i] + (speeds[i] * day) < 100){

                day++;

            }

            temp[day]++;

        }

        int cnt = 0;

        

        for(int n : temp){

            if(n != 0) cnt++;

        }

        

        int[] answer = new int[cnt];

        

        cnt = 0;

        for(int n : temp){

            if(n != 0) answer[cnt++] = n;

        }

        return answer;

    }

}

[ 제출2 - 람다식이용 ]

int[] dayOfend = new int[100];

int day = -1;

for(int i=0; i<progresses.length; i++) {

    while(progresses[i] + (day*speeds[i]) < 100) {

        day++;

    }

    dayOfend[day]++;

}

return Arrays.stream(dayOfend).filter(i -> i!=0).toArray();

제출1

테스트 1 〉	통과 (0.02ms, 77.4MB)
테스트 2 〉	통과 (0.03ms, 76.1MB)
테스트 3 〉	통과 (0.02ms, 76.7MB)
테스트 4 〉	통과 (0.03ms, 76.3MB)
테스트 5 〉	통과 (0.02ms, 73.3MB)

제출2

테스트 1 〉	통과 (3.39ms, 74.2MB)
테스트 2 〉	통과 (3.16ms, 77.1MB)
테스트 3 〉	통과 (2.88ms, 78.8MB)
테스트 4 〉	통과 (3.20ms, 84.3MB)
테스트 5 〉	통과 (1.92ms, 75.2MB)

람다식을 이용할 경우 코드가 간결하나 처리속도가 느림

[ 제출3 - 큐를 이용 ]

List<Integer> answerList = new ArrayList<>();

for (int i = 0; i < speeds.length; i++) {

    

    //종료시점 예측 = 소수점반올림((100 - 진행작업갯수) / 진행속도 )

    double remain = (100 - progresses[i]) / (double) speeds[i];

    int date = (int) Math.ceil(remain);

    if (!q.isEmpty() && q.peek() < date) {

        answerList.add(q.size());

        q.clear();

    }

    q.offer(date);

}

answerList.add(q.size());

int[] answer = new int[answerList.size()];

for (int i = 0; i < answer.length; i++) {

    answer[i] = answerList.get(i);

}

return answer;ys.stream(dayOfend).filter(i -> i!=0).toArray();

[ Scala ] Date Rang Foreach 예시

[ Code ]

import java.time.LocalDate

import java.time.format.DateTimeFormatter

val format = DateTimeFormatter.ofPattern("yyyy-MM-dd")

val start_dt = LocalDate.parse("2020-02-01", format)

val end_dt = LocalDate.parse("2020-02-20", format)

val date_diff = end_dt.toEpochDay() - start_dt.toEpochDay()

for (i <- 0l to date_diff by 1) {

  val s = start_dt.plusDays(i)

  val f_date = start_dt.plusDays(i)

  val date = f_date.toString().replaceAll("-","")

  val date_s = f_date + " 00:00:00"

  val date_e = f_date + " 23:59:59"

  val yymm = date.slice(0,6)

  println(s)

}

import java.time.LocalDate

import java.time.format.DateTimeFormatter

val format = DateTimeFormatter.ofPattern("yyyy-MM-dd")

val start_dt = LocalDate.parse("2021-05-04", format)

val end_dt = LocalDate.parse("2021-05-31", format)

val date_diff = end_dt.toEpochDay() - start_dt.toEpochDay()

[ Result ]

2020-02-01
2020-02-02
2020-02-03
2020-02-04
...
2020-02-18
2020-02-19
2020-02-20

[ Algorithm ] 로또의 최고 순위와 최저 순위

문제 ]

특정조건일때 각기다른 2개의 배열이 같은 값은 몇개 가지는지를

비교하는 문제

https://programmers.co.kr/learn/courses/30/lessons/77484

순위 당첨 내용

1 6개 번호가 모두 일치

2 5개 번호가 일치

3 4개 번호가 일치

4 3개 번호가 일치

5 2개 번호가 일치

6(낙첨) 그 외

구매했던 로또로 당첨이 가능했던 최고 순위와 최저 순위를 알아보고 싶어 졌습니다.

알아볼 수 없는 번호를 0으로 표기하기로 하고, 구매한 로또 번호 6개가 44, 1, 0, 0, 31 25라고 가정해보겠습니다. 당첨 번호 6개가 31, 10, 45, 1, 6, 19라면, 당첨 가능한 최고 순위와 최저 순위의 한 예는 아래와 같습니다.

로또 당첨 번호        31 10          45 1 6         19 결과

최고 순위 번호 31 0→10  44 1 0→6 25 4개 번호 일치, 3등

최저 순위 번호 31 0→11  44 1 0→7 25 2개 번호 일치, 5등

순서와 상관없이, 구매한 로또에 당첨 번호와 일치하는 번호가 있으면 맞힌 걸로 인정됩니다.

구매한 로또 번호를 담은 배열 lottos, 당첨 번호를 담은 배열 win_nums가 매개변수로 주어집니다. 이때, 당첨 가능한 최고 순위와 최저 순위를 차례대로 배열에 담아서 return 하도록 solution 함수를 완성해주세요.

제한사항

lottos는 길이 6인 정수 배열입니다.

lottos의 모든 원소는 0 이상 45 이하인 정수입니다.

0은 알아볼 수 없는 숫자를 의미합니다.

0을 제외한 다른 숫자들은 lottos에 2개 이상 담겨있지 않습니다.

lottos의 원소들은 정렬되어 있지 않을 수도 있습니다.

win_nums은 길이 6인 정수 배열입니다.

win_nums의 모든 원소는 1 이상 45 이하인 정수입니다.

win_nums에는 같은 숫자가 2개 이상 담겨있지 않습니다.

win_nums의 원소들은 정렬되어 있지 않을 수도 있습니다.

입출력 예

lottos win_nums result

[44, 1, 0, 0, 31, 25] [31, 10, 45, 1, 6, 19] [3, 5]

[0, 0, 0, 0, 0, 0] [38, 19, 20, 40, 15, 25] [1, 6]

[45, 4, 35, 20, 3, 9] [20, 9, 3, 45, 4, 35] [1, 1]

제출 ] 

import java.util.*;
import java.util.stream.IntStream;
import java.util.stream.Collectors;
import java.util.Iterator;

class Solution {
    public int[] solution(int[] lottos, int[] win_nums) {

        int[] answer = {7,7};
        Iterator<Integer> iter = 
            Arrays.stream(lottos).boxed().collect(Collectors.toList()).iterator();
        
        while (iter.hasNext()) {
            int tmp = iter.next();
            if(IntStream.of(win_nums).anyMatch(x -> x == tmp)){
                answer[0]--;
                answer[1]--;
            }
            if(tmp == 0) answer[0]--;
        }
        
        if(answer[0]==7) answer[0] = 6;
        if(answer[1]==7) answer[1] = 6;
        return answer;
    }
}

풀이 ]

int[] answer = { 최고순위, 최저순위 } 로 7,7로 시작해 

lottos의 배열 값이 win_nums배열에 있으며 최고순위, 최저순위 각 -1씩

lottos의 배열 값이 0 이면 최고순위 -1씩 하여 반환한다. 

Ex) 5개가 맞으면 7 - 5 = 2, 즉 2등

다른 사람 풀이 ]

import java.util.Arrays;
import java.util.stream.LongStream;

class Solution {
    public int[] solution(int[] lottos, int[] winNums) {

        return LongStream.of(
            (lottos.length + 1) - Arrays.stream(lottos)
            .filter(l -> Arrays.stream(winNums).anyMatch(w -> w == l) || l == 0)
            .count(),
            (lottos.length + 1) - Arrays.stream(lottos)
            .filter(l -> Arrays.stream(winNums).anyMatch(w -> w == l))
            .count()
        ).mapToInt(op -> (int) (op > 6 ? op - 1 : op)).toArray();

    }
}

람다식으로 표현

return Stream.of( 최고순위, 최저순위).(리턴값 7일 경우 예외처리)

[ Kubernetes ] 아키텍처

[ 쿠버네티스 아키텍쳐 ] 

1. 컨트롤 플레인

컨트롤 플레인은 클러스터의 두뇌 역할을 하며 

사용자의 API요청, CLI, UI서비스를 제공한다.

좀 더 자세히 살펴보면

아래 그림의 실선의 쿠버네티스 마스터 영역이 컨트롤 플레인이다.

• etcd : 쿠버네티스 관련 메타정보를 저장하는 데이터베이스
• kube-apiserver : 컨트롤 플레인의 프론트엔드 서버로 API요청을 처리
• kube-scheduler : 생선된 파드를 실행할 노드를 결정
• kube-controller-manager : 디플로이먼트와 같은 리소스 컨트롤러를 관리
• cloud-controller-manager : 클라우드 기반 클러스터의 로드밸런서나 디스트볼륨 자원을 관리

2. 노드 컴포넌트 

• kubelet : 노드에 예약된 워크로드를 실행하기 위해 컨테이너 런타임관리, 모니터링
• kube-proxy : 서로 다른 노드에 있는 파드간 통신, 네트워크 트래픽을 라우팅

3. 고가용성

• etcd 데이터베이스는 여러 노드에 걸쳐 복제되며, 절반이상이 사용가능하면 복구가능
• 컨트롤 플레인 장애시 App이 다운되진 않지만, 새로운 컨터에너 또는 리소스를 변경할 수 없으므로 가장 빠르게 복구해야 함
• 고가용성을 위해 상용시 최소 3개이상의 클러스터가 필요
• 워커노드장애시 쿠버네티스가 해당 노드의 파드들을 다른 노드로 재조정

[ Kubernetes ] 도커를 이용한 쿠버네티스 기본예제

[ 도커설치 ] 

sudo yum -y update

sudo yum -y install docker

docker version

sudo yum -y install git 

# demo 다운로드

git clone https://github.com/cloudnativedevops/demo

# 도커실행

sudo systemctl status docker

sudo systemctl start docker

# demo 실행

sudo docker container run -p 9999:8888 --name hello cloudnatived/demo:hello

*웹 브라우저 http://localhost:9999 로 들어가

Hello, 世界 글귀를 확인

[ 도커이미지 빌드 ]

sudo docker image build -t myhello .

sudo docker container run -p 9999:8888 myhello

# 도커 컨테이너 레지스트리

https://hub.docker.com/ 에서 도커계정생성

# 도커로그인

sudo docker login

...

Username : user_id

Password : ****

...

Login Succeeded

# 도커 이미지 푸쉬

sudo docker image tag myhello yourID/myhello

sudo docker image push yourID/myhello

# 인터넷이 연결된 곳이라면 어디서든 내 컨테이너 이미지 사용가능

sudo docker container run -p 9999:8888 yourID/myhello

[ 쿠버네티스 설치 ]

단일클러스터에서 실행이기 때문에 minikube 설치

curl -Lo minikube https://storage.googleapis.com/minikube/releases/latest/minikube-linux-amd64 && chmod +x minikube

sudo mkdir -p /usr/local/bin/

sudo install minikube /usr/local/bin/

* 아래와 같은 에러가 난다면

  - docker: Not healthy: "docker version --format {{.Server.Os}}-{{.Server.Version}}" exit status 1: Got permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock: Get "http://%2Fvar%2Frun%2Fdocker.sock/v1.24/version": dial unix /var/run/docker.sock: connect: permission denied

sudo groupadd docker

sudo usermod -aG docker $USER

newgrp docker 

참조 : https://docs.docker.com/engine/install/linux-postinstall/

minikube status

만약 클러스터가 실행 중이면, minikube status 의 출력은 다음과 유사해야 한다.

host: Running
kubelet: Running
apiserver: Running
kubeconfig: Configured

정지시 

minikube stop

[ 쿠버네티스로 데모 애플리케이션 실행 ]

kubectl run demo --image=yourID/myhello --port=9999 --labels app=demo

kubectl port-forward pod/demo 9999:8888

-> 웹 브라우저 http://localhost:9999 확인

kubectl get pods --selector app=demo
NAME   READY   STATUS    RESTARTS   AGE
demo   1/1     Running   0          6m9s

# Pod삭제시

kubectl delete pod 1234-56-7890-234234-456456

# Delete all pods

kubectl delete pods --all

# 포트포워딩 후 http://yourip:9999 브라우저에서 output확인

kubectl port-forward yourDir/demo 9999:8888

[ Algorithm ] 폰켓몬 - Array to Set

[ 문제 ] 

https://programmers.co.kr/learn/courses/30/lessons/1845

* 제한사항

nums는 폰켓몬의 종류 번호가 담긴 1차원 배열입니다.

nums의 길이(N)는 1 이상 10,000 이하의 자연수이며, 항상 짝수로 주어집니다.

폰켓몬의 종류 번호는 1 이상 200,000 이하의 자연수로 나타냅니다.

가장 많은 종류의 폰켓몬을 선택하는 방법이 여러 가지인 경우에도, 선택할 수 있는 폰켓몬 종류 개수의 최댓값 하나만 return 하면 됩니다.

*입출력 예

nums result

[3,1,2,3] 2

[3,3,3,2,2,4] 3

[3,3,3,2,2,2] 2

[ 제출 ]

import java.util.*;

class Solution {
    public int solution(int[] nums) {

        int answer = 0;
        Set<Integer> p = new HashSet<Integer>();

        for(int e : nums)  p.add(e);
        if(p.size() > nums.length/2)
            answer = nums.length/2;
        else
            answer = p.size();

        return answer;
    }
}

[ 풀이 ]

nums의 int형 배열을 set으로 자료형변환을 통해 중복을 제거한다.

최대값은 N/2이므로 set에 담긴 요소의 갯수가 더 크다면 N/2를 반환한다.

[ 다른사람풀이 ]

import java.util.Arrays;

import java.util.stream.Collectors;

class Solution {
    public int solution(int[] nums) {

        return Arrays.stream(nums)
            .boxed()
          .collect(Collectors.collectingAndThen(Collectors.toSet(),phonekemons 

                -> Integer.min(phonekemons.size(), nums.length / 2)));

    }
}

첫번째 풀이의 기능을 자바 스트림을 이용하여 실행

코드가 좀더 간결해지고 내부반복자를 사용하기에 병렬처리가 용이하다.