데이터 엔지니어링

1. 문제 소개

• 주어진 배열에서 입력값이 삽입될 적합한 위치를 출력하면 된다.
• leetcode 난이도 easy 수준의 문제다.
• 이분 탐색을 이용하면 O(logN) 으로 해결할 수 있다.

2. 코드

class Solution:
    def searchInsert(self, nums: List[int], target: int) -> int:
        start = 0 
        end = len(nums)-1
        mid = 0  
        while(start <= end ):
            mid = ( start + end ) //2 
            if nums[mid]==target:
                return mid
            elif nums[mid] <= target:  # search right partitiion
                start = mid+1
            else:               # search left partition
                end = mid-1

        if nums[mid] < target:
            return mid+1
        else:
            return mid

3. 코멘트

• 입력 케이스가 배열에 존재하지 않는 경우에는 가장 근접한 위치를 찾게 되는데, 정답을 출력할 때는 적합한 위치가 되도록 조건문을 이용해 보정했다.
• 개인 일정으로 알고리즘 스터디에 계속 불참 하고 있는데, 다음 주에는 꼭 참여해야겠다....

1. 문제 소개

• 주어진 배열에서 최대 부분합을 구하는 문제다.
• leetcode 난이도 easy 수준의 문제다.
• 카데인 알고리즘을 이용하면 O(N)으로 해결할 수 있다.

2. 코드

class Solution:
    # Brute Force - 
    def get_maximum_sum(self,arr:List[int]):
        start = 0 
        end = len(arr)
        maximum_sum=arr[start]
        while(start<end):
            mid=start+1
            pre_sum=sum(arr[start:mid])
            while(mid<end):
                mid+=1
                if pre_sum < sum(arr[start:mid]):
                    pre_sum = 	sum(arr[start:mid])	
            if maximum_sum < pre_sum:
                maximum_sum=pre_sum
            start+=1

        return maximum_sum
    
    def kadane_algorithm(self,arr:List[int]):
        start = 0
        end = len(arr)
        maximum_sub_array = arr[start]
        start=0
        current_sum = 0
        while(start < end):
            current_sum = max(arr[start],current_sum+arr[start])
            if current_sum > maximum_sub_array:
                maximum_sub_array = current_sum
            start+=1

        return maximum_sub_array


    def maxSubArray(self, nums: List[int]) -> int:
        
        return self.kadane_algorithm(nums)

3. 코멘트

• 카데인 알고리즘은 각 최대 부분합을 구할 때 이전 최대 부분합을 이용해서 구하는 것이 핵심이다.
  • 흐름을 텍스트로 설명하자면  
  •  subArray_1 = new element
  •  subArray_2 = subArray_1 + new element
  •  subArray_3 = subArray_2 + new element
  • .. 쭉 배열의 끝까지..
• 각각의 subArray (최대부분합)을 구할 때는 (이전 최대부분합+현재인덱스)과 현재 인덱스를 비교해서 큰 값을 구한다.  

1. 문제 소개

• 그리디 문제에 해당한다.
  
  

https://www.acmicpc.net/problem/1449

2. 코드

def set_test_case():
    N,L = input().split(" ")
    N,L = int(N),int(L)
    leak_pos = []
    leak_pos_org = input().split(" ")
    for v in leak_pos_org:
        leak_pos += [int(v)]

    return N,L,sorted(leak_pos)
         
def solution(N,L,leak_pos):
    cnt = 1 
    l = L
    
    start = 0 
    end = start+1

    while(end < N):
        if (leak_pos[end]-leak_pos[start]) >= L :
            cnt+=1
            start = end 
            end = start +1
        else:
            end+=1
        
    print(cnt)  

    

def main():
    N,L,leak_pos = set_test_case()
    solution(N,L,leak_pos)


if __name__=='__main__':
    main()

3.코멘트 

• 자연수 범위에는 0이 들어가지 않는데, 0의 입력케이스 까지 생각했다가 코드를 다시 수정했다..

1. 문제 소개

• 배열, 해시 테이블을 이용해야 하는 문제에 속한다.
• leetcode 에서도 난이도 easy 수준의 문제다.
• 최근 블라인드에 이 문제에 대한 이야기가 인기글로 올라와서, 풀어봤다. 

2. 코드

#Runtime: 60 ms, faster than 83.47% of Python3 online submissions for Two Sum.
#Memory Usage: 15.3 MB, less than 42.38% of Python3 online submissions for Two Sum.


class Solution:


    # Brute Force - 가장 단순한 접근 -> time complexity : (n2), space complexity : o(1)
    def twoSum(self, nums,target):

        target_indices = []

        for i in range(0,len(nums)):
            for j in range(i+1,len(nums)):
                if nums[i] + nums[j] == target:
                    target_indices+=[i,j]
                    return target_indices

        return target_indices


    # Two - pass Hash Table -> time complexity : (n), space complexity : o(n)
    def twoSum(self, nums,target):
        hash_map = {}
        target_indices = []
        # 자료구조를 해시 맵으로 생성
        for idx,va in enumerate(nums):
            hash_map[va]=idx
        
        # look up 
        for idx,va in enumerate(nums):
            minus_value = target - va
            if minus_value in hash_map and hash_map[minus_value] != idx:
                target_indices+=[idx,hash_map[minus_value]]        
                break
        
        return target_indices


    # One pass hash Table -> time complexity : (n), space complexity : o(n) 
    def twoSum(self, nums,target):
        hash_map={}

        for idx,va in enumerate(nums):
            minus_value = target-va
            if minus_value in hash_map:
                return [idx,hash_map[minus_value]]
            
            hash_map[va]=idx

    def test(self):
        nums = [2,1,94,3,-2]
        target = 92
        result = self.twoSum(nums,target)
        print(result)



if __name__ == '__main__':
    s = Solution()
    s.test()

3.코멘트

• 난 첫 번째 Brute Force 라고 부르는 방식으로 문제를 풀었다.  (심플한 방법을 제일 좋아하기도 하고, 저 방법이 가장 먼저 떠올랐다. )
• leetcode는 문제 제출에 대한 피드백이 비교적 잘 되있다. (시각화나, 해설, discussion도 있고..)
• 첫 번째 방식보다, 해시 기반의 자료구조를 이용했을때 시간복잡도가 매우 낮았다.
• 해시 맵을 만드는 방법도 생각을 안 한건 아니었지만, 해시 맵을 만든다고 하더라도, nums 배열의 full scan 비용  * 해시 맵 안에서 key 를 scan 비용을 더 하면 O(N2) 이 될거라고 생각했는데 아니었다. (해시 맵안에서는 키를 순회하는게 아니라 맵에 키가 있는지만 보면 되니까.. -_-; ) 
• Hash를 이용하면 Lookup 시간을 O(n) 에서 O(1)로 줄일 수 있다. 

1. 문제 소개

• 구현 문제에 해당한다.
• 외부 알고리즘 기법을 적용해야 하는 수준의 문제는 아니지만, 인덱스 범위를 초과하는 반례 케이스들을 생각해야 한다.

https://www.acmicpc.net/problem/124

2. 코드

# -*- encoding: utf-8 -*-
#import sys
#input = sys.stdin.readline

def set_test_case():
    commands = []
    switch_size= int(input())
    switch = input().split(' ')   
    switch = [int(x) for x in switch] 

    command_num = int(input())
    for i in range(0,command_num):
        command = input().split(' ')
        #command = map(lambda x: int(x),command)
        commands+=[[int(x) for x in command]]

    return switch,commands


def switch_process(switch,gender,switch_pos):
    switch=switch
    switch_interval = switch_pos
    if switch_pos == 0 :
        return switch

    # 남자의 경우 입력처리
    if gender == 1:
        while(switch_pos > 0 and switch_pos <= len(switch)):

            switch[switch_pos-1] = switch[switch_pos-1] ^ 1  # XOR 연산, index -1 처리
            switch_pos+=switch_interval   # 배수 처리 

    # 여자의 경우 입력처리
    elif gender == 2:
        symmetry = True     #대칭 여부의 flag 변수
        l =switch_pos -2  # 대칭점 기준 왼쪽 포인터
        r = switch_pos  # 대칭점 기준 오른쪽 포인터
        
        switch[switch_pos-1] = switch[switch_pos-1] ^ 1  # 대칭 여부와 상관없이, 대칭점의 스위치는 전환

        while(symmetry and l >= 0 and r < len(switch)): # 대칭되는 스위치를 찾는 범위가 리스트의 범위를 벗어나지 않는 조건

            if switch[l] == switch[r]:
                switch[l] = switch[l] ^ 1
                switch[r] = switch[r] ^ 1
                l -= 1
                r += 1    
            else:
                # 대칭되지 않는 조건
                symmetry=False

    # 입력 케이스 예외처리 부분
    else:
        #exception
        pass
    return switch

def solution(switch,commnads):
    for command in commnads:
        gender,pos = command[0],command[1]
        switch = switch_process(switch,gender,pos)
    
    # 최대 출력 스위치 조건을 만족하기 위한 처리
    num_by_line = 20
    for i in range(0,len(switch),num_by_line):
        print(*switch[i:i+num_by_line],sep=' ')


def main():
    switch,commands = set_test_case()
    solution(switch,commands)

if __name__ == '__main__':
    main()

3. 코멘트

• 구현은 어렵지 않았지만, 반례 케이스를 너무 가볍게 생각하여 2차례 실패가 있었다.
• 처리 시간과 메모리 공간, 그리고 반례 케이스에 대한 고민을 깊게 하는 습관이 필요하다.

1. Pyspark?

파이썬으로 스파크를 사용하기 위한 목적으로 만들어진 인터페이스이다.

파이썬 API 를 이용할 수 있으면서, 스파크 애플리케이션을 작성할 수 있다. 

interactive 하게 pyspark shell 도 제공한다.

Spark SQL, DataFrame, streaming , MLlib 그리고 Spark Core 를 지원한다.

2. 왜 PySPark?

java, scala 로 작성된 코드가 성능이 조금 더 뛰어나다고 알려져 있다. 

하지만 빅데이터 진영에서는 파이썬이 친숙한 사람이 많고, 생산성이 더 난다고 생각한다.

3. Pyspark 기초 -  Dataframe 생성 및 다루기

** 개발할 때 비교적 자주 사용되는 함수만 다룰 것 이다.

- 조건 조회와 출력, 그리고 withColumn 은 정말 많이 사용되는 것 같다.

from pyspark.sql.types import *
from pyspark.sql.functions import col,lit

# 멤버 컬럼
member_columns = ['group_cd','join_dt','name','login_cnt']

# 멤버 데이터
member_data = [
    ('k-00001','2021-07-31','name1',10),
    ('k-00002','2021-06-31','name2',10),
    ('k-00003','2021-08-31','name3',10),
    ('k-00004','2021-03-31','name4',12),
    ('k-00001','2021-04-31','name5',11),
    ('k-00002','2021-05-31','name6',15),
]

# 데이터 프레임 생성
member_df = spark.createDataFrame(data=member_data,schema=member_columns)

#스키마 구조 출력 
member_df.printSchema()

# withColumn  - 새 컬럼 추가. literal 로 초기화해줘야 한다.
member_df= member_df.withColumn("use_flag",lit(False))

member_df.printSchema()

#pyspark 에서 countDistinct 사용을 위해서는 lib 를 선언 해줘야 하다니..
from pyspark.sql.functions import countDistinct
from pyspark.sql import functions as F


# 행 출력
member_df.show(10,truncate = False)

# 특정 컬럼 출력
member_df.select('join_dt','name','group_cd').show(10)

# groupby 출력
member_df.groupBy("group_cd").count().sort("count",ascending=True).show()

# filter 조건 출력 
member_df.filter(member_df.login_cnt > 10).show()

# filter 조건을 복수 개로 줘보자
member_df.filter(
    (F.col("group_cd") == "k-00001") &
    (F.col("login_cnt") > 10)
).count()


# where 조건 출력
member_df.where(my_df.acntid=='name1').show()

member_df.filter("name == 'name1'").show()

1.zeppelin 기반 로컬 개발 환경 구성이란?

spark 를 공부할 때, zeppelin 이라는 노트북을 이용하면 편하다.

python 에서 주피터와 비슷한 역할의 툴이다.

spark 개발을 안한지 오래되서, 다시 공부를 하기 위해 구성해보려고 한다.

요즘 데이터 엔지니어들이 보는 코딩 테스트에는 알고리즘 문제 뿐만 아니라, 스파크를 이용한 문제도 출제 된다.

역시 docker-compose 를 이용해서 구성한다.

2.왜 Docker-compose ?

오픈 소스 버전을 그대로 로컬에 받아서 설치하려면, 적지 않은 시간이 소요된다.(java 설치 lib 등등의 환경 설정. 네트워크 설정 이나 연동 등 .. )

누군가 고생해서 만들어 놓은 docker image 와 docker-compose 에 설정이 기술된 yaml 파일을 이용하여 손쉽게 올릴 수 있다. (보통 docker compose [오픈 소스 이름] 을 치면 쉽게 찾을 수 있다. )

3. zeppelin, spark 설치 

구성해보기보다는 따라하기에 가깝다. 내가 참고한 블로그를 참조 한다. 여기를 참고하는게 더 정확하니까 아래 링크로 이동하자.

https://omrisk.medium.com/apache-spark-3-playground-back-to-school-with-zeppelin-notebooks-pt-3-4ebc18da68f7

a) 설치 스크립트

# git clone 다운로드
$ https://omrisk.medium.com/apache-spark-3-playground-back-to-school-with-zeppelin-notebooks-pt-3-4ebc18da68f7

# 컨테이너 올리기
$ docker-compose -f docker-compose.yml -f docker-compose-zeppelin.yml up -d

b) 브라우저 접속

• http://localhost:9090 접속, (zeppelin) 
• 마찬가지로 spark mster GUI 에도 접근이 가능하다.
• 기본적으로 예제 코드와 스파크 앱을 띄우는 노트가 있다. ( spark server 입장에서는 zeppelin 도 하나의 appication 이기 때문에 spark 를 사용하기 위해서는 인터프리터가 구동되어야 한다.) 

** 보통 zeppelin 과 spark 를 연동 할때 spark_home ,master 엔드포인트를 잡아주는데, 아래에서 처럼 노트 안에서 잡아줄수도 있는 것 같다. 

시간이 생기면 추가로 보충하겠다... 너무 바쁘다 요즘.

1.apache superset 이란?

비지니스 인텔리젼스 웹 애플리케이션다.

에어비엔비에서 사용하고 있는 시각화 툴이다.

2. 왜 superset ?

오픈소스이면서 다양한 데이터소스를 지원한다.

UI 가 직관적이다.

에어비엔비가 쓰니까 공신력? 있어보인다.

3. spuerset 설치 (docker-compose)

how ? ( 구성 방법 )

• docker, docker-compose 가 미리 설치되어 있어야 한다.
• https://github.com/apache/superset repository 에서 컨테이너 정보와 설정이 기술된 yaml 을 받아 올린다.
• 참고한 가이드 - https://superset.apache.org/docs/installation/installing-superset-using-docker-compose

a) 설치 스크립트

• non-dev 버전으로 띄웠더니 app, worker, db, cache (redis) 프로세스가 분리되어있다.

  # yaml 소스 클론
  $ git clone https://github.com/apache/superset.git
  
  # 릴리즈 버전으로 체크아웃
  $ git checkout 1.2.0
  $ git status
  
  # 컨테이너로 올리는 명령 실행
  $ docker-compose -f docker-compose-non-dev.yml up -d​

b) 브라우저 접속

• 초기 비밀번호는 admin/admin
• http://localhost:8088 접속 (서버에 설치하신 분들은 localhost에 설치한 서버의 host 로 대체)
• 기본적으로 예제 데이터가 추가되어있는데, yaml 을 찾아보면 off 할 수 있다.

c) 데이터소스 추가 ( mysql)

• 상단 Data → Databases 클릭 → 우측상단에 + database 클릭 → DB 호스트 정보 기입 후 추가
•  https://superset.apache.org/docs/databases/mysql

차트 또는 대시보드를 구성하기 전에, dataset 을 생성해야 한다.

CSV 등의 데이터 업로드를 위해서는 database advanced 에서 권한 을 줘야 한다.

대시보드 구성은 다음 편에 ..  :D

1. lambda 란?

간단하게 서버리스로 코드를 실행할 수 있는 aws 컴퓨팅 서비스라고 할 수 있다.

2. lambda 를 왜 쓰는가?

프로그램 실행을 위한 런타임 환경이나 실행가능한 서버가 필요없다.  - 서버리스

동시성등을 고려한 개발을 프로그래머가 직접 해줄 필요가 없다 - 옵션 에서 동시성이나 병렬 제어가 가능하기 때문에

모니터링이 편하다. 

데이터 엔지니어 입장에서 aws lambda 를 통한 스트리밍 처리도 가능하다고 생각한다. (허용되는 데이터 발생량이나 규모에 따라)

3. lambda 만들기

람다는 웹 콘솔에서 GUI 로도 손쉽게 만들 수 있다.

하지만 이번 포스팅은 로컬 서버에서 aws cli 를 이용하여 lambda 함수를 만들어 본다.

실습 예제는 aws 공식 홈페이지에 나와있는 내용들이다. ( 공식 홈페이지 가이드가 더 내용이 풍부하니까 그 쪽을 먼저 보길 바란다) 

a) 사전 준비 (시간이 부족해서 생략) 

• 로컬 PC 에서 aws cli 가 설치 및 IAM 구성이 되어 있어야 한다.
• lambda 생성과 실행 등 필요한 IAM 권한이 미리 준비되어 있어야 한다 .
• 실행권한 arn 이 필요하다.

b) lambda 생성

# 워킹 디렉토리 생성 및 이동
$ mkdir lambda_python_sample 
$ cd lambda_python_sample

# 람다 코드 작성
$ vi lambda_function.py  


# 사용되는 패키지를 working 디렉토리에 다운로드
$ pip install --target ./package requests 
$ cd package

# 패키지 압축
$ zip -r ../jssvs-development-package.zip .

# lambda 코드 압축
$ zip -g jssvs-deployment-package.zip lambda_function.py  

# lambda 생성 명령어  - 함수이름, 압축 파일 경로, 이벤트 핸들러 명(function 이름) # 역할 arn 순으로.
$ aws lambda create-function --function-name lambda-jssvs-dev \
--zip-file fileb://[home dir]/WorkSpace/aws/lambda_python_sample/jssvs-deployment-package.zip \ 
--handler lambda_function.main --runtime python3.7  \
--role arn:aws:iam::[ars code]:role/gamebi-lambda-role

-- lambda 코드 

import requests
def main(event, context):
    response = requests.get("https://www.test.com/")
    print(response.text)
    return response.text
if __name__ == "__main__":
    main('', '')

c) lambda 실행

# 기본 실행 및 출력 
$ aws lambda invoke --function-name lambda-jssvs-dev out \
--log-type Tail --query 'LogResult' --output text | base64 -d

lambda 의 출력 포맷은 base64로 인코딩 되어 있기 때문에, 우리가 보려면 base64 로 디코딩 해야 한다.

d) lambda 삭제 

$ aws lambda delete-function --function-name lambda-jssvs-dev

e) lambda 리스트 조회 및 검색

# 검색 
$ aws lambda list-functions --max-items 10
# 조회
$ aws lambda get-function --function-name lambda-jssvs-dev

** lambda 기본적으로 지원할 것 같지만 지원하지 않는 라이브러리들이 (requests 등 ) 있기 때문에, 개발자가 직접 소스를 올려줘야 한다.

** 웹 콘솔에서는 레이어 라는 이름으로 외부 패키지나 라이브러리를 올려서 사용할 수 있다.

** 함수를 생성할때 파이썬 버전은 꼭 맞춰주길 바란다. (다른 언어도 마찬가지. )   

1. 문제 소개

• 구현 문제에 해당된다
• undo 명령구간에 모든 명령어는 취소하고, 남은 type 명령의 문자를 저장해서 출력하는 문제다.
• undo 명령도 undo 에 의해 취소 될 수 있다.

2.코드

#입력 테스트 케이스 처리
def setTestCase():
    N=input()
    COMMAND_LIST = []
    for i in range(int(N)):
        COMMAND_LIST.append(input())
    return COMMAND_LIST

#커맨드 라인 파싱 
def command_parser(input_command_list):
    command_list = []
    for i in input_command_list:
        split_command = i.split(" ")
        command_type,command_content,seconds = split_command[0],split_command[1],int(split_command[2])
        command_list.append((seconds,command_content))
    
    return command_list


# 입력 처리 
def process_command(command_list):
    text = ''   # 결과 문자열을 담을 변수
    rollback_point=-1   # undo 의 분기를 찾을 카운트
    
    # 명령어는 역순으로 처리
    for command_set in reversed(list(command_list)):
        seconds,command = command_set
        # undo 가 적용되는 마지막 명령어 분기
        if seconds == rollback_point:       
            rollback_point = -1 
            continue
        # undo 가 해당되는 구간의 명령어 skip 분기
        elif rollback_point != -1 and seconds > rollback_point:
            continue
        # undo 명령어를 만났을 때 분기
        elif command.isdigit():
            rollback_point = seconds - int(command)
            if rollback_point < 0:
                rollback_point = 0
            continue
        # 문자 저장
        text+=command
    
    return text[::-1]   # 역순으로 결과 문자열이 저장되었기 때문에 reverse 처리 
        


def solution():
    input_command_list = setTestCase()
    command_list=command_parser(input_command_list)
    result = process_command(command_list)

    print(result)

if __name__=='__main__':
    solution()

3.코멘트

• 스택구조의 프로그램 호출 처리 흐름을 모방하여, 명령어를 역순으로 undo 범위에 해당되는 명령은 취소하여 문제를 해결했다.  
• 생각한 테스트 케이스는 문제가 없었는데 실제 제출시 실패가 있었다.
• 함께 스터디한 스터디원에게 반례 케이스 정보를 얻어 문제를 해결할 수 있었다.
• type/undo 가 0초일때 , 그리고 같은 초에 여러 명령어의 입력이 들어올 경우를 추가하여 테스트 했다. 
• 언제나 그랬지만 함께 열심히 공부한 스터디원에게서 지식을 얻는다. .. (스터디원들에게 고맙다 )