[R&D] 프로듀서

라이브러리 목록 (json dumps = https://www.daleseo.com/python-json/)

카프카 프로듀서로 보낼 때 result set의 row를 json으로 직렬화 해서 보냄. json으로 변환하면 이종 시스템 간 통신이 가능하다.

import cx_Oracle
import os
from json import dumps
from kafka import KafkaProducer

 

cx_Oracle로 오라클DB 연결하고 SQL 실행을 위한 cursor를 선언. Cursor는 result set으로부터 fetch rows 하기 위한 pointer이다. 아래처럼 cursor 객체로 쿼리문을 실행하고 결과를 받아온다. Cursor 별로 다른 query 결과를 담아두고 필요데 따라 fetch하면서 사용하는 식의 프로세스라면 여러 cursor를 쓸 수 있다. 보통은 1개로 반복 사용함. (https://devluna.blogspot.com/2015/09/python-oracle-03-cxoracle-sql-cursor.html)

#DB접속하기
conn = cx_Oracle.connect('cpsrndver/cpsrndver123@222.122.47.39:1521/orcl', encoding='UTF-8')

#V_STD_CDP_SREG(학생정보 테이블) 가져오기
V_STD_CDP_SREG = conn.cursor()
V_STD_CDP_SREG.execute('SELECT * FROM V_STD_CDP_SREG')

# 커서 rowfactory로 지정
V_STD_CDP_SREG.rowfactory = makeDictFactory(V_STD_CDP_SREG)

 

이후 cursor로 fetchall() 호출해서 result set 불러오고 각 row마다 프로듀서로 카프카 브로커에 보낸다. 하나의 프로듀서로도 토픽이 자유자재로 변경 가능하다.

# 테이블에서 데이터 가져오기
results = V_STD_CDP_SREG.fetchall()

for result in results:
    data = result
    print(result)
    producer.send('V_STD_CDP_SREG', value=data)

print('V_STD_CDP_SREG 테이블 전송완료')

V_STD_CDP_SREG.close()