데이터 파이프라인 - 추출

Jiwan Ahn·2023년 7월 3일

데이터 파이프라인 학습기록

데이터 엔지니어링

목록 보기

3/4

: 매우 단순한 데이터 파이프라인 패턴 (이 구조에서 응용이 되어 다양한 파이프라인이 생성된다.)

추출 방식

전체 추출 : 원본 데이터 소스에서 추출 작업을 실행할 때마다 전체 데이터를 (레코드의 모든 테이블) 읽어와서, 데이터 웨어하우스에 저장 (원본 데이터를 대체하거나 갱신)

OrderID	OrderStatus	LastUpdated
1	BackOrdered	2020-06-01 12:00:00

증분 추출 : 추출작업을 할 때마다 데이터의 변경사항만을, 즉 변경하거나 추가된 원본 테이블의 레코드만 추출

OrderID	OrderStatus	LastUpdated
1	BackOrdered	2020-06-01 12:00:00
1	Shipped	2020-06-09 12:00:25

: 대부분의 경우에는 증분 추출을 사용하는데, 전체 추출과 관련하여 몇 가지 이점이 있기 때문이다.

전체 데이터를 대체할 필요가 없어, 실시간 스트리밍 데이터 처리에 유리하다.
성능 면에서 전체 추출보다 우세하다.
후에 설명할 이진 추출의 경우, 증분 추출 방식을 채용한다. (변경 사항 추출)

각 DB 종류별 추출

: 위 책에서 실습하는 내용은 각 프레임워크별 (MySQL, PostgreSQL, REST API 등) 데이터베이스에서 데이터를 추출한 후, S3 버킷에 적재하는 방식이다.

MySQL (PostgreSQL도 매우 유사함)

: 전체 추출과 증분 추출의 방식이 다르다.

전체 추출

import pymysql
import csv
import boto3
import configparser

parser = configparser.ConfigParser()
parser.read("pipeline.conf")
hostname = parser.get("mysql_config", "hostname")
port = parser.get("mysql_config", "port")
username = parser.get("mysql_config", "username")
dbname = parser.get("mysql_config", "database")
password = parser.get("mysql_config", "password")

conn = pymysql.connect(host=hostname,
        user=username,
        password=password,
        db=dbname,
        port=int(port))

if conn is None:
  print("Error connecting to the MySQL database")
else:
  print("MySQL connection established!")

m_query = "SELECT * FROM Orders;"
local_filename = "order_extract.csv"

m_cursor = conn.cursor()
m_cursor.execute(m_query)
results = m_cursor.fetchall()

with open(local_filename, 'w') as fp:
  csv_w = csv.writer(fp, delimiter='|')
  csv_w.writerows(results)

fp.close()
m_cursor.close()
conn.close()

# load the aws_boto_credentials values
parser = configparser.ConfigParser()
parser.read("pipeline.conf")
access_key = parser.get("aws_boto_credentials", "access_key")
secret_key = parser.get("aws_boto_credentials", "secret_key")
bucket_name = parser.get("aws_boto_credentials", "bucket_name")

s3 = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)

s3_file = local_filename

s3.upload_file(local_filename, bucket_name, s3_file)

ConfigParser로 pipeline 설정사항을 읽어온다. (pipeline.conf)

parser = configparser.ConfigParser()
parser.read("pipeline.conf") # MySQL 서버와 연결하기 위한 Configuration 설정
hostname = parser.get("mysql_config", "hostname")
port = parser.get("mysql_config", "port")
username = parser.get("mysql_config", "username")
dbname = parser.get("mysql_config", "database")
password = parser.get("mysql_config", "password")

⇒ pymysql을 통해 mysql과 연결 과정에서 필요한 정보들을 읽어온다.

Query문을 실행하여 데이터를 읽어온다.

m_query = "SELECT * FROM Orders;" #전체 추출 쿼리문
local_filename = "order_extract.csv"

m_cursor = conn.cursor()
m_cursor.execute(m_query)
results = m_cursor.fetchall()

with open(local_filename, 'w') as fp:
  csv_w = csv.writer(fp, delimiter='|') # '|' 로 열 구분
  csv_w.writerows(results)

⇒ .csv 파일에 읽어온 데이터들을 구분자 (’|’) 들을 통해 열을 구분하고 로드한다.

.csv 파일을 (데이터) S3 버킷에 로드한다.

parser = configparser.ConfigParser() #AWS의 S3 버킷과 연결하기 위해 설정
parser.read("pipeline.conf")
access_key = parser.get("aws_boto_credentials", "access_key")
secret_key = parser.get("aws_boto_credentials", "secret_key")
bucket_name = parser.get("aws_boto_credentials", "bucket_name")

s3 = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)

s3_file = local_filename

s3.upload_file(local_filename, bucket_name, s3_file)

증분 추출

import pymysql
import csv
import boto3
import configparser
import psycopg2

# get db Redshift connection info
parser = configparser.ConfigParser()
parser.read("pipeline.conf")
dbname = parser.get("aws_creds", "database")
user = parser.get("aws_creds", "username")
password = parser.get("aws_creds", "password")
host = parser.get("aws_creds", "host")
port = parser.get("aws_creds", "port")

# connect to the redshift cluster
rs_conn = psycopg2.connect(
    "dbname=" + dbname
    + " user=" + user
    + " password=" + password
    + " host=" + host
    + " port=" + port)

rs_sql = """SELECT COALESCE(MAX(LastUpdated), '1900-01-01')
    FROM Orders;"""
rs_cursor = rs_conn.cursor()
rs_cursor.execute(rs_sql)
result = rs_cursor.fetchone()

# there's only one row and column returned
last_updated_warehouse = result[0]

rs_cursor.close()
rs_conn.commit()

# get the MySQL connection info and connect
parser = configparser.ConfigParser()
parser.read("pipeline.conf")
hostname = parser.get("mysql_config", "hostname")
port = parser.get("mysql_config", "port")
username = parser.get("mysql_config", "username")
dbname = parser.get("mysql_config", "database")
password = parser.get("mysql_config", "password")

conn = pymysql.connect(host=hostname,
        user=username,
        password=password,
        db=dbname,
        port=int(port))

if conn is None:
  print("Error connecting to the MySQL database")
else:
  print("MySQL connection established!")

m_query = """SELECT *
    FROM Orders
    WHERE LastUpdated > %s;"""
local_filename = "order_extract.csv"

m_cursor = conn.cursor()
m_cursor.execute(m_query, (last_updated_warehouse,))
results = m_cursor.fetchall()

with open(local_filename, 'w') as fp:
  csv_w = csv.writer(fp, delimiter='|')
  csv_w.writerows(results)

fp.close()
m_cursor.close()
conn.close()

# load the aws_boto_credentials values
parser = configparser.ConfigParser()
parser.read("pipeline.conf")
access_key = parser.get(
    "aws_boto_credentials",
    "access_key")
secret_key = parser.get(
    "aws_boto_credentials",
    "secret_key")
bucket_name = parser.get(
    "aws_boto_credentials",
    "bucket_name")

s3 = boto3.client(
    's3',
    aws_access_key_id=access_key,
    aws_secret_access_key=secret_key)

s3_file = local_filename

s3.upload_file(
    local_filename,
    bucket_name,
    s3_file)

변경된 사항만을 저장하므로 가장 마지막으로 업데이트 된 시점으로 부터의 변경사항을 포착하여 데이터 웨어하우스에 적재한다.

rs_sql = """SELECT COALESCE(MAX(LastUpdated), '1900-01-01')
    FROM Orders;""" # 데이터 웨어하우스에 가장 마지막으로 데이터가 적재된 시점을 불러온다.
# 이 시점 이후의 변경사항들을 불러와야 하기 때문.
rs_cursor = rs_conn.cursor()
rs_cursor.execute(rs_sql)
result = rs_cursor.fetchone()

#가장 최근 시점만 불러오므로, 한 개의 행만이 반환된다.
last_updated_warehouse = result[0]

rs_cursor.close()
rs_conn.commit()

...
m_query = """SELECT *
    FROM Orders
    WHERE LastUpdated > %s;"""

m_cursor = conn.cursor()
m_cursor.execute(m_query, (last_updated_warehouse,)) # 최근 시점 이후의 데이터들을 불러온다.
results = m_cursor.fetchall()

⇒ 위의 rs_sql 쿼리문은 데이터 웨어하우스에 데이터가 마지막으로 적재된 시점을 가져온다. 그 후, 그 시점 이후로의 데이터들을 원본 데이터베이스에서 불러와 (mysql 데이터베이스에 변경사항이 있다고 가정) 변경 사항들을 가져온다.

변경 사항들을 S3 버킷에 적재한다.

with open(local_filename, 'w') as fp:
  csv_w = csv.writer(fp, delimiter='|')
  csv_w.writerows(results)

fp.close()
m_cursor.close()
conn.close()

...

s3 = boto3.client(
    's3',
    aws_access_key_id=access_key,
    aws_secret_access_key=secret_key)

s3_file = local_filename

s3.upload_file(
    local_filename,
    bucket_name,
    s3_file)

⇒ 변경 사항들을 .csv로 만든 다음, S3 버킷에 적재한다.

💡 이 과정을 요약하면, 데이터 웨어하우스를 최신화하는 것이 목적이므로, 데이터 웨어하우스의 마지막 변경시점 이후의 모든 변경사항을 추출하여 S3 버킷에 적재하는 것이다.

REST API

: API의 엔드포인트에서 JSON 형식의 데이터들을 로드한 후, 데이터 웨어하우스에 적재하는 방식이다. (흔한 방식)

import requests
import json
import configparser
import csv
import boto3

lat = 42.36
lon = 71.05
lat_log_params = {"lat": lat, "lon": lon}

api_response = requests.get(
    "http://api.open-notify.org/astros.json", params=lat_log_params
)

requests_json = json.loads(api_response.content)

all_passes = []
for response in requests_json['people']:
    current_pass = []

    current_pass.append(lat)
    current_pass.append(lon)

    current_pass.append(response['name'])
    current_pass.append(response['craft'])

    all_passes.append(current_pass)

export_file = "export_file.csv"

with open(export_file, 'w') as fp:
    csvw = csv.writer(fp, delimiter = '|')
    csvw.writerows(all_passes)

fp.close()

parser = configparser.ConfigParser()
parser.read('../pipeline.conf')
access_key = parser.get("aws_boto_credentials", "access_key")
secret_key = parser.get("aws_boto_credentials", "secret_key")
bucket_name = parser.get("aws_boto_credentials", "bucket_name")

s3 = boto3.client('s3', aws_access_key_id=access_key, aws_secret_access_key=secret_key)

s3.upload_file(export_file, bucket_name, export_file)

API에서 요구하는 파라미터들을 적절히 기입한 다음, get을 통해 데이터를 수신한다.

lat = 42.36
lon = 71.05
lat_log_params = {"lat": lat, "lon": lon} # API GET을 위한 Parameter 설정

api_response = requests.get(
    "http://api.open-notify.org/astros.json", params=lat_log_params
)

requests_json = json.loads(api_response.content) #response를 통해 요청한 후, json을 받아온다.

원하는 형식으로 데이터들을 추출한 후에 .csv에 저장한다.

all_passes = [] # .csv의 한 행에 넣을 행 데이터 배열
for response in requests_json['people']: #페이지의 json 파싱
    current_pass = []

    current_pass.append(lat) 
    current_pass.append(lon)

    current_pass.append(response['name']) # {name : ~~}
    current_pass.append(response['craft']) #{craft : ~~}

    all_passes.append(current_pass)

export_file = "export_file.csv"

with open(export_file, 'w') as fp:
    csvw = csv.writer(fp, delimiter = '|') # 열 구분
    csvw.writerows(all_passes)

fp.close()

여기서는 api.open-notify.org의 공식 API를 사용하므로, 해당 JSON을 알맞게 파싱한 후에 .csv에 적재하고, S3 버킷에 로드한다.

Kafka 및 Debezium (Kafka 스트리밍을 이용한 CDC) 을 통한 스트리밍 데이터 수집

: Kafka의 실시간 스트리밍 데이터 수집을 이용한 CDC 구조

수집 및 적재 과정

Apache Kafka Connect에서 트랜잭션 로그를 모니터링하며 데이터의 변화를 감지한다.
데이터의 변화가 감지될 경우, Apache Kafka Connect는 이를 Kafka의 토픽으로 변환하여 전송한다.
Kafka 토픽에 변경사항들이 저장되므로, 토픽에서 이를 증분 추출하여 다시 Apache Kafka Connect를 통해 데이터 웨어하우스에 적재하게 된다.
- 보통 Kafka에는 Producer(생산자), Consumer(소비자)로 나뉘어져 있는데, 데이터 웨어하우스가 Consumer 역할을 한다.

Jiwan Ahn

Engineer, to be a Pioneer.

이전 포스트

데이터 파이프라인 - 배경 지식

다음 포스트