$$config title: [kafka] 카프카 keywords: - IT - ... $$ # 계획서 1. ~~카프카 빠르게 시작해보기~~ 2. 카프카 기본 개념 설명 - 브로커 - 클러스터 - 주키퍼 - 토픽 - 파티션 - 레코드 - 카프카 클라이언트 - 카프카 스트림즈 - 카프카 커넥트 - 카프카 미러메이커2 3. 카프카 상세 개념 설명 4. 카프카 실전 프로젝트 5. 클라우드 카프카 서비스 # kafka 작동 구조 - 입력: 프로듀서 - 토픽: 여러개의 파티션 (FIFO 큐 구조) - 출력: 컨슈머 # kafka 데이터 포맷 - 제한 없음 (ByteArray로 통신하기 때문) - 자바에서 가능한 모든 객체 지원 - 데이터를 묶음 단위로 전송 (그로인한 결과로 낮은 지연과 높은 데이터 처리량을 가짐) => 대역폭이 넓다 # kafka를 사용하는 회사 - 넷플릭스 - 우버 - 월마트 - 에어비엔비 > 사용하지 않는 곳이 없다 > 빅데이터를 사용하는 곳이라면 당연히 사용한다 # 데이터 레이크 (data lake) - 데이터 호수 - 데이터가 모이는 저장 공간 - 원래의 의미는 정제되지 않은 데이터들이 모두 모이는 곳 > 1:1로 연결하기에는 복잡도가 올라가므로 데이터의 추출, 변경, 적재하는 과정이 필요한데 이것을 데이터 파이프라인을 구축한다고 한다 # 데이터 레이크 아키텍쳐 2가지 - 람다 아키텍처 - 배치 레이어: 일괄처리 공간 - 서빙 레이어: 사용자가 사용할 수 있는 데이터들이 모인 공간 - 스피드 레이어: 실시간 분석 공간, 사용자가 바로 데이터 사용도 가능 > 결국 배치레이어와 스피드레이어에 있는 데이터가 서빙 레이어로 모임 - 카파 아키텍쳐 (카프카에서 사용하는 아키텍처) - 스피드 레이어 > 람다 아키텍처에서 배치레이어를 제거한 아키텍쳐 (로직의 파편화, 디버깅, 배포, 운영관리에 대한 이슈로 인해) > `스트림데이터`로 데이터를 처리 (스트림데이터란 시작 데이터와 끝 데이터가 명확히 정해지지 않은 데이터로 배치데이터와 반대이다) > 배치데이터가 스트림데이터로 바라본 관점은 데이터를 log로 바라보는 관점에서 시작한다. > 아직은 데이터 레이크 본래의 의미를 완성하지 못하고 계속해서 발전하고 있다. # 주키퍼란 - 분산 코디네이션 서비스 - 분산 시스템 내부에 상태 정보를 저장하고 데이터를 key/value 저장소로 저장 및 제공하는 서비스 - 일종의 카프카의 설정들을 모아두는 곳 # kafka 힙메모리 설정 - 환경변수로 힙메모리 설정 ```bash vi ~/.bashrc ``` - `.bashrc`에 기입 ```bash export KAFKA_HEAP_OPTS="-Xmx400m -Xms400m" ``` ```bash source ~/.bashrc echo $KAFKA_HEAP_OPTS ``` # 카프카 설정 ```bash vim config/server.properties ``` - `advertised.listeners=PLAINTEXT://:9092`를 `advertised.listeners=PLAINTEXT://localhost:9092`로 수정 # 주키퍼 실행 ```bash bin/zookeeper-server-start.sh -daemon config/zookeeper.properties ``` # 카프카 실행 ```bash bin/kafka-server-start.sh -daemon config/server.properties ``` # 주키퍼와 카프카 실행 확인 - `jps -vm | grep 400` > jvm에 관한 정보 # 카프카 작동확인 (api정보를 가져오기로 작동 확인) ```bash curl https://archive.apache.org/dist/kafka/2.5.0/kafka_2.12-2.5.0.tgz --output kafka.tgz tar -xvf kafka.tgz ``` ```bash bin/kafka-broker-api-versions.sh --bootstrap-server localhost:9092 ``` - api 정보를 가져오면 성공 # 토픽 생성 ```bash bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --topic hello.kafka ``` ```bash bin/kafka-topics.sh \ --create \ --bootstrap-server localhost:9092 \ --topic hello.kafka.2 \ --partitions 3 \ --replication-factor 1 \ --config retention.ms=172800000 \ ``` # 토픽 리스트 조회 ```bash bin/kafka-topics.sh --bootstrap-server localhost:9092 --list ``` # 토픽 상세 조회 ```bash bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic hello.kafka.2 ``` # 토픽 설정 변경 - 파티션 변경 (kafka-topics.sh를 통한 설정) ```bash bin/kafka-topics.sh \ --bootstrap-server localhost:9092 \ --topic hello.kafka \ --alter \ --partitions 4 ``` - 데이터 기간 설정 (kafka-configs.sh를 통한 설정) ```bash bin/kafka-configs.sh \ --bootstrap-server localhost:9092 \ --entity-type topics \ --entity-name hello.kafka \ --alter --add-config retention.ms=86400000 ``` # 토픽에 데이터 넣기 - `kafka-console-producer.sh` - UTF-8을 기반으로 Byte로 변환되고 ByteArraySerializer로만 직렬화 된다.(즉, string 타입으로만 전송을 할 수 있다.) ```bash bin/kafka-console-producer.sh \ --bootstrap-server localhost:9092 \ --topic hello.kafka ``` # 토픽에 있던 데이터 받기 ```bash bin/kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic hello.kafka.2 \ --from-beginning \ --property print.key=true \ --property key.separator="-" \ --group hello-group ``` ```bash kafka-console-consumer.sh \ --bootstrap-server localhost:9092 \ --topic test1 \ --property print.offset=true \ --property print.partition=true \ --property print.headers=true \ --property print.timestamp=true \ --property print.key=true \ --from-beginning ``` # 토픽에 있던 데이터 순서 - 파티션이 여러개이면 데이터의 순서가 보장되지 않는다 - 데이터의 순서를 보장 받으려면 파티션을 1개로 설정하면 된다. # 토픽을 받는 컨슈머 그룹짓기 - 컨슈머 생성시 `--group hello-group`옵션을 달아주면 자동으로 그룹 생성 # 컨슈머 그룹 조회 ```bash bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --list ``` # 파티션 - 파티션안에는 레코드들이 있다. # 컨슈머 그룹 상세보기 ```bash bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 --group hello-group --describe ``` ```bash GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID hello-group hello.kafka.2 1 1 1 0 - - - hello-group hello.kafka.2 0 0 0 0 - - - hello-group hello.kafka.2 2 2 2 0 - - - hello-group hello.kafka 3 4 5 1 - - - hello-group hello.kafka 2 2 2 0 - - - hello-group hello.kafka 1 4 5 1 - - - hello-group hello.kafka 0 1 1 0 - - - ``` - `PARTITION` => 파티션 번호 ( ex) hello.kafka.2는 파티션이 3개이므로 0,1,2) - `CURRENT-OFFSET` => 파티션안에 있는 레코드 번호 ( ex) hello.kafka.2의 파티션1은 레코드가 2개 들어있다) # 카프카의 토픽이 잘 작동하는지 테스트하기 - 토픽만들고 데이터 넣기 ```bash bin/kafka-verifiable-producer.sh --bootstrap-server localhost:9092 \ --max-messages 10 \ --topic verify-test ``` - 토픽 데이터 받기 ```bash bin/kafka-verifiable-consumer.sh --bootstrap-server localhost:9092 \ --topic verify-test \ --group-id test-group ``` # 토픽의 파티션안에 있는 레코드 삭제하기 - `offset-json-file` 생성 ```bash vim delete-topic.json ``` ```json { "partitions": [{ "topic": "hello.kafka", "partition": 0, "offset": 1 }], "version": 1 } ``` - 레코드 삭제 실행 ```bash bin/kafka-delete-records.sh --bootstrap-server localhost:9092 \ --offset-json-file delete-topic.json ``` > 가장 오래된 오프셋부터 지정한 오프셋까지 삭제된다 > 즉, 특정 데이터를 골라 삭제할 수 없다 # 토픽 작명하기 ``` <환경>.<팀-명>.<애플리케이션-명>.<메시지-타입> ex) prd.marketing-team.sms-platform.json ``` - 토픽은 이름변경 지원 X # 프로듀서 프로젝트 만들기 - 자바 프로젝트 생성 - gradle - build.gradle에 추가 ```java /** * kafka clients */ implementation 'org.apache.kafka:kafka-clients:3.2.0' /** * slf4j simple */ implementation 'org.slf4j:slf4j-simple:1.7.36' ``` - SimpleComsumer.java 생성 ```java package com.example; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.Properties; public class SimpleProducer { private final static Logger logger = LoggerFactory.getLogger(SimpleProducer.class); private final static String TOPIC_NAME = "test"; private final static String BOOTSTRAP_SERVERS = "localhost:9092"; public static void main(String[] args) { Properties configs = new Properties(); /** * 서버 설정 */ configs.put( ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS ); /** * key 역직렬화 설정 */ configs.put( ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName() ); /** * value 역직렬화 설정 */ configs.put( ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName() ); KafkaProducer producer = new KafkaProducer<>(configs); String messageValue = "last testMessage"; ProducerRecord record = new ProducerRecord<>(TOPIC_NAME, messageValue); producer.send(record); logger.info("{}", record); producer.flush(); producer.close(); } } ``` # 컨슈머 프로젝트 만들기 - 자바 프로젝트 생성 - gradle - build.gradle에 추가 ```java /** * kafka clients */ implementation 'org.apache.kafka:kafka-clients:3.2.0' /** * slf4j simple */ implementation 'org.slf4j:slf4j-simple:1.7.36' ``` - SimpleConsumer 클래스 생성 ```java import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.Arrays; import java.util.HashMap; import java.util.Map; import java.util.Properties; public class SimpleConsumer { private final static Logger logger = LoggerFactory.getLogger(SimpleConsumer.class); private final static String TOPIC_NAME = "test"; private final static String BOOTSTRAP_SERVERS = "localhost:9092"; private final static String GROUP_ID = "test-group"; public static void main(String[] args) { Properties configs = new Properties(); /** * 서버 설정 */ configs.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS ); /** * 컨슈머 그룹설정 */ configs.put( ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID ); /** * key 역직렬화 설정 * - String */ configs.put( ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName() ); /** * value 역직렬화 설정 * - String */ configs.put( ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName() ); /** * 수동커밋 설정 * - 수동 false * - 자동 true */ configs.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false ); KafkaConsumer consumer = new KafkaConsumer<>(configs); consumer.subscribe(Arrays.asList(TOPIC_NAME)); while (true) { ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); Map currentOffset = new HashMap<>(); for (ConsumerRecord record : records) { logger.info("{}", record); currentOffset.put( new TopicPartition(record.topic(), record.partition()), new OffsetAndMetadata(record.offset() + 1, null) ); /** * 동기적 commit * - 레코드 한개씩 커밋 */ // consumer.commitSync(); /** * 비동기적 commit * - 레코드 한개씩 커밋 */ consumer.commitAsync(new OffsetCommitCallback() { @Override public void onComplete(Map offsets, Exception exception) { if (exception != null) System.err.println("Commit failed"); else System.err.println("Commit succeeded"); if (exception != null) logger.error("Commit failed for offsets {}", offsets, exception); } }); } } } } ``` > 레코드 순서가 필요없으면 비동기적 commit > 레코드 순서가 필요하다면 동기적 commit # 컨슈머 프로젝트 만들기 2 - 파티션 지정하기 => consumer.assign() - 안전한 종료 => consumer.close() ```java import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; import org.apache.kafka.common.serialization.StringDeserializer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.*; public class ExactPartitionConsumer { private final static Logger logger = LoggerFactory.getLogger(SimpleConsumer.class); private final static String TOPIC_NAME = "test"; private final static String BOOTSTRAP_SERVERS = "localhost:9092"; private final static String GROUP_ID = "test-group"; private static final int PARTITION_NUMBER = 0; private static KafkaConsumer consumer; private static Map currentOffset = new HashMap<>(); public static void main(String[] args) { Properties configs = new Properties(); configs.put( ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS ); configs.put( ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID ); configs.put( ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName() ); configs.put( ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName() ); configs.put( ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false ); consumer = new KafkaConsumer<>(configs); /** * 파티션 정하기 * - 리밸런싱이 일어나지 않음 */ consumer.assign(Collections.singleton(new TopicPartition(TOPIC_NAME, PARTITION_NUMBER))); try { while (true) { ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord record : records) { logger.info("{}", record); } consumer.commitSync(); } } catch (WakeupException e) { logger.warn("Wakeup consumer"); } finally { /** * 컨슈머가 종료되었음을 클러스터에게 전달 */ consumer.close(); /** * 안전한 종료 * kill -TERM {프로세스 번호} 호출 */ Runtime.getRuntime().addShutdownHook(new ShutdownThread()); } } static class ShutdownThread extends Thread { public void run() { logger.info("Shutdown hook"); consumer.wakeup(); } } } ``` # 어드민 프로젝트 만들기 - 브로커 정보 조회 - 토픽 정보 조회 ```java public class SimpleKafkaAdminClient { private final static Logger logger = LoggerFactory.getLogger(SimpleKafkaAdminClient.class); public static void main(String[] args) throws ExecutionException, InterruptedException { Properties configs = new Properties(); configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); AdminClient admin = AdminClient.create(configs); logger.info("== Get broker information =="); for (Node node : admin.describeCluster().nodes().get()) { logger.info("node : {}", node); ConfigResource cr = new ConfigResource(ConfigResource.Type.BROKER, node.idString()); DescribeConfigsResult describeConfigs = admin.describeConfigs(Collections.singleton(cr)); describeConfigs.all().get().forEach((broker, config) -> { config.entries().forEach(configEntry -> logger.info(configEntry.name() + "= " + configEntry.value())); }); } logger.info("== Get topic description =="); Map topicDescription = admin.describeTopics(Collections.singleton("hello.kafka")).allTopicNames().get(); logger.info("{}", topicDescription); admin.close(); } } ``` # 카프카 스트림즈 - 이미 만들어 놓은 프로듀서와 컨슈머의 집합 # 카프카 스트림 구조 - 소스 프로세서: 토픽에서 데이터를 가져옴 - 스트림 프로세서: 데이터 처리 - 싱크 프로세서: 토픽 저장 # 카프카 스트림 API 2가지 - 스트림즈DSL - 프로세서 API # 스트림즈 DSL - KStream: 토픽의 모든 레코드 조회 - KTable: 레코드 업데이트 - GlobalKTable: KStream과 KTable 데이터 조인 > GlobalKTable은 각 태스크마다 모든 데이터가 저장되므로 소규모 용량을 작업할 때 사용 > 대용량은 KTable # 프로세서 API - stream(): 특정 토픽을 KStream형태로 가져오기 - to(): KStream의 데이터를 특정 토픽으로 저장 # 스트림즈 DSL 만들기1 - stream()과 to()사용 > filter 없이 ```java public class SimpleStreamsApplication { private static String APPLICATION_NAME = "streams-application"; private static String BOOTSTRAP_SERVERS = "localhost:9092"; private static String STREAM_LOG = "stream_log"; private static String STREAM_LOG_COPY = "stream_log_copy"; public static void main(String[] args) { /** * KafkaStreams 환경설정 */ Properties props = new Properties(); props.put( StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_NAME ); props.put( StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS ); props.put( StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass() ); props.put( StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass() ); StreamsBuilder builder = new StreamsBuilder(); /** * topic data 불러오기 */ KStream streamLog = builder.stream(STREAM_LOG); /** * topic data 저장하기 */ streamLog.to(STREAM_LOG_COPY); KafkaStreams streams = new KafkaStreams(builder.build(), props); /** * kafka stream 시작 */ streams.start(); } } ``` # 스트림즈 DSL 만들기2 ```java public class StreamFilter { private static String APPLICATION_NAME = "streams-filter-application"; private static String BOOTSTRAP_SERVERS = "localhost:9092"; private static String STREAM_LOG = "stream_log"; private static String STREAM_LOG_FILTER = "stream_log_filter"; public static void main(String[] args) { Properties props = new Properties(); props.put( StreamsConfig.APPLICATION_ID_CONFIG, APPLICATION_NAME ); props.put( StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS ); props.put( StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass() ); props.put( StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass() ); StreamsBuilder builder = new StreamsBuilder(); /** * stream */ KStream streamLog = builder.stream(STREAM_LOG); /** * filter * - value 값이 5이상인 것만 거르기 */ KStream filterStream = streamLog.filter((key, value) -> value.length() > 5); /** * to */ filterStream.to(STREAM_LOG_FILTER); /** * start */ KafkaStreams streams = new KafkaStreams(builder.build(), props); streams.start(); } } ```

게시글 상세