Like Sherlock Data Scientist

문제

• 스택을 이용해 다음 연산을 지원하는 큐 자료구조를 구현하라
  • push(x) : 값 x를 큐 마지막에 삽입하라
  • pop() : 큐의 처음에 있는 값을 제거한다.
  • peek() : 큐의 처음에 있는 값을 조회한다.
  • empty() : 큐가 비어있는지 여부를 확인한다.
• 원본 url : https://leetcode.com/problems/implement-queue-using-stacks/

문제 풀이

• 스택의 연산만을 활용하기 위해서는 2개의 스택이 필요하다
• 이유는 맨 마지막 값만을 활용할 수 있기 때문에 스택의 연산으로 값을 추출하면 값이 곧 최신에 들어간 값(마지막 값)이다.
• 그런데 큐는 맨 처음에 들어간 값이 추출돼야 해서 스택의 연산으로 할 경우 이동이 필요하다.
• output 값이 아무것도 없기 전까지는 재입력하는 알고리즘이 돌아가지 않는다
• 시간복잡도는 O(1) 로 계산되게끔 구현한다
• 실행속도 : 18 ms

class MyQueue(object):

    def __init__(self):
        self.input = []
        self.output = []

    def push(self, x):
        """
        :type x: int
        :rtype: None
        """
        self.input.append(x)


    def pop(self):
        """
        :rtype: int
        """
        # input 값을 재정렬하는 과정 거친 후
        self.peek()
        # 제일 처음에 들어온 값을 추출
        return self.output.pop()

    def peek(self):
        """
        :rtype: int
        """
        # 출력값이 아무것도 없다면 입력값 확인해서 값 재정렬
        if not self.output:

            # 새로 들어온 값이 있다면
            # 해당 값은 output으로 queue 구조로 재입력 돼야 함
            while self.input:

                # 우선 input값에서 마지막 값을 빼고
                # 해당 값을 output에 집어 넣는다
                self.output.append(self.input.pop())

        # 결과값이 맨 마지막 위치한 값이 제일 처음 들어온 값으로 위에서 세팅했음
        return self.output[-1]


    def empty(self):
        """
        :rtype: bool
        """
        # 입력값과 결과값 stack을 모두 확인해야 함
        # 입력값에 값이 들어온 후 결과값에 재정령 안 돼 있을 수도 있기 때문
        return self.input == [] and self.output == []

문제

• 큐를 이용해 다음 연산을 지원하는 스택(Last-In-First-Out)을 구현
  • push(a) : 값 a를 스택에 삽입한다
  • pop() : 스택의 첫번째 요소를 삭제하고 반환한다. (제일 나중에 들어간 값이 첫번째 요소를 의미할 것이다)
  • top() : 스택의 첫번째 요소를 가져온다.
  • empty() : 스택이 비어 있는지 여부를 확인한다. (True / False)
• 원본 url : https://leetcode.com/problems/implement-stack-using-queues/

문제 풀이

• 큐는 파이썬 자료구조 중 deque를 활용한다
• 큐의 연산(맨 앞에 있는 값을 제일 먼저 가져오는)만을 사용하기 위해 스택의 LIFO를 구현해줘야 한다
• 즉, 맨 나중에 들어오는 값이 맨 앞(맨 왼쪽에 위치하게끔 바꿔야 한다.
• 실행 속도 : 39ms

class MyStack(object):

    def __init__(self):
        self.q = collections.deque()

    def push(self, x):
        """
        :type x: int
        :rtype: None
        """
        # Step1. 값을 집어 넣는다
        self.q.append(x)

        # Step2. 데이터를 재정렬한다.
        # 현재 들어 있는 값중 맨 마지막 값은 가만히 있으면 되기 때문에
        # 전체 개수에서 1개 뺀만큼만 이동하면 된다.
        for _ in range(len(self.q)-1):
            # deque 자료구조의 연산중 popleft는 결국 que의 선입선출 연산구조를 의미한다.
            self.q.append(self.q.popleft())

    def pop(self):
        """
        :rtype: int
        """
        return self.q.popleft()

    def top(self):
        """
        :rtype: int
        """
        return self.q[0]


    def empty(self):
        """
        :rtype: bool
        """
        return len(self.q)==0

문제

• 매일 화씨 온도 리스트 temperature를 입력받을 때, 더 따뜻한 날씨을 위해서는 몇일을 기다려야 하는지 출력
• 더 따뜻한 날씨가 없다면 0으로 표기
• 원본 url : https://leetcode.com/problems/daily-temperatures/

예시

Example 1:

Input: temperatures = [73,74,75,71,69,72,76,73]
Output: [1,1,4,2,1,1,0,0]

Example 2:

Input: temperatures = [30,40,50,60]
Output: [1,1,1,0]

Example 3:

Input: temperatures = [30,60,90]
Output: [1,1,0]

문제 풀이 1

• 특별한 자료구조 형태를 적용하지 않고, 단순히 Brute Force 방식으로 풀이
• O(n) 속도로 예상됨. 기본적인 for문과 if문으로 구성

class Solution(object):
    def dailyTemperatures(self, temperatures):
        """
        :type temperatures: List[int]
        :rtype: List[int]
        """
        result=[]
        for i, t in enumerate(temperatures):
            # 다음 따뜻한 날을 체크하기 위한 변수
            day=0

            # 전체 온도의 개수에서 2개를 뺀 이유는 리스트의 index를 사용하는데
            # index에서 +1 하기 때문에 마지막 index 값에서 1을 더하면 out of range error 발생 
            length=len(temperatures)-2

            # 맨 마지막 값인지 아닌지 체크
            if length >= i:
                # 다음 온도가 현재 온도 보다 높은지 체크하고 만약 낮거나 같다면 day와 i가 +1 증가하여
                # 그 다음 온도를 체크하면서 day값을 측정
                while t >= temperatures[i+1] and i < length:
                    day += 1
                    i += 1

                # 특별한 예외 상황이 있는데
                # 만약 끝까지 확인했는데 더 높은 온도가 없으면 0으로 입력해야 함
                # 이 부분이 없으면 맨 끝까지 확인하면서 증가한 day가 입력될 것으로 예상
                if i==length and t >= temperatures[i+1]:
                    result.append(0)
                else:
                    result.append(day+1)

            # 맨 마지막 값이라면 무조건 0으로 입력
            else:
                result.append(0)

        return result

문제 풀이 2

• 현재 index를 stack 자료구로로 쌓아두다가, 이전보다 상승하는 지점에서 현재 온도와 stack에 쌓아둔 index 지점의 온도 차이 비교
• 만약 더 높다면, stack의 값을 꺼내 현재 index와 비교하여 그 차이를 정답으로 입력한다.
• 즉, 현재 온도가 이전 온도들과 비교할 때 stack에 쌓여 있는 index를 활용할 것이고, 이 index의 값이 현재 온도보다 작다면 해당 index는 추출되고 그 차이가 정답으로 옮겨져서 결국, stack에는 해당 index가 안 쌓여 있을 것이다.
• 속도 : 31ms

class Solution(object):
    def dailyTemperatures(self, temperatures):
        """
        :type temperatures: List[int]
        :rtype: List[int]
        """
        # 정답 리스트 세팅 (0으로 세팅해두면, 알아서 아래 로직이 거치지 않는 곳은 0으로 세팅 됨)
        answer=[0] * len(temperatures)

        # 위 온도 값의 index를 담을 자료 구조
        stack=[]

        for i, cur in enumerate(temperatures):
            # step1. stack 값이 있는지 확인
            # step2. 현재 온도가 stack의 쌓여 있는 index의 값보다 크다면
            while stack and cur > temperatures[stack[-1]]:
                # stack에서 해당 index 추출
                last = stack.pop()
                # 해당 index 위치에 정답 입력
                # 현재 i 와 차이값으로 세팅
                answer[last]=i-last

            # 현재 온도의 index가 추가 돼야 추후 온도들과 비교 가능
            stack.append(i)

        return answer

pytorch 배경

• 페이스북에서 루아 언어라는 언어로 개발한 torch 프레임 워크를 python으로 다시 개발해서 pytorch로 개발
• 초기에는 numpy와 같이 데이터 분석 프레임워크와 같은 느낌으로 기능 제공
• 추후 GPU를 사용해서 딥러닝 병렬처리가 가능하게끔 발전

pytorch 구조

• low-level : 하위 부분은 하드웨어 입장에서 GPU 접근 등 처리 계산을 빠르게 하기 위한 구조로 개발돼 있음, C, CUDA와 같은 저수준 언어 사용
• middle-level : C++ 활용하여 중간 레벨의 언어로 개발. 엔진 역할을 담당
• top-level : Python 모듈 구조로 wrapping돼 있는 API 제공

pytorch 구성 요소

• torch : 텐서와 같은 수학 함수 기능 포함
• torch.autograd : 자동 미분 기능 제공하는 라이브러리
• torch.nn : 딥러닝 Neural Network 신경망 구성을 위한 라이브러리
• torch.multiprosessing : 병렬처리 계산 기능 제공
• torch.optim : 신경망 파라미터 최적화 알고리즘 제공
• torch.utils : 데이터 조작 기능 제공
• torch.onnx : ONNX(Open Neural Network Exchange) -> pytorch로 구현한 딥러닝 모델을 다른 프레임워크 형태의 모델로 공유할 수 있도록 도와주는 라이브러리

Tensor

• 데이터 표현하는 하나의 구조로서 다차원 데이터를 효율적으로 표현 가능
• 수치형 데이터를 담는 컨테이너 구조로 이뤄져 있음
• GPU 사용해서 tensor 계사할 때 연산 가속 기능 제공
• scalar : 0D, 차원이 없는
• vector : 1D, 1차원
• matrix : 2D, 2차원
• 그 외 다차원 tensor 확장 가능

Tensor 초기화

# pytorch import & 버전 확인
import torch
torch.__version__

# 초기화 되지 않은 tensor 생성 (3,3) 2차원 matrix
non_init_tensor = torch.empty(3,3)
print(non_init_tensor)
"""
tensor([[1.0286e-38, 1.0653e-38, 1.0194e-38],
        [8.4490e-39, 1.0469e-38, 9.3674e-39],
        [9.9184e-39, 8.7245e-39, 9.2755e-39]])
"""

# random(무작위)으로 초기화된 tensor
init_random_torch=torch.rand(4,2)

"""
tensor([[0.8102, 0.2902],
        [0.7740, 0.6910],
        [0.5866, 0.2347],
        [0.6740, 0.8726]])
"""


# 데이터 타입이 long type이면서 0으로 채워진 텐서
x = torch.zeros(4,2, dtype=torch.long)
"""
tensor([[0, 0],
        [0, 0],
        [0, 0],
        [0, 0]])
"""


# 사용자 정의 입력값 tensor 초기화
x = torch.tensor([3, 2.3])
"""
tensor([7.0000, 4.1000])
"""


# 4 x 2 크기의 double 타입으로 1이 채워진 tensor
# 기존 만들어진 tensor를 변경
x = x.new_ones(4, 2, dtype=torch.double)
"""
tensor([[1., 1.],
        [1., 1.],
        [1., 1.],
        [1., 1.]], dtype=torch.float64)
"""


# 이전 tensor의 크기로, 무작위 tensor 생성, 데이터 타입은 float
new_x = torch.randn_like(x, dtype=torch.float)
"""
tensor([[-2.3465, -2.2080],
        [-0.2785,  0.6612],
        [-1.4324,  0.8104],
        [ 1.1557, -1.0430]])
"""


# 텐서 사이즈 확인
x.size()
"""
torch.Size([4, 2])
"""

데이터 타입 확인

• 다양한 tensor 데이터 타입 제공

# floattensor로 데이터타입 지정
ft = torch.FloatTensor([3,6,2])
print(ft)
print(ft.dtype)
"""
tensor([3., 6., 2.])
torch.float32  # ==> 기본이 32bit
"""


# tensor 데이터 타입 casting
print(ft.short())
print(ft.int())
print(ft.long())
"""
tensor([3, 6, 2], dtype=torch.int16)      # ==> int 16bit
tensor([3, 6, 2], dtype=torch.int32)    # ==> int의 default도 32bit
tensor([3, 6, 2])                        # ==> 64 bit
"""


# IntTensor 확인
it = torch.IntTensor([4,6,9])
print(it)
print(it.dtype)
"""
tensor([4, 6, 9], dtype=torch.int32)
torch.int32
"""


# IntTensor를 Type Casting
print(it.float())
print(it.double())
print(it.half())
"""
tensor([4., 6., 9.])
tensor([4., 6., 9.], dtype=torch.float64)
tensor([4., 6., 9.], dtype=torch.float16)
"""


# 실제 데이터 값 확인
x = torch.randn(1)
print(x.item())
print(x.dtype)
"""
-0.4453190863132477
torch.float32
"""


# tensor 연산 device setting

# device setting
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(device)

# 위에서 만든 x와 같은 size로 1 값이 tensor 생성하고, device 부여
y = torch.ones_like(x, device=device)
print(y)

# x 또한 device 부여 (to 메서드 사용)
x = x.to(device)
print(x)

# 연산
z = x + y
print(z)

# 위 연산된 결과 값을 cpu로 이동, dtype은 double로
z.to('cpu', dtype=torch.double)
print(z)

참조 url : https://leetcode.com/problems/remove-duplicate-letters/

Example 1:

Input: s = "bcabc"
Output: "abc"

Example 2:

Input: s = "cbacdcbc"
Output: "acdb"

문제 풀이를 위한 해설 특징

• 사전식 순서의 적용은 중복된 문자에만 적용된다
• 그러다 보니 중복을 제거하고 나머지 문자에 대해 사전식으로 정렬하는 것이 아니라 중복이 아닌 문자의 위치는 그대로 있는 상황 속에서 나머지 중복문자들만 하나만 남기고 사전식으로 정렬해야 한다
• 예를 들어 bcabc 라는 문자의 중복문자를 제외하면 사전에서 제일 먼저 찾을 수 있는 순으로 나열하면 abc 일 것이다.
• 만약 e가 붙은 ebcabc라면 eabc가 남을 것이고 e가 중복으로 붙는다면 ebcabce 에서 e가 중복으로 제거 되어 abce로 남을 것이다

재귀 함수를 활용한 풀이 1

• 일반적인 재귀 함수를 활용하여 풀이
• 실행 속도 : 49 ms

스택을 활용한 풀이 2

• 현재 문자(char)가 스택에 쌓여 있는 문자(이전 문자보다 앞선 문자) and 뒤에 붙일 문자가 남아 있다면(즉, counter가 0이라면), 스택에 쌓아 놓은 걸 없앤다
• 문자별 개수를 counting하기 위해 collections.Couter() 라이브러리 사용
• 이미 처리된 문자 여부를 확인할 집합(set) 데이터도 필요하다.
• 스택 연산에서 정의한 기능 외 python 리스트 자료구조의 특성을 이용한 풀이 version
• 실행 속도 : 36 ms

스택을 활용한 풀이 3

• 스택에서 가능한 연산만을 활용해서 풀이
• seen 집합에서 검색 기능 수행

총평

• 현재 문제를 정확하게 이해하지 못했고, 특별히 코드 중에 while문으로 다음 문자를 확인하는 부분이 있는데 왜 이런 로직이 사용됐는지 정확히 미파악
• python의 리스트로 스택을 구현했으며 pop() 메서드를 통해 스택 자료구조 특징을 사용했다.

층 정규화(Layer Normalization)

긴 시퀀스를 활용해서 RNN 신경망 모델로 구현하면 불안정한 그레디언트 문제가 발생합니다. 그레이던트의 소실, 폭주 혹은 긴 훈련 시간, 불안정한 훈련 등 다양한 문제에 봉착할 수 있습니다. 이런 긴 시퀀스를 처리하기 위해서 다양한 해결방법이 있지만 그 중 층 정규화에 대해 먼저 살펴보겠습니다.

층 정규화는 RNN 모델 각 층 사이에서 feature dimention에 대해 정규화하는 방법입니다. 즉, 각 타임 스텝 사이에서 층 정규화를 통해 긴 시퀀스 데이터를 학습할 때 생기는 정보 손실과 불안정한 그레디언트를 예방할 수 있습니다. 간단하게 구현해 보겠습니다.

class LNSimpleRNNCell(tf.keras.layers.Layer):
	def __init__(self, units, activation='tanh', **kwargs):
    	super().__init__(**kwargs)
        
        # state는 이전 타임 스템의 은닉층 상태를 담은 매개변수로서
        # 하나 이상의 텐서를 담고 있다.
        
        # 각 state_size와 output_size는 unit 개수로 맞춰준다
        self.state_size = units
        self.output_size = units
        
        # SimpleRNN Cell을 구성하는데 활성화 함수가 없는 이유는
        # 이전 스텝의 결과(output)와 현재 input(state, 은닉층)을 선형 연산 후 활성화 함수 이전에 층 정규화를 수행하기 위해서힙니다.
        self.simple_rnn_cell = tf.keras.layers.SimpleRNN(units, activation=None)
        
        # 각 타임 스텝마다 층 정규화가 이뤄질 수 있도록 설정
        self.layer_norm = tf.keras.layer.LayerNormalization()
        # 앞에서 설정하지 않았던 활성화 함수를 층 정규화 이후에 세팅합니다.
        self.activation = tf.keras.activations.get(activation) # 탄젠트 함수를 activation으로 사용
        
    def call(self, inputs, states):
    	
        # SimpleRNN 셀을 사용하여 현재 입력(inputs)과 이전 은닉 상태(states)의 선형 조합을 계산합니다.
        # 여기서 출력은 은닉 상태와 동일합니다. 즉, new_stats[0] == outputs
        # 나머지 new_stats는 무시해도 괜찮음
        outputs, new_states = self.simple_rnn_cell(inputs, states)
        
        # 층 정규화와 활성화 함수 차례대로 적용
        norm_outputs = self.activation(self.layer_norm(outputs))
       
        # 출력을 두번 반환
        # 하나는 출력, 다른 하나는 새로운 은닉 상태
        return norm_outputs, [norm_outputs]

# 위 사용자 정의 Cell을 사용하려면 tf.keras.layers.RNN 층을 만들어 이 class 객체를 전달하면 됩니다.
model = tf.keras.models.Sequential([
    tf.keras.layers.RNN(LNSimpleRNNCell(20), return_sequences=True, input_shape=[None,1]),
    tf.keras.layers.RNN(LNSimpleRNNCell(20), return_sequences=True),
    tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(10))
])

위처럼 층 정규화 방법 외에도 dropout 비율 정의를 통해서도 긴 시퀀스를 다루기 위해 불안정한 그레디언트 문제를 감소할 수 있습니다. 그럼, 이제 단기 기억 문제에 대해 알아보고 이에 대한 방안인 LSTM Cell에 대해 알아보겠습니다.

LSTM

RNN 모델에서 타임 스텝이 계속 진행할수록 처음 타임 스텝에서 가져온 시퀀스 데이터의 정보들이 소멸됩니다. 즉, 단기 기억에만 머물게 되는 문제가 발생합니다. 이 문제를 해결하기 위해 장기기억을 담당하는 Cell이 개발됐습니다. 바로 LSTM입니다.

장단기 메모리 셀 즉, LSTM Cell은 훈련을 빠르게 진행하고, 시퀀스 데이터의 장기간 의존성을 유지할 것입니다. keras에서 구현은 간단합니다.

# 간단히 LSTM Cell 사용
model =tf.keras.models.Sequential([
    tf.keras.layers.LSTM(20, return_sequences=True, input_shape=[None, 1]),
    tf.keras.layers.LSTM(20, return_sequences=True),
    tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(10))
])

# RNN 층에 LSTMCell을 매개변수로 지정
model = tf.keras.models.Sequential([
     tf.keras.layers.RNN(tf.keras.layers.LSTMCell(20), return_sequences=True, inpit_shape=[None,1]),
     tf.keras.layers.RNN(tf.keras.layers.LSTMCell(20), return_sequences=True)
     tf.keras.layers.TimeDistributed(tf.keras.layers.Dense(10))
])

그런데 GPU 사용을 고려할 때, LSTMCell이 최적화 돼 있어서 일반적으로는 LSTMCell을 직접사용하고, 사용자 정의 Cell을 만들 때는 RNN layer를 사용하는게 일반적입니다.

LSTM Cell의 작동 구조는 매우 복잡합니다.

키포인트는 LSTM 네트워크에서 장기 상태로 저장할 값, 버릴 값, 읽어드릴 값을 학습하는 것입니다. 이전 장기 기억인 (Cell) state는 삭제 게이트를 지나서 일부 정보를 손실하고, 입력 게이트를 지나 선택된 정보들을 추가합니다. 이렇게 만들어진 새로운 장기 기억은 다음 장기 기억(Next Cell State)으로 전달됩니다. 그래서 각 타임 스텝 마다 일부 정보는 손실되고, 추가됩니다. 또한 이 새롭게 만들어진 장기 기억은 tanh 활성화 함수를 거쳐, 출력 게이트를 통과합니다. 이렇게 해서 만들어진 단기 상태가 Output(\(h_t\)) 값으로 전달됩니다.

간략히 정리해보면 LSTM 네트워크는 중요한 입력은 인식하고, 장기 상태를 최적의 정보로 유지하기 위해 보존하고, 필요할 때마다 이를 추출하며 학습합니다. 이런 LSTM 모델은 긴 텍스트, 오디오 등 장기 패턴을 발견하는 데 우수한 성능을 보이고 있습니다.

LSTM 변종 GRU

GRU 셀은 LSTM 셀의 변종 버전으로 구조가 간소화되고, 유사하게 작동합니다. 간소화된 내용을 확인해 보면, LSTM 셀은 새로운 장기 기억과 은닉 상태 값 2개가 전달됐습니다. GRU는 이 2개를 합쳐서 하나의 벡터 값으로 전달됩니다. 그리고 하나의 게이트 제어기가 삭제 게이트와 입력 게이트 역할을 동시에 수행합니다. 마지막으로 출력 게이트가 없습니다. 즉, 전체 타임 스텝마다 해당 state 벡터가 출력됩니다. 그래서 새로운 게이트 제어기가 만들어졌는데 각 타임 스텝에서 생성한 state 벡터 중 어떤 값을 main layer에 노출시킬지 결정하는 게이트 제어기가 있습니다.

GRU 셀은 RNN의 대표적인 성공 모델입니다. 한가지 아쉬운 점은 기존 RNN, LSTM에 비해 긴 시퀀스를 잘 다루긴 하지만, 매우 제한적인 단기 기억을 가지고 있습니다. 이 문제를 해결하기 위한 방법은 다음 포스팅에서 소개하겠습니다.