[LAMBDA]SQS를 이용한 dynamoDB bulk insert(결과: 실패)

  10개, 100개의 데이터를 DB에 한번에 넣는건 문제가 되지 않지만, 만약 30000개의 데이터가 한 번에 들어가면 어떻게 될까? 시간이 엄청 오래 걸릴 뿐더러 데이터 소실이 발생할 수 있다. 이런 경우에 30000개의 데이터를 100개의 프로세스가 300개씩 나눠서 작업한다면 한층 수월할 것이다.

  하지만 python은 기본적으로 GIL을 따르기 때문에 병렬처리가 안된다. 이 때 다중스레드, 다중프로세스 모듈을 임포트하는 방법을 사용하면 된다. 위 링크로 들어가면 자세한 내용을 볼 수 있다.

  그럼 lambda에서 빅데이터를 받아 SQS로 보내고, 그 데이터를 DB에 저장하려면 어떻게 해야할까?우선 3개의 프로세스가 데이터를 10개씩 2번 DB에 저장하는 작업을 시도해봤다.

✅ 외부 python 파일 생성

from multiprocessing import Process, Queue
import requests
import time
import json

def post(num, url, q):
    total = 0
    headers = {'Content-Type' : 'application/json'}
    data = {
    }

    for _ in range(num):
        total += 1
        requests.post(url, headers=headers, data=json.dumps(data))

    q.put(total)

    print(f"Result : {total}")

if __name__ == "__main__":
    start_time = time.time()
    q = Queue()

    for i in range(3):
        p = Process(target=post, args=(2, 'https://hrylzqvpa2.execute-api.ap-northeast-2.amazonaws.com/post/ikaria', q))
        processes.append(p)
        
    p.start()
    p.join()

    q.put('exit')
    total = 0

    while True:
        tmp = q.get()
        if tmp == "exit":
            break
        else:
            total += tmp

    print(f'Total: {total}')
    print(f"End : {time.time() - start_time}")

  url 주소에 POST API 주소를 넣고 돌려본다.

그럼 60개의 데이터가 DB에 저장돼야하는데,

  ?

✅ 시도한 방법

1. 프로세스 1개

  시간이 오래걸리고, 데이터 200개쯤 부터 race condition 발생

2. 프로세스 2개 이상

  1개만 실행됨. 데이터 200개쯤 부터 race condition 발생. 당연하다, 1개만 실행됐으니까...

3. SQS 표준 대기열을 삭제하고, FIFO 방식으로 교체

  first in first out 방식이라면 race condition이 발생하기 않을까 싶었지만... 결과적으로 차이가 없다.

4. batch_writer() 사용

  dynamoDB에 데이터를 bulk로 넣을 수 있는 함수라고 해서 구글링해서 사용해본 결과...

# lambda ikaria.db
import json
import boto3

def lambda_handler(event, context):
    records = event['Records']
    
    for record in records:
        payload = json.loads(record["body"])
 
    dynamodb = boto3.resource('dynamodb')
    table = dynamodb.Table('ikaria_db')
    
    with table.batch_writer() as writer:
        for item in payload:
            writer.put_item(Item=item)

  race condition 발생. 효과 없다

✅ 결과: 실패

  바로 내일부터 다른 프로젝트를 시작해야해서 실패로 마무리지었다... 가장 의심되는 상황은 dynamoDB의 성능인데, 대량의 데이터를 넣었을 때 속도가 어마어마하게 느리고, 270개 이상부터는 잘 들어가지도 않는다. DB 인스턴스를 싹 다 지우니까 한 번에 100개씩 들어오고... 270개 이상부터는 또 안들어오다가 다시 비우니 들어오고... 아무래도 의심이된다...

  그리고 하나의 프로세스만 실행되는것도 구글링해보니,

왜 우리가 프로세스 기반 병렬 처리(다중 처리 모듈 사용)를 사용하고 스레드 기반(스레딩 모듈 사용)을 사용하지 않았는지 궁금할 것입니다. 간단한 대답은 이 실험을 CPU 집약적인 기능으로 제한했다는 것입니다. 스레드를 사용하면 Python의 Global Interpreter Lock 또는 GIL이 시작되어 병렬 처리에 대한 모든 노력을 방해합니다. 빠른 Google 검색에 표시되는 여러 기사에서 Python의 GIL에 대해 자세히 읽을 수 있습니다. 그러나 핵심은 Python의 GIL이 한 번에 하나의 프로세스 스레드만 Python의 인터프리터에 액세스할 수 있도록 허용한다는 것입니다.

  왜 스레드가 아닌 프로세스로 병렬처리를 하는지 검색해본거였는데 실패를 겪고 다시보니 python의 GIL 때문인가 싶다.