MS SQL 데이터 송수신 자동화

Posted on 2024-01-24 Edited on 2024-01-26 In Infra Views:

서론

목적

진행하는 프로젝트는 다른 외부 서비스로부터 우리 서비스에서 필요로하는 데이터를 수신받아야 하는데, 해당 외부 서비스에서 지원하는 DB가 MS SQL, Oracle 두 종류로 국한되는 상황이다. 반면 우리 프로젝트에서 사용할 DB는 PostgreSQL 이다. 때문에 해당 외부 서비스로부터 데이터아 우리 서버로 옮겨줄 중계 DB가 필요한데, 이를 Naver Cloud Platform 내 클라우드 서버를 통해 수신받을 예정이다. 해당 클라우드 서버 설정은 이전에 다루었고 이번엔 수신받은 데이터를 어떻게 우리 서버로 옮길지에 대해 다루도록 하겠다.

고찰

네이버 클라우드상에 위치한 DB 서버로부터 데이터를 추출하기 위해선 다양한 방법이 있을 것 같다. 우선적으로 여러 방법들 중 어떤 방법이 가장 적절한지에 대해 고민하는 것이 선행되어야 할 필요가 있다. CSV 파일을 추출하여 Gitlab에 업로드하는 것도 고려하였으나, 사원 정보를 Gitlab과 같은 퍼블릭 저장소에 올리는 것은 적절한 방법이 아닌 것 같아 고려하지 않았다.

방법

가장 처음 생각한 것은 데이터베이스에 데이터가 INSERT 되는 것을 감지하여 일련의 작업을 수행하는 것이었다. 찾아보니 MSSQL Trigger를 통해 데이터가 INSERT 되는 것을 감지할 수는 있으나, 해당 코드는 SQL 쿼리문으로만 작성되어야 했다. 수신 DB가 MSSQL로 동일하게 구현되었을 경우는 SQL 쿼리문을 통해 데이터 복사가 가능하였으나, 본 프로젝트엔 수신 DB가 PostgreSQL로 구현될 예정이기에 해당 방법은 배제하였다.

스크립트를 통한 자동화

쉘 스크립트를 통해 데이터베이스로 쿼리를 날려 데이터 조회를 자동화할 수 있다. 데이터 추출은 원활히 동작하나, 실제 사용할 프로젝트 DB에 데이터를 전송하기 위해선 별도의 코드를 작성해야하기에 적절한 방법이 아닐 것 같아 배제하였다.

# sh get_data_from_table.sh > data.txt

CONTAINER_NAME=your_own_container_name
DB_USER=your_own_username
DB_PASS=your_own_pw
DB_NAME=master

SQL_SCRIPT=./data/sql_script.sql

# Run SQL script against MSSQL in Docker container
docker exec -it $CONTAINER_NAME /opt/mssql-tools/bin/sqlcmd \
    -S localhost -U $DB_USER -P $DB_PASS -d $DB_NAME \
    -Q "SELECT * FROM dbo.TABLENAME"

Python을 통한 자동화

파이썬 코드를 통해 데이터 추출 & 복사를 자동화할 수 있다. 해당 파이썬 코드를 크론잡을 통해 일정 주기마다 동작하도록 스케줄링하면 원활히 동작할 것으로 기대된다. 아래 코드를 통해 네이버 클라우드 서버상에 동작에 필요한 파이썬 및 필수 패키지를 설치할 수 있다.

# Python 설치
sudo apt install python3

# pip 설치
sudo apt install python3-pip

# 필수 패키지 설치
pip3 install pyodbc psycopg2

`pyodbc` 설치 과정에서 발생한 오류

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-17_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_10 26 10

`sql.h` 헤더가 존재하지 않아 문제가 발생한 모습

sql.h 헤더가 존재하지 않아 설치 과정에서 오류 발생하여 아래 명령어를 순차적으로 수행하여 해결

1 2	sudo apt-get install unixodbc-dev pip install pyodbc

`psycopg2` 설치 과정에서 발생한 오류

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-17_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_10 33 49

`psycopg2` 설치 과정에서 문제가 발생한 모습

설치 과정에서 이미지와 같은 오류가 발생하였는데, setuptools 설치 과정에서 오류가 발생한 걸로 추측된다. 아래 명령어를 통해 해결할 수 있다.

1	sudo -H pip3 install --upgrade --ignore-installed pip setuptools

https://musclebear.tistory.com/131 참조

이후 아래 코드를 통해 소스 DB로부터 타겟 DB로 데이터를 복사할 수 있다. 해당 코드는 파이썬 코드상으로 직접 DB에 접근하여 특정 쿼리를 통해 데이터를 얻어온 뒤, 이를 CSV 파일로 저장한다. 이후 타겟 DB에 접속한 다음 추출된 CSV 파일을 기반으로 데이터를 옮기는 간단한 방법이다.

# copy_data.py

import pyodbc
import psycopg2
import csv
import os

# Original DB
MSSQL_SERVER = "your_server_ip"
MSSQL_PORT = "your_server_port"
MSSQL_DATABASE = "your_server_database"
MSSQL_USERNAME = "your_server_username"
MSSQL_PASSWORD = "user_server_pw"

# Target DB
PGSQL_SERVER = "your_server_ip"
PGSQL_PORT = "your_server_port"
PGSQL_DATABASE = "your_server_database"
PGSQL_USERNAME = "your_server_username"
PGSQL_PASSWORD = "your_server_pw"

sql_query = "SELECT * FROM your_table_name"

temp_file = "temp_data.csv"

mssql_conn_string = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={MSSQL_SERVER},{MSSQL_PORT};DATABASE={MSSQL_DATABASE};UID={MSSQL_USERNAME};PWD={MSSQL_PASSWORD}'
mssql_conn = pyodbc.connect(mssql_conn_string)
cursor = mssql_conn.cursor()

with open(temp_file, 'w', newline='', encoding='utf-8') as csvfile:
    csv_writer = csv.writer(csvfile)
    cursor.execute(sql_query)

    for row in cursor:
        csv_writer.writerow(row)

cursor.close()
mssql_conn.close()

pgsql_conn = psycopg2.connect(
    host=PGSQL_SERVER,
    port=PGSQL_PORT,
    database=PGSQL_DATABASE,
    user=PGSQL_USERNAME,
    password=PGSQL_PASSWORD
)
pgsql_cursor = pgsql_conn.cursor()

with open(temp_file, 'r', encoding='utf-8') as f:
    pgsql_cursor.copy_expert("COPY FROM STDIN WITH CSV HEADER", f)

pgsql_conn.commit()
pgsql_cursor.close()
pgsql_conn.close()

os.remove(temp_file)

상기 코드는 소스 데이터 테이블로부터 모든 데이터를 조회하여 전송하는 코드로, 실제 사용할 때에는 테이블의 전체 데이터를 조회하는 만큼 오버헤드가 발생할 것으로 생각된다. 따라서 상기 코드를 일정 간격으로 스케줄링 하여 해당 스케줄링 시간 간격만큼의 데이터만 조회하여 타겟 DB로 업데이트 하는 것이 적절한 방법일 것 같다.

아래는 매일 파이썬 함수를 실행한다고 할 때 datetime 모듈을 통해 오늘 날짜를 조회하여 쿼리문을 요청하는 예시이다.

# 오늘 날짜를 기준으로 한 쿼리문
import datetime


(코드 중략)


current_date = datetime.datetime.now().strftime("%Y%m%d")
sql_query = f"SELECT * FROM your_table_name WHERE column LIKE '{current_date}%'"

crontab을 통해 해당 파이썬 코드 수행을 스케쥴링하여 매일 아침 7시에 수행될 수 있도록 자동화하였다. 또한 추후 확인을 위해 출력값을 cron.log 파일에 기록하여 확인할 수 있도록 하였다.

# crontab 규칙
* * * * *  유저 이름 명령어
┬ ┬ ┬ ┬ ┬
│ │ │ │ └─ 요일 (0 - 6) (0:일요일, 1:월요일, 2:화요일, …, 6:토요일)
│ │ │ └─ 월 (1 - 12)
│ │ └─일 (1 - 31)
│ └─ 시 (0 - 23)
└─ 분 (0 - 59)

# crontab 작업 추가
# 매일 아침 7시에 copy_data_to_pgsql.py를 수행
$ crontab -e
0 7 * * * /usr/bin/python3 /root/bin/copy_data_to_psql.py >> /root/logs/cron.log 2>&1

# crontab 조회
$ crontab -l
0 7 * * * /usr/bin/python3 /root/bin/copy_data_to_psql.py >> /root/logs/cron.log 2>&1

오류 로그 기록을 위해 try-except 및 다른 기능들을 적용한 최종 코드는 아래와 같다. 본 코드에서는 테스트를 위해 Naver Cloud Server와 AWS EC2 Instance를 활용하였다.

import datetime
import pyodbc
import psycopg2
import csv
import os

# Original DB
MSSQL_SERVER = "your_server_ip"
MSSQL_PORT = "your_server_port"
MSSQL_DATABASE = "your_server_database"
MSSQL_USERNAME = "your_server_username"
MSSQL_PASSWORD = "user_server_pw"

# Target DB
PGSQL_SERVER = "your_server_ip"
PGSQL_PORT = "your_server_port"
PGSQL_DATABASE = "your_server_database"
PGSQL_USERNAME = "your_server_username"
PGSQL_PASSWORD = "your_server_pw"

current_date = datetime.datetime.now().strftime("%Y%m%d")

log_directory = "/root/logs"
temp_file = os.path.join(log_directory, f"{current_date}_data.csv")

# 로그 디렉토리 생성
if not os.path.exists(log_directory):
    os.makedirs(log_directory)

try:
    print(f"{current_date} 일자 데이터 전송 시작\n")

    sql_query = f"SELECT * FROM your_table_name WHERE column LIKE '{current_date}%'"
    mssql_conn_string = f'DRIVER={{ODBC Driver 17 for SQL Server}};SERVER={MSSQL_SERVER},{MSSQL_PORT};DATABASE={MSSQL_DATABASE};UID={MSSQL_USERNAME};PWD={MSSQL_PASSWORD}'
    mssql_conn = pyodbc.connect(mssql_conn_string)

    cursor = mssql_conn.cursor()
    cursor.execute(sql_query)

	# Table Column header 추출
    columns = [column[0] for column in cursor.description]

    with open(temp_file, 'w', newline='', encoding='utf-8') as csvfile:
        csv_writer = csv.writer(csvfile)
        csv_writer.writerow(columns)

        for row in cursor:
            csv_writer.writerow(row)

    cursor.close()
    mssql_conn.close()

    pgsql_conn = psycopg2.connect(
        host=PGSQL_SERVER,
        port=PGSQL_PORT,
        database=PGSQL_DATABASE,
        user=PGSQL_USERNAME,
        password=PGSQL_PASSWORD
    )
    pgsql_cursor = pgsql_conn.cursor()

    with open(temp_file, 'r', encoding='utf-8') as f:
        pgsql_cursor.copy_expert("COPY your_table_name FROM STDIN WITH CSV HEADER", f)

    pgsql_conn.commit()
    pgsql_cursor.close()
    pgsql_conn.close()

    print(f"{current_date} 일자 데이터 전송 완료\n")
except Exception as e:
    print(f"{current_date} 일자 작업 중 오류가 발생하였습니다 : {e}")

# /root/logs/cron.log
20240117 일자 데이터 전송 시작
20240117 일자 데이터 전송 완료

# /root/logs/20240117_data.csv
your_table_data

# PostgreSQL 테이블 생성에 사용한 쿼리문

> CREATE TABLE your_table_name (

    ...

    PRIMARY KEY (PK1, PK2)
);

서버 자동 시작 & 중지

상기 과정을 완료하였다면 네이버 클라우드 DB 상의 데이터를 일정 시간마다 자동으로 전송하는 환경 구성이 완료되었다. 해당 클라우드 서버는 다른 서비스로부터 일정 시간마다 데이터를 수신받는데, 이렇게 되면 해당 수신 서버를 항상 활성화해 놓는 것은 비용적인 측면에서 낭비일 수 있다. 따라서 이번에는 Naver Cloud Functions를 활용해 자동으로 해당 서버를 일정 시간마다 시작 & 중지 할 수 있는 환경을 구성하려 한다.

Naver Cloud Functions 이란

Untitled

Cloud Functions 상세 기능

상기 이미지는 Naver Cloud Functions 에 대한 기능을 간단히 도식화한 이미지이다. 간단히 설명하면 특정 이벤트 트리거를 통해 우리가 Python, JAVA등의 언어로 정의한 비즈니스 로직을 실행시켜 네이버 클라우드 플랫폼(NCP)에서 제공하는 다른 서비스들을 제어할 수 있는 것이다. 본 글에서는 cron(이벤트 트리거)를 통해 Python(비즈니스 로직) 코드를 실행시켜 직접 Server(서비스)를 제어하고자 한다.

설정

우선적으로 트리거(Trigger)를 설정해주어야 한다. 말 그대로 특정 이벤트 발생을 감지하여 일련의 코드를 동작시키기 위함으로, 본 프로젝트에서는 cron 을 트리거로 사용할 예정이다. Naver Cloud Functions에서 Trigger 탭 내의 ‘생성’ 버튼을 누르면 다음과 같은 화면을 볼 수 있다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-18_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_10 57 46

Cloud Functions cron 생성 페이지

cron 작성 규칙은 위에서도 언급하였지만 다시 말하면 아래와 같다. 본 프로젝트에서는 7-9, 11-13, 15-17, 19-21 시간에만 서버를 활성화할 예정이다. 또한 주말에는 서버를 활성화할 필요가 없으므로 최종적으로 다음과 같이 2개의 cron Trigger 를 생성하였다.

# crontab 규칙
* * * * *  유저 이름 명령어
┬ ┬ ┬ ┬ ┬
│ │ │ │ └─ 요일 (0 - 6) (0:일요일, 1:월요일, 2:화요일, …, 6:토요일)
│ │ │ └─ 월 (1 - 12)
│ │ └─일 (1 - 31)
│ └─ 시 (0 - 23)
└─ 분 (0 - 59)

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-18_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_11 03 11

server-start-cron 설정 이미지 내 시간이 잘못되었는데, 7-21이다

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-18_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_11 04 07

server-stop-cron 설정

트리거 설정을 완료하였다면 이를 기반으로 적절한 액션(Action)을 설정해야한다. 액션은 네이버 클라우드에서 제공하는 API들을 파이썬 코드를 통해 호출하여 서버를 제어하기 위함이다. 자세한 API는 아래 링크에서 확인할 수 있다

https://api.ncloud-docs.com/

본 글에서는 서버 시작, 정지를 위해 startServerInstances, stopServerInstances 두 개의 API를 호출할 예정이다. 이를 위해 바로 액션을 설정하기에 앞서 패키지(Package)를 통해 두 API에서 공통적으로 필요한 데이터들을 관리하고자 한다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-18_%E1%84%8B%E1%85%A9%E1%84%8C%E1%85%A5%E1%86%AB_11 09 08

Cloud Functions Package 생성 페이지

패키지 이름은 자유롭게 설정하면 되며 본 프로젝트에서는 server-control 로 정의하였다. 디폴트 파라미터로는 아래와 같이 설정해주었다.

{
	"API_URL": "https://ncloud.apigw.ntruss.com",
	"ACCESS_KEY":"<클라우드 계정 ACCESS_KEY>",
	"SECRET_KEY":"<클라우드 계정 SECRET_KEY>",
	"SERVER_LIST":"?serverInstanceNoList.1=<본인의 서버 Instance ID>"
}

API_URL을 제외한 나머지 값들의 경우 사람들마다 다르니 키 값의 경우 계정 관리 > 인증키 관리 > API 인증키 관리를 기반으로 값을 할당해주면 된다. SERVER_LIST는 아래 이미지처럼 본인이 생성한 서버 정보를 통해 확인할 수 있다.

296932656-a56631ac-d0c9-450b-832a-248cee0aa14a

Cloud Server 상세 정보

이후 이를 기반으로 실제 수행할 코드를 담은 Action을 생성하고자 한다. 마찬가지로 Action또한 cron Trigger와 마찬가지로 서버 시작, 중지를 위해 2개를 필요로 한다.

서버 시작용 Action

트리거 종류 : cron
트리거 이름 : server-start-cron (개인마다 상이할 수 있음)
패키지 : server-control (개인마다 상이할 수 있음)
소스코드 : python:3.7
디폴트 파라미터 : {"API_URI": "/server/v2/startServerInstances"}

import sys
import os
import hashlib
import hmac
import base64
import requests
import time
import json

def main(args):
	# 패키지 파라미터
    access_key = args["ACCESS_KEY"]
    secret_key = args["SECRET_KEY"]
    api_server = args["API_URL"]
    api_uri = args["API_URI"]
    server_list = args["SERVER_LIST"]

	# 인증키 생성 시작 ===================
    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    secret_key = bytes(secret_key, 'UTF-8')

    method = "GET"
    uri = api_uri + server_list
        
    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())
	# 인증키 생성 종료 ===================

    http_header = {
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
    }

    response = requests.get(api_server + uri, headers=http_header)
    return {"message": "서버 시작 완료"}

서버 중지용 Action

트리거 종류 : cron
트리거 이름 : server-stop-cron (개인마다 상이할 수 있음)
패키지 : server-control (개인마다 상이할 수 있음)
소스코드 : python:3.7
디폴트 파라미터 : {"API_URI": "/server/v2/stopServerInstances"}

import sys
import os
import hashlib
import hmac
import base64
import requests
import time
import json

def main(args):
	# 패키지 파라미터
    access_key = args["ACCESS_KEY"]
    secret_key = args["SECRET_KEY"]
    api_server = args["API_URL"]
    api_uri = args["API_URI"]
    server_list = args["SERVER_LIST"]

	# 인증키 생성 시작 ===================
    timestamp = int(time.time() * 1000)
    timestamp = str(timestamp)

    secret_key = bytes(secret_key, 'UTF-8')

    method = "GET"
    uri = api_uri + server_list
        
    message = method + " " + uri + "\n" + timestamp + "\n" + access_key
    message = bytes(message, 'UTF-8')
    signingKey = base64.b64encode(hmac.new(secret_key, message, digestmod=hashlib.sha256).digest())
	# 인증키 생성 종료 ===================

    http_header = {
        'x-ncp-apigw-signature-v2': signingKey,
        'x-ncp-apigw-timestamp': timestamp,
        'x-ncp-iam-access-key': access_key
    }

    response = requests.get(api_server + uri, headers=http_header)
    return {"message": "서버 종료 완료"}

두 코드 모두 기본적으론 동일하나 사용된 디폴트 파라미터의 API_URI가 다름에 주의할 필요가 있다. 참고한 공식 문서는 다음과 같다.

https://api.ncloud-docs.com/docs/common-ncpapi
https://api.ncloud-docs.com/docs/compute-server
https://api.ncloud-docs.com/docs/compute-server-startserverinstances
https://api.ncloud-docs.com/docs/compute-server-stopserverinstances

이후 대시보드에서 설정한 시간마다 작업이 수행됨을 확인할 수 있다.

%E1%84%89%E1%85%B3%E1%84%8F%E1%85%B3%E1%84%85%E1%85%B5%E1%86%AB%E1%84%89%E1%85%A3%E1%86%BA_2024-01-18_%E1%84%8B%E1%85%A9%E1%84%92%E1%85%AE_12 48 38

서버 중지 작업이 성공적으로 수행된 모습

도커 자동실행

서버가 재부팅되면서 내부적으로 동작하고 있던 MS SQL 도커 컨테이너가 중단됨을 확인할 수 있다. 때문에 완전 자동화를 하기 위해선 재부팅할 때 도커 컨테이너를 자동으로 시작하도록 할 필요가 있다.

이전에 작성해 둔 스크립트를 crontab 에 등록하여 서버가 부팅될 시 자동으로 스크립트를 수행하여 도커 컨테이너를 실행하도록 구현하였다. 다만 이전 스크립트를 그대로 사용하기엔 docker-compose 환경변수나 실행 디렉토리 경로 문제로 인해 조금 수정하였다.

# restart_docker.sh
cd /root/bin/
/usr/local/bin/docker-compose -p incoroutine down
/usr/local/bin/docker-compose -p incoroutine up -d

1
2
3

# crontab에 작업 등록
> crontabe -e
@reboot /root/bin/restart_docker.sh >> /root/logs/reboot_log.log 2>&1

이후 서버를 재시작하면 정상적으로 도커 컨테이너가 실행됨을 확인할 수 있다.

참고 링크

Ncloud API

stopServerInstances

CLOUD FUNCTION으로 원하는 시간대에 서버를 시작하고 중지하자

네이버클라우드플랫폼 사용시 Signature Key얻을 때!

The action didn’t produce the valid JSON object in python file