Batch Application

[yeollow]

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을 통해 주기적으로 실행시켜주도록 한다.

[yeollow]

kafka의 producer와 consumer를 spring-kafka client로 작성하고, consumer에서 읽어들인 데이터파일을 spring-hadoop을 이용하여 HDFS에 저장하는 방식도 고민

[yeollow]

Workflow

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

• DB의 경우 local은 h2, dev와 prod는 MySQL을 사용
• Batch 및 Spark Application은 Jenkins Job을 통해 주기적으로 실행함
• Kafka, Hadoop 등 k8s 동적구성 또한 고민

Batch Application

• Spring Batch작업을 위한 metadata Table이 필요. (각 csv파일마다 적절한 schema 정의)
  • h2 DB는 spring Boot가 자동으로 생성하지만, MySQL과 같은 DB연동 시에는 사용자가 직접 생성해야함.
    • Spring Batch의 application.yml에서 spring profile과 datasource를 정의하여 MySQL이나 h2 DB 등을 연동하여 사용하도록 한다.
  • Spring Batch의 ChunkTasklet을 따르기 위해 Job과 Step으로 구성
  • csv data -> FlatFileItemReader -> (ItemProcessor) -> KafkaItemWriter 순으로 Tasklet이 진행되며 kafka broker의 topic에 데이터가 저장됨.
  • ItemProcessor는 전처리 과정으로 서로 다른 csv file의 data schema를 살펴 공통된 부분들로만 전처리 하기 위해 정의한다.

위의 Batch Application의 kafkaItemWriter로 topic에 데이터를 전송하지 못하는 경우, batch application에서는 ItemWriter를 통해 MySQL에 데이터를 저장하고, JPA를 통해 DB있는 데이터를 모두 읽어 broker의 topic에 데이터를 전송하는 아래와 같은 kafka Producer Application을 고안하도록 한다.

Kafka Producer Application

• producer에서 send()를 통해 topic에 따라 BatchApplication에서 JPA를 통해 받아들여오는 데이터를 broker의 topic으로 전송
  • producer client는 공식문서 참조
    • core module에 읽을 data정보와 관련된 dto를 정의 후 repository에서 entitymanager를 통해 모든 데이터를 읽어 List 로 반환 -> String.format을 통해 Dto정보를 정의 후 List의 크기만큼 모든 정보를 iteration하여 kafka topic에 send().

fluentd 연동

• Input plugin : kafka 연동 후 topic data를 읽음
• Output plugin : webhdfs 연동 후 hdfs에 topic data를 .txt형태로 저장

Spark Application

• DataFrame GPS class를 생성
  • class GPS(latitude : String, longitude : String)
• HDFS의 .txt 데이터를 읽어 RDD로 MR작업 이후 DataFrame으로 변환
  • RDD로 부터 .toDF를 호출하여 생성 가능
  • RDD.map { record =>\ val splitRecord = recore.split(",") \ val longi = splitRecord(0) \ val lati = splitRecord(1) \ GPS(longi,lati) \ }.toDF
• Spark DataFrame을 MySQL로 전송
  • read / write함수를 통해 쉽게 데이터를 가져오거나 저장할 수 있음.
    • MySQL driver와 Properties정의
    • DF.write.mode(SaveMode.Append).jdbc()를 통해 기존 테이블에 append하는 방식으로write