yeollow / return-home-safely

Bigdata pipelining / Spring Batch

우선 Bigdata pipelining과정은 아래와 같으며, Spring Batch 의 실행으로부터 일어난다.
Spring Batch 과정은 job 을 기본단위로 실행하며 각 job 은 아래와 같은 구조를 가지는 하나의 tasklet 즉, step 을 단위로 실행된다.

• ItemReader : FlatFileItemReaderBuilder<>() 를 통해 공공데이터 성격 별 .csv file을 읽어들인다.
• ItemProcessor : 읽어들인 .csv file을 원하는 형식으로 전처리 한다.
• ItemWriter : KafkaItemWriter<>() 를 통해 읽어들인 성격 별 data를 Spring-kafka 를 통해 생성해놓은 kafka broker 의 topic 으로 전송한다.

이후 kafka 와 hadoop 과 연동된 fluentd plugin 을 통해 hadoop HDFS 에 kafka topic data 를 저장한 뒤, Spark 를 통해 MR작업을 거쳐 원하는 data schema형태의 dataFrame을 얻어 MySQL 에 table형태로 저장하는 과정을 거친다.

Spring API Server

위와 같이 Spring Batch 및 Bigdata pipelining 과정을 거쳐 MySQL 에 table형태로 저장된 성격 별 공공데이터를 통해 사용자의 요청에 따라 조회 후 반환하며 기본적으로 Spring MVC pattern 을 따른다.

• Controller : HTTP 를 통해 들어오는 요청을 받고, 실행한 business logic에 대한 결과물을 application/json data형식으로 다시 반환한다.
• Service : geoHash계산, CRUD작업 위임 등의 business logic을 처리한다.
• Repository : 연동된 MySQL 에서 Service에서 위임되었던 CRUD작업을 진행한다.

또한 Nginx WS를 구축하여 HTTP 기본 포트인 80을 reverse proxy한다.

GeoHash : geographic location을 문자와 숫자로 이루어진 짧은 String형태로 반환하여 공간을 사각형으로 분할하는 계층적인 공간 데이터 구조로 나눈다. GeoHash내용 참조

프로젝트 보완 사항

결과적으로 geoHash관련 부분은 Redis가 제공한다고 하니, Spring Redis연동으로 구현해본다. 해당 Application을 monolothic architecture가 아닌, docker / kubernetes를 활용한 cloud native 환경에 걸맞는 MSA구조로의 변환을 염두해본다.

return-home-safely's Issues

build gradle multi-module project

multi module project 구성

모든 의존성 관리는 rootProject 경로의 build.gradle에서 관리토록 한다.

• allprojects{ } : rootProject내부의 모든 모듈에 대한 필요 의존성을 등록
• subprojects{ } : rootProject 경로의 settings.gradle에 include되어있는 모듈들에 대한 필요 의존성을 등록
• project(':{module}'){ } : 각 모듈에서만 필요한 의존성을 등록
  • 각 모듈경로 내부의 build.gradle이 아님을 주의하자.

:batch

• spring-kafka client가 kafka broker의 topic에 data를 push (publisher)
• topic data를 fluentd가 hadoop에서 pull한 뒤 HDFS에 .txt파일으로 저장 (subscriber)
• HDFS의 .txt파일을 MR과정을 통해 GPS(longitude, latitude)를 뽑아내어 MariaDB에 저장 (SparkApplication)
  Spring Application과 Spark Application은 독립적인 Application이며 Jenkins Job을 통해 주기적으로 실행시켜줌.

:api

• front-end에서 request한 data를 실시간으로 DB에서 조회하여 반환
• front-end에서 전달하는 현재 GPS위치를 logging하여 file로 저장

:core

• 다른 module에서 공통적으로 사용하는 domain이나 entity등을 정의
  • core module의 경우 실행가능한 jar로 packaging하는 모든 모듈에 공통으로 들어가는 모듈로 main이 필요하지 않다.
  • 이렇게 단순 참조용 클래스만 있는 module을 위해 gradle에서는 기존에는 bootRepackage를 제공하였으나, 스프링 부트 2.0부터 bootJar와 bootWar로 Spring boot gradle plugin이 변경되었다.
    • 해당 프로젝트는 jar packaging을 목적으로 하며 bootJar를 비활성화 하기 위해 bootJar { enabled = false }, jar { enabled = true } 를 build.gradle의 project(':core'){} 에 추가해준다. 공식 문서 참고

./gradlew :module-name:build 와 같은 명령어로 module별 jar packaging 가능. [gradle docs 참조]
build이후 생성된 Jar파일은 module root의 build/libs에 생성됨

core module의 경우 모든 모듈이 공통적으로 내용을 담고 있기 때문에, 따로 jar를 패키징할 필요 없이 각 module의 jar파일 생성 시 자동으로 패키징된다.

Add reviewer lottery github actions

https://github.com/uesteibar/reviewer-lottery

[Batch] : kafkaProducer

Kafka Producer Batch proces

workflow
상기 workflow는 개발 환경(local, dev, prod)에 따라 다르게 구성

Spring Batch Job and Step
.csv파일을 읽어 kafka topic에 전달한다.

• Tasklet

  • FlatFileItemReader : .csv파일을 읽어들인다.
  • ItemProcessor : MR과정은 추후 Spark Application에서 진행 후 DB에 저장한다.
  • KafkaItemWriter : 읽어들인 .csv파일을 kafka topic에 batch작업으로 전달한다.
    • consumer를 통해 확인할 수 있어야 한다.

• CCTV / Police등 각 데이터에 대해 Tasklet을 진행한다.

Spring profile 등록

Spring profile 등록

개발 환경에 따라 설정값을 달리 해야 할 필요가 있을 때, Spring Boot에서는 profile로 취급한다.

• application.yml에는 모든 공통 설정 내용
• application-local.yml에는 local환경의 설정 내용
• application-dev.yml에는 dev환경의 설정 내용
• application-prod.yml에는 prod환경의 설정 내용

application-{phase}.yml을 다르게 지정함으로써 아래와 같이 gradle을 통한 springBoot실행 시 active profile을 지정할 수 있음.
gradle :api:bootRun -Dspring.profiles.active={phase}

code convention 규칙

Google Java Style Guide

기본적으로 Google Java Style Guide을 따른다.

• IntelliJ에 적용시킨다.
• tab size와 Indent는 4로 사용한다.

Kubernetes Cluster

kubeadm을 통한 Kubernetes Cluster 구축 및 helm chart를 통한 OSS 구축

공식 홈페이지 참조

helm chart git hub

• kafka git hub
• fluentd
  • fluentd kafka plugin git hub
  • fluentd hadoop HDFS plugin git hub
• hadoop

Batch Application

Spring Batch Application

CCTV / Police등에 대한 공공데이터 .csv파일에 대해 빅데이터 배치 프로세스를 진행한다.

• spring-kafka, spring-batch 등을 통해 kafka broker의 topic에 data를 저장한다. (producer)

• broker의 topic data는 hadoop내에 fluentd를 통해 pull하여 HDFS에 저장한다. (consumer)

• HDFS에 저장된 topic data를 spark application을 통해 MR과정을 거쳐 원하는 GPS(latitude, longitude)를 뽑아 MySQL에 저장한다.

  • spring batch와 spark application은 독립적인 application이며 Jenkins job을 통해 주기적으로 실행시켜주도록 한다.

[Documentation] Development requirements

각 항목에 대한 공부 내용 및 구현 내용에 대해 issue를 link하여 issue내용 및 Linked pull requests 참조

Spring

• spring-batch module (local)
  • kafka producer batch job
• spring-batch module (dev/prod)
• spring-api module (local)
  • local api module
• spring-api module (dev/prod)

Spark

• spark application

AWS

• download docker engine

• building kubernetes cluster

  • kubeadm

• building and install kafka

  • without helm chart
  • via helm chart

• building and install hadoop

  • without helm chart
  • via helm chart

• building and install spark

  • without helm chart
  • via helm chart

• download MySQL

  • without helm chart
  • via helm chart

• building jenkins CI/CD

  • without helm chart
  • via helm chart

kubernetes 및 helm chart를 통한 OSS 구축 및 컨테이너 애플리케이션 배포, 작업 등이 힘들면 우선 빅데이터 워크플로우 구현에 집중하고, kubernetes는 차차 공부토록 한다.

AWS ssh aliasing

ssh 접속

ssh [username]@[host] -i {pem경로}를 통해 접속할 수 있지만, ~/.ssh/config에 config 정보를 아래와 같이 입력해주면 지정해준 hostname을 통해 ssh hostName으로 바로 접근 할 수 있음.

• .pem file은 base64로 encoding된 ASCII 텍스트파일으로 개인키, 서버 인증서 및 SSL발급 요청 시 생성하는 포맷이다.
• 본 프로젝트에서는 AWS instance 4개를 가용하며 기본적인 config 정보는 아래와 같다.
  • {}안의 내용을 사용자 편의대로 기술하면 되며 pem file의 경우 AWS Instance생성 시 발급 받았던 key-pair의 private key file의 이름을 기술해야한다.

Host {hostName}
	HostName {AWS ip address}
	User ubuntu
	IdentityFile ~/.ssh/{pemFileName}.pem
	
Host {hostName}
	HostName {AWS ip address}
	User ubuntu
	IdentityFile ~/.ssh/{pemFileName}.pem
			
Host {hostName}
	HostName {AWS ip address}
	User ubuntu
	IdentityFile ~/.ssh/{pemFileName}.pem
				
Host {hostName}
	HostName {AWS ip address}
	User ubuntu
	IdentityFile ~/.ssh/{pemFileName}.pem

[API] : api module

Spring Boot Api Module

workflow
상기 workflow는 개발환경(local, dev, prod)에 따라 아래와 같이 다르게 구성

• local

  • local에 MySQL을 연동하여 batch 및 api module을 실행

• dev / prod

  • AWS instance에 MySQL을 다운받은 다음, MySQL 드라이버와 연동.
  • AWS instance에 Spring boot application을 jar형태로 배포
  • jenkins CI/CD를 구축

우선 Batch작업을 통해 JpaItemWriter로 DB에 저장. Batch module의 Bigdata Pipelining구축 및 Spark Application 개발이 완료되면 연계 진행 및 Kubernetes orchestration 공부 및 진행

Batch Job 을 통해 저장된 MySQL data를 front요청 시 반환토록 한다.

• REST API설계 규칙에 따라 url을 mapping한다.
• Spring MVC pattern을 따라 원하는 data의 GPS정보를 반환한다.

Establish

Project-Setup

• Spring Initializer
• Gradle multi project
• Spring profile
• Loback
• swagger
• code convention
• AWS ssh aliasing

init SpringBoot Project

Spring Initializer

• Gradle Project
• Java 11
• 2.4.4 Version Spring Boot
• Jar Packaging

[AWS] : building kafka cluster

kafka / zookeeper cluster 구축 내용

각 hostname이 worker1, worker2, worker3인 AWS instance 3개를 통해 cluster를 구축한다.

zookeeper : 분산 시스템을 위한 코디네이터로 kafka의 동작은 zookeeper에 의해 관리된다. 별도로 zookeeper를 구축해도 되지만, 해당 프로젝트 에서는 kafka 설치 시 들어있는 zookeeper를 사용한다.

• zookeeper download : apache-zookeeper-3.7.0

  • wget을 명령어를 통해 각 AWS instance의 /usr/local에 다운받는다.
  • tar zxf를 통해 압축을 풀어준다.
  • 압축을 푼 zookeeper-3.7.0 폴더에symbolic link를 생성해준다.
    • 버전 정보가 있는 폴더는 symbolic link를 생성하여 추후 버전이 바뀌는 경우 배포 스크립트 등에 설정된 경로를 모두 변경하지 않아도 되도록 하는 것이 좋다.
    • symbolic link는 ln -s 를 통해 생성한다.

• zookeeper 설정

  • zookeeper가 지노드의 복사본인 스냅샤과 트랜잭션 로그들을 저장할 /zoo 경로를 생성한다.
    • 원하는 경로를 따로 지정해도 된다.
    • 이후 zookeeper node를 구분하기 위한 id를 /zoo하단에 myid라는 파일을 만들고 각 숫자를 입력한다.
      • worker1 -> myid : 1
      • worker2 -> myid : 2
      • worker3 -> myid : 3
  • /usr/local/zookeeper/conf아래에 zookeeper의 환경설정 파일 zoo.cfg를 생성한다.
    • zoo.cfg에서 필요한 내용 구성을 아래와 같이 기술한다.
      • zookeeper Admin docs 참조

• zookeeper 실행 : zookeeper에서 제공하는 zookeeper/bin/zkServer.sh start와 같이 start하거나 stop할 수 있다.

  • systemd 등록 : 여러 프로세스를 systemd에 등록하여 운영하도록 한다.
    • /etc/systemd/system에 zookeeper-server.service라는 파일명으로 zookeeper용 systemd파일을 아래와 같이 생성.
    • systemd의 파일을 새로 만들거나 수정하였을 경우에는 systemd 재시작을 위해 systemctl daemon-reload를 입력해야 하며 이후 systemctl명령어를 통해 zookeeper-server.service를 start하거나 stop하는 등의 명령어로 worker1,worker2,worker3에서 사용할 수 있다.
      • zookeeper가 정상적으로 실행 되었음은 systemd에 등록하였기 때문에 systemctl명령어를 통해 알아낼 수 있다.
    • daemon process란?
    • systemctl 명령어 사용법

kafka : 기본적으로 pub/sub model을 구현한 분산 메시징 시스템이다. kafka는 수평적인 확장을 위해 cluster를 구성하며 broker의 clustering을 위해 zookeeper를 통해 관리한다.

• kafka download : kafka-2.8.0 / scala 2.12

  • wget을 명령어를 통해 각 AWS instance의 /usr/local에 다운받는다.
  • tar zxf를 통해 압축을 풀어준다.
  • 압축을 푼 kafka_2.12-2.8.0 폴더에 대한 symbolic link를 zookeeper와 같이 생성해준다.
    • 버전 정보가 있는 폴더는 symbolic link를 생성하여 추후 버전이 바뀌는 경우 배포 스크립트 등에 설정된 경로를 모두 변경하지 않아도 되도록 하는 것이 좋다.
    • symbolic link는 ln -s 를 통해 생성한다.

• kafka 설정 : server별 broker id, kafka 저장 디렉토리, zookeeper 정보 등을 설정

  • kafka는 일반 message queue와는 다르게 pub/sub구조이므로 consumer가 메시지를 가져가더라도 저장된 data를 임시 보관하는 기능이 존재하며 분산 저장을 위한 여러 저장 디렉토리를 생성해도된다.
    • 원하는 경로 및 디렉토리 명을 따로 지정해도 된다.
    • 이 프로젝트 에서는 /kafka1과 /kafka2를 생성한다.
  • zookeeper server와의 통신 확인 : kafka broker server들은 zookeeper server와의 통신이 되어야하기 때문에, 방화벽으로 포트 접근이 제한된 환경인지 아닌지 등을nc 명령어를 통해 확인해봐야 한다.
    • nc -v hostname 2181을 통해 각 instance들에 대해 확인해본다.
  • /usr/local/kafka/config아래의 server.properties파일에 위에서 언급했던 broker.id, 디렉토리, zookeeper 정보를 아래와 같이 설정한다.
    • 환경설정 내용을 적용시키려면 수정 후 반드시 kafka cluster를 재시작 해야 한다.

• kafka 실행 : kafka에서 제공하는 kafka 설치경로 하위 경로의 bin에 kafka-server-start.sh를 kafka 환경설정 파일 /usr/local/kafka/config/server.properties를 옵션으로 주어 실행할 수 있으며 background실행 시 &나 -daemon option을 추가할 수 있다. stop도 마찬가지로 bin에 kafka-server-stop.sh를 실행시킨다.

  • systemd 등록 : 여러 프로세스를 systemd에 등록하여 운영하도록 한다.
    • /etc/systemd/system에 kafka-server.service라는 파일명으로 kafka용 systemd파일을 아래와 같이 생성.
    • systemd의 파일을 새로 만들거나 수정하였을 경우에는 systemd 재시작을 위해 systemctl daemon-reload를 입력해야 하며 이후 systemctl명령어를 통해 kafka-server.service를 start하거나 stop하는 등의 명령어로 worker1,worker2,worker3에서 사용할 수 있다.
      • kafka가 정상적으로 실행 되었음은 systemd에 등록하였기 때문에 systemctl명령어를 통해 알아낼 수 있다.
    • systemctl 명령어 사용법

• kafka 상태 확인: zookeeper와 kafka의 상태가 모든 instance에 대해 systemd를 통한 실행 시 active(running)상태인 경우 kafka와 zookeeper연결 확인을 위한 로그 확인

  • TCP 포트 확인 : kafka의 기본 TCP 포트는 9092, zookeeper의 기본 TCP 포트는 2181이므로 netstat명령어를 통해 상태를 확인해본다.
    • netstat -ntlp | grep {portNum} 이후 _LISTEN_하고 있는지를 kafka와 zookeeper 기본 포트에 대해 모든 instance를 확인.

LogBack 설정

Application별 로그를 분리하여 Logfile로 남기기

Tomcat 환경에서 Application별 log확인이 매우 어려움. SLF4J interface를 구현한 LogBack을 사용.

Logback의 설정 항목

• Level : TRACE -> DEBUG -> INFO -> WARN -> ERROR 레벨이 있으며, 출력 레벨 이상의 로그들만 출력함.
• Appender : event마다 log를 기록하는 기능을 처리하는 객체로 log의 출력 위치, 형식 등을 정의함. 기본적으로 ConsoleAppende, FileAppender, RollingFileAppender, SMTPAppender, DBAppender등이 있다. 기타 Appender내용
  • 해당 프로젝트 에서는 ConsoleAppender와 FileAppender를 사용한다.
• Logger : 실제 log기능을 수행하는 객체.

각 profile마다 logback설정을 다르게 할 수 있다.

Api Application

Spring Api Application

Front-end 에서 요청하는 CCTV / Police 위치에 대한 정보를, Batch Application 과정을 통해 저장된 MariaDB를 조회하여 실시간 반환한다.

• Front에서 전달 받은 현 GPS위치를 기반으로 주변 위치와 인접한 CCTV나 Police정보를 파싱하여 보여줄 수 있도록 한다.

• Front에서 전달 받은 GPS위치를 로그로 수집하며, 시각화 까지 목표로 한다. //카프카 컨수머 그룹을 따로 만들어 ES를 적용고민..?

  1. beats 로그 파일 데이터 수집
  2. beats로 수집한 데이터를 logstash를 이용하여 <K,V> 형태의 정형 데이터로 가공
     • Java API를 사용하여 logstash와 kafka를 연동
  3. Elastic Search를 이용하여 데이터를 인덱싱 하고 저장
  4. Kibana를 통한 데이터 분석 및 시각화

Swagger API문서 작성

Swagger로 API문서 spec을 자동화

프로젝트에 Swagger설정 Bean과 의존성을 등록하면 localhost:8080/swagger-ui.html에서 @controller로 지정된 java파일들에 대해 모든 path의 모든 method들이 출력된다.

• 원하는 path의 하위 메소드들만 확인하고 싶다면 SwaggerConfig파일에서 변경 가능하다.
• Swagger-ui에서 API메소드들의 기능 또한 테스트할 수 있다.