My data lab

kafka 로그 설정

메시지가 브로커 컴퓨터에 보존되도록 구성하는 방법을 설정.

설정항목

• log.segment.bytes : 바이트 단위로 최대 세그먼트 크기를 정의. 기본값은 1GB. 
  세그먼트 파일이 지정된 크기에 도달하면 새로운 세그먼트 파일이 생성됨. 토픽은 로그 디렉터리에 여러 세그먼트 파일로 저장되며, 토픽단위로 설정이 가능.
• log.roll.{ms,hours} : 새로운 세그먼트 파일이 생성된 이후 크기한도에 도달하지 않았더라도 다음 파일로 넘어갈 시간 주기를 정의. 기본값은 7일, 토픽단위로 설정가능
• log.cleanup.policy : delete 또는 compact로 설정하며 기본값은 delete. delete로 설정된 경우 로그 세그먼트는 시간이나 크기제한에 도달할 때 주기적으로 삭제됨.
  compact로 설정된 경우 불필요한 레코드를 없애기 위해 압축을 사용. 토픽단위로 설정 가능.
• log.retention.{ms,minutes,hours} : 로그 세그먼트를 보유할 시간을 정의. 기본값은 7이며 토픽별로 정의 가능.
• log.retention.bytes :  삭제하기 전에 보관할 파티션 당 로그 수. 기본값은 -1. 토픽별로 설정이 가능하며, 로그의 시간이나 크기제한에 도달하면 삭제됨.
• log.retention.check.interval.ms : 로그의 보유 정책을 만족하기 위해 삭제할 대상을 확인하는 시간 주기. 기본값은 5분
• log.cleaner.enable : 압축을 활성화 하려면 true로 설정
• log.cleaner.threads : 로그의 압축을 위한 작업자 스레드 수를 지정.
• log.cleaner.backoff.ms : 정리가 필요한 로그가 있는지 확인하는 주기
• log.index.size.max.bytes : 바이트 단위로 오프셋 인덱스의 최대 크기를 설정. 기본값은 1GB. 토픽별로 설정 가능.
• log.index.interval.byte : 새로운 항목이 오프셋 인덱스에 추가되는 주기. 기본값은 4096. 
  데이터를 가져오는 개별요청에서 브로커는 가져오기를 시작하고 끝낼 로그 내의 올바른 위치를 찾기 위한 바이트 수에 대해 일정하게 살펴본다.
  값을 높게 설정한 경우 인덱스 파일이 커지고 더 많은 메모리를 사용하게 되지만 검사하는 횟수는 줄어듬.
• log.flush.interval.message : 디스크로 내보내기 전에 메모리에 보유할 메시지의 개수. 기본값은 922337036854775807. 확실하게 보유하는지 보장하는 것은 아니나, 알맞게 제어하도록 도와줌.
• log.flush.interval.ms : 디스크로 내보내기 전에 메모리에 보유할 토픽 내의 메시지에 대한 최대 시간을 밀리초 단위로 설정.

참고

• 설정관련 더욱 자세한 내용은 공식문서 brokerconfigs 항목 참고.
• 로그압축에 관한 정보는 공식문서 compaction 항목 참고

추가 로그 정책 관련 설명

ref: https://stackoverflow.com/questions/40369238/which-directory-does-apache-kafka-store-the-data-in-broker-nodes

ambari 기준으로 kafka 관련된 로그는 2군데에 저장 된다.

log.dirs : 메세지 보관을 위한 오프셋을 포함한 로그이다.

/var/log/kafka : kafka 자체 로그이며 kafka.env 파일에서 설정 할 수 있다.

1. Topic 에 관한 로그 는 아래 정책으로 log.dirs 경로(server.properties 파일에 설정) 에 저장된다.

로그의 잘려진 segment 는 상응하는 index 파일과 같이 저장된다.  그 파일 내임은 Base offset 을 표현하기도 한다.

이것을 설명하자면 log 파일 이저장되는 구조를 이해 해야하는데, 

log 파일은 실제메세지를 strucuted message format 으로 저장한다.  each message 는 처음 64 비트에 증가된 offset 을 포함한다.

그러므로 이파일에서 특정 오프셋의 메세지를 찾는것은 log 파일이 커질 수록 무거워진다.

또한 메세지를 공급하기 위해서는 브로커는 가장 나중의 offset 을 기억하고 메세지를 정확하게 받아들이기 위해 진행 시킨다.

그러므로 로그파일의 offset 기억하기 위해 index 파일이 존재한다. 

index 파일은 아래와 같은 항목을 저장한다.

1. 4 Bytes: Relative Offset
2. 4 Bytes: Physical Position

이때 파일내임은 base offset 을 표현한다. index 파일은 로그파일의 index.interval.bytes 의 단위마다 index 파일을 새로 쓴다.(default 4096)

2. kafka 서버자체의 로그 (INFO / ERROR / WARN 등) 은 kafka_env.sh 안에 로그 경로가 지정되어있다.

주로내용은 자바 info 로 채워지는데 이 또한 시간이 지나면 일주일에1~2G 정도 용량이 찬다.

log4j 설정을 이용해 로그양을 줄일 수 있다.  2번 은 말그대로 log 이다.

파일전송을 KAFKA PRODUCER 쪽으로 보내기위한 AGENT 단의 구현이 필요 했다.

보통 JDBC 등의 소스에서는 kafkaconnector 가 지원하는데,

단순 경로에서 spool 하여 Producer 쪽에 쏴주는 kafkaconnector 는 존재 하지 않기 때문에 ,

직접구현하고 github 에 구현하였다.

https://github.com/Sangil55/SimpleFileConnetor

출처 : https://kafka.apache.org/quickstart /

궁극적으로 BIG DATA 실시간 분석이 목적인데,

Spark Streaming 하기 위한 전단계 중 Kafka Infra 를 앞단에 놓고 실시간 Streaming 을 사용 하여 구현하게된다.

Kafka 는 개략적으로

이런 모양을 하고 cluseter, 즉 broker 가 중간 역할을 하여 data 를 전달 한다.

프로듀서 - 컨슈머 동작을 하기위해선 아래 순서로 설치한다.

1.일단 KAFKA 설치 (직접설치 또는 ambari 등을 이용)

2.zookeeper 환경설치 (직접설치 또는 ambari 등을 이용)

3.Kafka 경로의 server.properties 파일에서 설정 확인

Listener = PLAINTEXT://localhost:6667 또는 9092포트

advertisment.Listener(deprecated)

>>으로 설정 이후 Zookeeper 는 6667 포트를 기준으로 consumer / producer 를 보내게 된다.

default.replication.factor = 1

offsets.topic.replication.factor  = 1 (3 으로설정되어있으면 consumer 가 topic 이 3개 기준으로 동작하기 때문에 추후 조정)

4. 토픽 생성 

/usr/bin/kafka-topics.sh --create --zookeeper localhost:6667 --replication-factor 1 --partitions 1 --topic test

5. 프로듀서 테스트

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-producer.sh --broker-list swinno01.cs9046cloud.internal:6667 --topic Hello 

6.컨슈머 테스트

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-consumer.sh --zookeeper swinno01:2181 --topic Hello

/usr/hdp/2.6.5.0-292/kafka/bin/kafka-console-consumer.sh --bootstrap-server swinno01.cs9046cloud.internal:6667 --topic Hello --from-beginning 

이때 컨슈머가 제대로 윗라인으론 동작을 하는데, 아래 bootstrap 옵션으로 동작 안하는 경우가있다.

그땐 다수의 삽질과 / 구글링등을 통해서 아래를 확인해보자

a.broker 의 netstat -tnlp 를 통해 6667 (9092) 포트가 오픈되있는지 확인

b.실행환경에서 브로커ip:port 로 telnet 확인

c. zookeeper 스크립트를 통한 브로커 상태확인

/usr/hdp/2.6.5.0-292/kafka/bin/zookeeper-shell.sh localhost:2181   

 get /brokers/ids/1001  (ids num 은 확인필요)

d. topic/ broker 모두 삭제후 KAFKA 재시작

e.  offsets.topic.replication.factor  = 1 확인

연결이 상태가 안 좋다면 아래글 참고

https://github.com/wurstmeister/kafka-docker/wiki/Connectivity

지금도 forum / stackoverflow 등에는 위의 이유등으로 컨슈머가 bootstrap 에서 안된다는 이슈는 꾸준히 나오고있다 -_-ㅋ

이런 설정문제이거나/ 연결문제이거나 / 브로커 잔재 문제이거나

컨슈머까지 스크립트에서 동작하는것을 확인하면

spark 를 이용한 streaming 하기 위한 전단계 까지 완료 !