데브포유

🧩 NiFi란 무엇인가요?Apache NiFi는 데이터 흐름을 시각적으로 설계하고 자동화할 수 있는 오픈소스 플랫폼입니다. 복잡한 ETL 작업, 실시간 스트리밍, IoT 데이터 처리까지 다양한 분야에서 활용되며, GUI 기반의 인터페이스 덕분에 코딩 없이도 강력한 데이터 파이프라인을 구축할 수 있어요.🔧 NiFi의 주요 기능그래픽 기반 워크플로우 설계드래그 앤 드롭 방식으로 데이터 흐름을 구성실시간 모니터링 및 수정 가능다양한 프로세서 내장300개 이상의 내장 프로세서로 파일 처리, API 호출, DB 연동 등 가능커스텀 프로세서도 개발 가능데이터 라우팅 및 필터링조건에 따라 데이터 흐름을 분기하거나 필터링실패 시 자동 재처리 및 경로 변경 가능보안 및 접근 제어SSL, 인증서 기반 암호화사용자별 권..

간만에 살펴 본 db 랭킹.Snowflack와 Databricks 성장이 눈에 띄고요.기타로 Clickhouse 도 랭킹 많이 올라왔습니다. AI 시대에 대용량 데이터 처리가 필수이니 동반해서 성장한 것으로 보입니다.

위 처럼 설정하면 10분 마다 idle 상태인 connection의 유효성 검사를 수행합니다.10개의 커넥션이 연결되어 있을 경우, 10분마다 디폴트 값에 의해서 3개의 connection이 유효성 검사를 수행됩니다.오라클 DBMS를 사용할 경우에 v$session에 있는 값을 통해서 확인할 수 있습니다. SELECT SID, PREV_EXEC_STARTFROM V$SESSIONWHERE USERNAME = '' AND OSUSER = ''

데이터베이스에서 특정 조건에 맞는 데이터를 조회해서 Json 형식으로 변환하고, 변환된 데이터에값을 추가하거나 변환하는 작업을 거친 후에, RestAPI로 전달하고 응답 결과를 파일로 저장하는 예제를만들어 보았습니다.Jolt Transform을 처음 사용해 보았는데, 아래 사이트가 많은 도움이 되었습니다.아래 예제는 Json에 디폴트 값을 추가하고, Json 데이터의 Key Name을 대문자에서 카멜 형식으로 변환하는 작업을 2개의 프로세서로 했는데, Jolt 문법으로 한번에 할 수도 있습니다.https://jolt-demo.appspot.com/#inception

import groovy.json.JsonOutputimport org.apache.commons.text.StringEscapeUtilsdef data = [message: '안녕하세요']def json = JsonOutput.toJson(data)// 유니코드 이스케이프 시퀀스를 UTF-8로 직접 변환def decodedJson = StringEscapeUtils.unescapeJson(json)println "Decoded JSON: $decodedJson"// 파일에 UTF-8로 저장new File('output.json').write(decodedJson, 'UTF-8')

특정 테이블 (tc_ymd)를 조회한 결과를 json으로 BATCH_SIZE (100건) 씩 Flow File로 만들어서 다음 프로세서로전달하는 예제입니다.지난번에 포스팅 했던  Json 데이터를 Oracle Merge문으로 변환하는 프로세서로 전달하기 위해서 만들어 봤습니다.Groovy Script로 작성한 이유는 조회한 데이터 일부를 특정 처리 (ex. 암호화)를 한 후에  다음 프로세서에 전달하기위해서 입니다. * tc_ymd.groovyimport org.apache.nifi.dbcp.DBCPServiceimport org.apache.nifi.processor.io.StreamCallbackimport org.apache.nifi.processor.io.OutputStreamCallbackimp..

import org.apache.nifi.controller.ControllerService def lookup = context.controllerServiceLookup def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find { cs -> lookup.getControllerServiceName(cs) == "HF_TSTDB" } def dbcpService = lookup.getControllerService(dbcpServiceId) //def dbcpService = context.controllerServiceLookup.getControllerService("43c54e78-0191-1000-9a98..

ConvertJSONToSQLWithMerge.java최근에 Oracle 데이터베이스를 목적지로 데이터를 이관하기 위해서 Nifi를 검토하던 중에, Nifi에는 Oracle Merge을 지원하는프로세서가 없다는 걸 알게되었습니다.Upsert 형식을 지원하는 프로세서가 있지만, 대상 데이터베이스에는 Oracle이 없습니다.그래서 구글링 해보니 Github에 커스텀 프로세서 코드가 있어서 사용해 보았습니다.출처: https://github.com/dawsongzhao1104/nifi출처: https://mdnice.com/writing/1e7798958878469cbd54e1dfe85126a7 그런데 실제로 테스트를 해보니, 커스컴 프로세서을 인식해서 사용할 수는 있는데, Merge문이 온전하게생성이 되지 ..