Flatmap → map 으로 메세지 개수를 조정 한뒤 아래와 같은 테스트를 진행 하였다.

1, Agent 에서 메세지당 KB 를 조정하여 AGENT 의 CPU 와  KSTREAM 의  전체 Throuput 을 관찰하는 것이 목적이다.

Partition 1 - Thread 1 - Sleep 0 기준 테스트

2.같은 Thruput 을 Partition3 = thread 3 option 에서 관찰

3.2와같은 환경에서 timesleep = 30ms 조건에서 관찰

4. sleep time 당 Throghput in(주황) 과 CPu 점유율(파랑)

• 컨트롤러 선정
• 브로커 메타 데이터
• 토픽 메타데이터
• 클라이언트 할당^quota 정보

컨트롤러 선정

컨트롤러 - 파티션 관리를 책임지는 브로커로 파티션 관리 범위는 아래와 같다.

• 리더 선정
• 토픽 생성
• 파티션 생성
• 복제본 관리

하나의 노드 또는 서버가 꺼지면 컨트롤러는 팔로워 중에서 파티션 리더를 선정한다.

카프카는 컨트롤러를 선정하기 위해 주키퍼의 메타데이터 정보를 활용한다.

주키퍼는 현재의 컨트롤러에 장애가 나면 새로운 컨트롤러가 선정되는 것을 보장한다.

브로커 메타데이터

주키퍼는 카프카 클러스터의 일부인 각 브로커에 대해 상태정보를 기록한다. 클러스터 내에서 각 브로커의 모든 메타데이터를 기록한다.

프로듀서와 컨슈머는 주키퍼와의 상호 작용으로 브로커의 상태 정보를 얻는다.

토픽 메타 데이터

주키퍼는 파티션수, 특정한 설정 파라미터 등의 토픽 메타데이터를 기록한다.

클라이언트 할당 정보

할당량은 카프카 토픽의 메시지를 읽고 쓰는 클라이언트에 대한 바이트 비율의 임계 값을 제한하며, 모든 정보와 상태는 주키퍼가 관리한다.

카프카 토픽 ACLs

카프카는 내장된 인증모듈(ACL, Access Control List)을 가지고 있다.

이러한 ACLs는 사용자 역할과 관련된 토픽에 대해 읽기와 쓰기 권한 종류를 결정한다. 카프카는 ACLs를 저장하는데 주키퍼를 사용한다.

파티션 규칙(default partitioner)

• key가 있는경우 :  key를 hash하여 특정 파티션으로 보냄. 키와 파티션간의 mapping은 topic의 파티션 개수가 변경되지 않는경우 유지됨.
• key가 없는 경우 : 임의로 사용가능한 파티션으로 보낸다. 파티션간의 균형을 맞추는데는 Round Robin 알고리즘을 사용

파티션 개수의 산정방법

파티션 개수를 산정할 때 고려할 점은 다음과 같다

• 단위시간당 토픽의 처리량(throughput). 예를 들어, 초당 100KB 또는 1GB를 예상 하는가?
• 한 파티션의 데이터를 읽을 때 목표로 하는 최대 처리량은? 파티션 하나는 항상 한 컨슈머가 소비한다._{(컨슈머 그룹을 사용하지 않을 때도 컨슈머는 항상 해당 파티션의 모든 메시지를 읽어야 하기 때문)}  따라서 처리 속도가 느린 컨슈머가 피티션을 읽어서 그 데이터를 데이터베이스에 쓸 때 데이터를 쓰는 스레드마다 초당 50MB 까지만 처리할 수 있다면 파티션을 소비하는 최대 처리량이 초당 50MB로 제한 된는것을 염두해야 한다.
• 하나의 파티션에 데이터를 생성하는 프로듀서당 최대 처리량도 컨슈머와 같은 방법으로 산정할 수 있다. 그러나 대개 프로듀서는 컨슈머보다 훨씬 빠르게 처리되므로 처리량을 조사하지 않아도 무방.
• 키^key를 사용해서 파티션에 메시지를 쓰는 경우에는 향후에 파티션을 추가할 때 개수 산정이 어려울 수 있다. 따라서 현재보다는 향후에 예상되는 사용 방법을 기준으로 처리량을 추산하자.
• 브로커마다 파티션 개수와 디스크 용량 및 네트워크 처리속도를 고려하자.
• 파티션 개수를 너무 많이 고려하지 말자. 각 파티션은 브러커의 메모리와 그 외 다른 자원을 사용하므로 리더선정에 더 많은 시간이 소요되기 때문.

이 모든 것을 고려할 때 파티션 개수를 너무 많이 산정하지 않아야 한다. 토픽의 목표 처리량과 컨슈머의 예상 처리량에 관한 추정치가 있다면, 목표 처리량을 컨슈머 예상 처리량으로 나누는 방법으로 파티션 개수를 산출할 수 있다. 예를 들어 초당 1GB로 토픽을 읽고 쓰기 원하는데 각 컨슈머가 초당 50MB만 처리할 수 있다면 최소한 20개(1000/50)의 파티션이 필요하다는 것을 알 수 있다. 이 경우 20개의 컨슈머가 토픽을 읽으므로 초당 1GB의 목표 처리량을 성취할 수 있다. 그러나 만일 이러한 자세한 정보가 없다면 디스크에 보존하는 파티션 크기를 하루에 6GB미만으로 제한할 것을 권함.

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-manager

github : https://github.com/yahoo/kafka-manager

kafka 매니저는 yahoo에서 개발한 kafka 관리 도구이며 아래의 기능을 제공한다.

• 다수의 클러스터 관리
• 간단하게 클러스터 상태 조회 가능(topics, consumers, offsets, brokers, replica distribution, partition distribution)
• Preferred replica election 실행

• 사용할 브로커 선택 옵션을 사용하여 파티션 할당 생성

• 파티션 재할당 (based on generated assignments)
• 토픽 생성시 옵션 설정 가능(0.8.1.1 has different configs than 0.8.2+)
• 토픽 삭제(버전 0.8.2 이상, delete.topic.enable=true 설정시)
• 삭제표시된 토픽리스트 조회 (only supported on 0.8.2+)
• 다수의 토픽에 일괄 파티션생성 및 사용할 브로커 할당
• 다수의 토픽에 파티션 재할당 일괄 실행
• 기존 토픽에 파티션 추가.
• 기존 토픽 설정 업데이트
• (Optional) enable JMX polling for broker level and topic level metrics.
• (Optional) filter out consumers that do not have ids/ owners/ & offsets/ directories in zookeeper.

설치

SBT(simple build tool) 설치

• windows
  
  • 다운로드페이지((https://www.scala-sbt.org/download.html)에서 설치파일을 받아 설치

• CentOS
  참고페이지
  아래 커맨드를 이용해 repo 파일을 받고, yum을 이용해 설치

  $ curl https://bintray.com/sbt/rpm/rpm | sudo tee /etc/yum.repos.d/bintray-sbt-rpm.repo $ sudo yum install sbt

Kafka-manager 설치

1. 파일 다운로드 : 다운로드페이지에서 압축파일(zip or tar.gz)을 다운로드.

2. 압축해재 후 sbt를 이용해 배포

   C:\dev\kafka-manager>sbt clean dist

    실행 결과

   `sbt run` 명령어를 입력해 실행 가능. 또는 \target\universal 경로에 생성되는 zip 파일을  원하는 경로에 압축 해제

kafka-manager 설정변경

압축 해제한 경로로 이동하여 설정파일을 수정해야 한다.

• 주키퍼 호스트 설정 : conf/application.conf 파일 수정

  conf/application.conf

  #zookeeper 서버가 한 대인 경우 $ kafka-manager.zkhosts="localhost:2181" #사용하는 zookeeper 정보에 맞춰 수정     #zookeeper 서버가 다수인 경우 $ kafka-manager.zkhosts="my.zookeeper.host.com:2181,other.zookeeper.host.com:2181" #쉼표를 이용해 다수의 주키퍼 호스트 입력가능

  더 자세한 내용은 https://github.com/yahoo/kafka-manager#configuration 참조

• JMX port 설정
  kafka-manager에서 metric 정보를 확인하고자 하면 jmx를 활성화 해 주어야 한다. 
  `bin/kafka-server-start.sh` 파일을 열어 JMX 포트 설정을 추가한 뒤 kafka를 재기동 해준다.

  ${KAFKA_HOME}/bin/kafka-server-start.sh

  export JMX_PORT=9999

  
  Kafka Manager UI에 접속해 Clusters메뉴에서 modify 버튼은 클릭해 JMX 사용옵션을 활성화 한다.
  
  

  

  
  

kafka-manager 구동

압축 해제한 경로로 이동해 프로그램을 실행한다.

기본적으로 9000번 포트를 사용하며 실행시 아래 옵션을 통해 설정파일의 위치를 지정하거나 포트를 설정할 수 있다.

Usually, you don't need to modify these settings, however, if you want to extract every last bit of performance from your machines, then changing some of them can help. You may have to tweak some of the values, but these worked 80% of the cases for me:

• message.max.bytes=1000000
• num.network.threads=3
• num.io.threads=8
• background.threads=10
• queued.max.requests=500
• socket.send.buffer.bytes=102400
• socket.receive.buffer.bytes=102400
• socket.request.max.bytes=104857600
• num.partitions=1

Quick explanations of the numbers:

• message.max.bytes: This sets the maximum size of the message that the server can receive. This should be set to prevent any producer from inadvertently sending extra large messages and swamping the consumers. The default size is 1000000.
• num.network.threads: This sets the number of threads running to handle the network's request. If you are going to have too many requests coming in, then you need to change this value. Else, you are good to go. Its default value is 3.
• num.io.threads: This sets the number of threads spawned for IO operations. This is should be set to the number of disks present at the least. Its default value is 8.
• background.threads: This sets the number of threads that will be running and doing various background jobs. These include deleting old log files. Its default value is 10 and you might not need to change it.
• queued.max.requests: This sets the queue size that holds the pending messages while others are being processed by the IO threads. If the queue is full, the network threads will stop accepting any more messages. If you have erratic loads in your application, you need to set queued.max.requests to a value at which it will not throttle.
• socket.send.buffer.bytes: This sets the SO_SNDBUFF buffer size, which is used for socket connections.
• socket.receive.buffer.bytes: This sets the SO_RCVBUFF buffer size, which is used for socket connections.
• socket.request.max.bytes: This sets the maximum size of the request that the server can receive. This should be smaller than the Java heap size you have set.
• num.partitions: This sets the number of default partitions of a topic you create without explicitly giving any partition size.

Number of partitions may have to be higher than 1 for reliability, but for performance (even not realistic :)), 1 is better.

These are no silver bullet :), however, you could test these changes with a test topic and 1,000/10,000/100,000 messages per second to see the difference between default values and adjusted values. Vary some of them to see the difference.

You may need to configure your Java installation for maximum performance. This includes the settings for heap, socket size, and so on.

출처 : https://www.confluent.io/product/connectors/ , https://docs.confluent.io/current/connect/connect-filestream/filestream_connector.html

소스 업데이트 >  https://github.com/Sangil55/SimpleFileConnetor/

Understanding Kafka Connect and Kafka Stream

Kafka Connect is a tool that provides the ability to move data into and out of the Kafka system. Thousands of use cases use the same kind of source and target system while using Kafka. Kafka Connect is comprised of Connectors for those common source or target systems.

Connect makes it simple to use existing connector implementations for common data sources and sinks to move data into and out of Kafka. A source connector can ingest entire databases and stream table updates to Kafka topics. It can also collect metrics from all of your application servers into Kafka topics, making the data available for stream processing with low latency. A sink connector can deliver data from Kafka topics into secondary indexes such as Elasticsearch or batch systems such as Hadoop for offline analysis.

Kafka Connect can run either as a standalone process for running jobs on a single machine (e.g., log collection), or as a distributed, scalable, fault tolerant service supporting an entire organization. This allows it to scale down to development, testing, and small production deployments with a low barrier to entry and low operational overhead, and to scale up to support a large organization’s data pipeline.

•  Import Connectors are used to bring data from the source system into Kafka topics. These Connectors accept configuration through property files and bring data into a Kafka topic in any way you wish. You don't need to write your own producer to do such jobs. A few of the popular Connectors are JDBC Connector, file Connector, and so on.
• Export Connectors: Unlike Import Connectors, Export Connectors are used to copy data from Kafka topics to the target system. This also works based on the configuration property file, which can vary based on which Connector you are using. Some popular Export Connectors are HDFS, S3, Elasticsearch, and so on.

kind of import Connectors 

• Active MQ -  AMAZONE 서버용 Apache ActiveMQ용 관리형 메시지 브로커 서비스,  
• IBM MQ -  Source Connector is used to read messages from an IBM MQ cluster and write them to a Kafka topic.
• JDBC (Source) - for jdbc
• Kinetica (Source) - for GPU, RDBMS
• Azure IoTHub (Source)IoT, messaging
• Files/Directories (Source)File System, Directories, Logs  Community Community 1Community 2 (confluent 사)
• FileSystem Connector (Source)File System, S3, HDFS  Community Community
• ..... various
  

kind of export Connectors 

• Amazon S3 (Sink)
• HDFS (Sink)
• Cassandra (Sink)
• .. File System out
• ...various

등등, 여러가지 connector 플랫폼이 지원되 는데, FILE SYSTEM에서도 connetor 사용 용도는 있지만

주 용도는 직접 data 를 직접 stream 하기 어려운 환경을 위한 지원 툴이 Connector 라고 보면된다. AMAZON SERVER / DB / GPU / AMAZON &  IBM MQ(legacy) / IOT 등 다양한 곳에서 producer 로 data를 stream 하기 위한 connector 의 용도별로 컨풀르엔트사에서 공식 지원 하고 있다.

기업형 메세징 플랫폼의 경우에는 사내 서버이고, 별도 AP AGENT 단의 MQ 구축이 어렵기 때문에  (비교적 간단한 MESSAGE 서버로 NATS SERVER  , 간편 CDC 등도 조사필요)

 import 용 Connector 는 confulent 에서 제공하는 connect-api 를 사용 해서 구현 해야한다.
완성된 번들로도 제공 하며, JAVA 로는 아래 dependency 등 SourceTask을 참조하여 링크처럼 커스터마이징 하여 구현 할 수 있다 (참고 . https://docs.confluent.io/current/connect/devguide.html )

• *완성된 파일 커넥터를 사용 하기 위해선 기본 컨플루엔트 솔루션을 설치하라고 이야기한다.  https://docs.confluent.io/current/connect/quickstart.html#connect-quickstart
• 그렇게 하지 않으려면 , JAR 형태로 전체 빌드하여 사용하는것을 생각 해보아야 한다. (다만 파일 읽고 offset 설정하는 부분 직접 구현 필요)

Export 용 Connector 는 우리 상황에서는  Confulent (Default) or  HDFS 용 Export connector 로 고려 할 만하다.

다만 >> KAFKA STREAMS API 를 사용하거나 SPARK - SREAMING 을 사용해서 별도 출력 하게되면, 별도의 output connector 의 사용의미가 없어진다.

KAFKA CONNECTOR   VS   FILEBEAT (ELK)

본격적으로 성능 비교를 하자면, 

ELK 기반의 FILE BEATS AGENT 도 OUT FORMAT 으로 KAFKA topic 을 입력 가능하다.

Filebeat 의 yml 설정파일

KAFKA 에서 제공하는 Connector 사용과 FILEBEAT 를 비교하여 아래 특성등을 비교할 필요가 있다.

• producer import perfomance
  
            >> FILE BEAT 의 경우 아래  테스트 진행

• >>>Kafka Connector의 경우 https://docs.confluent.io/current/connect/quickstart.html#connect-quickstart 을 통해 Bundle script 로 테스트
  Java 로 재빌드 등을 시도 하였으나,

         - bundle connector 의 경우 파일 크기가 일정 크기 이상이면, 동작 안함, 800K 까지는 정상 동작확인, - 약 4166 rows

          -java code 의 경우 에는 그대로 빌드시 Context 관련 초기화 오류 발생 (implement 하여 재구성 필요!)

Confluent 에서 제공하는 CONNECTOR API 소스를 간략히 설명하자면, Java Buffer  reader 를 통해 polling 하는 구조이다.

          - 해당 class를 참고하여 재구현 할 경우 line input & offset 쓰고 읽는 부분에 대해서 직접 구현이 필요한데, 시간이 더 소요 될듯 하다. 

          - 이것을 Filesystem 으로 기반하여 구현한 OPEN SOURCE 등을 다시 참고해보면

            오픈소스1 :  https://github.com/jcustenborder/kafka-connect-spooldir/blob/master/src/main/java/com/github/jcustenborder/kafka/connect/spooldir/SpoolDirSourceConnector.java

             오픈소스2 : https://github.com/mmolimar/kafka-connect-fs

             위 2개중에 Connector 는 agent 단에 놓고,  내부 NETWORK 의 KAFKA BROKER 까지로 전송 할 수 있는 오픈소스를 찾아서 성능 테스트 해보아야한다.

             다만 위 오픈소스 2개 모두 , KAFKA CONFULENT 플랫폼 기반이기 때문에, 원정서버(MAS01~26) 에 일부 설치를 고려해야한다.(즉, Kafka Connector (stand alone등이)  자체적으로 실행 되어야 된다.)

• Replication / Partition  Performance

• Scalblity

•  fault tolerance 

• *source 수정 편의성

kafka vs flink  : https://www.confluent.io/blog/apache-flink-apache-kafka-streams-comparison-guideline-users/