Kafka 빠른 시작하기

Kafka 빠른 시작하기

"Kafka가 뭔지는 알겠는데, 막상 돌려 보려니 어디서부터 시작해야 하지?" 하는 분들을 위한 페이지예요. 제일 먼저 토픽을 만들고, 프로듀서로 이벤트를 쓰고, 컨슈머로 다시 읽는 것까지 그 흐름을 한 번에 잡아볼게요. 최신 릴리스(이 페이지 기준 4.3.1)를 받아서 시작합니다.

출처: Apache Kafka — Quickstart

1단계: Kafka 받기

최신 릴리스를 다운로드한 뒤 압축을 풀어요.

$ tar -xzf kafka_2.13-4.3.1.tgz
$ cd kafka_2.13-4.3.1

2단계: Kafka 환경 시작하기

로컬 환경에 **Java 17+**가 설치되어 있어야 해요. Kafka는 다운로드 파일의 로컬 스크립트 또는 도커 이미지로 실행할 수 있어요.

다운로드 파일로 실행

먼저 클러스터 UUID를 만들고,

$ KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

로그 디렉터리를 포맷한 뒤,

$ bin/kafka-storage.sh format --standalone -t $KAFKA_CLUSTER_ID -c config/server.properties

카프카 서버를 시작해요.

$ bin/kafka-server-start.sh config/server.properties

서버가 성공적으로 뜨면 기본 Kafka 환경이 실행 중인 상태예요.

JVM 기반 도커 이미지로 실행

$ docker pull apache/kafka:4.3.1
$ docker run -p 9092:9092 apache/kafka:4.3.1

GraalVM 네이티브 도커 이미지로 실행

$ docker pull apache/kafka-native:4.3.1
$ docker run -p 9092:9092 apache/kafka-native:4.3.1

3단계: 이벤트를 저장할 토픽 만들기

Kafka는 분산 이벤트 스트리밍 플랫폼이라서, 여러 대의 머신에 걸쳐 이벤트(문서에서는 record·message라고도 함)를 읽고 쓰고 저장하고 처리할 수 있어요. 결제 거래, 모바일 위치 업데이트, 배송 주문, IoT 센서 측정값 등이 다 이벤트가 돼요. 이 이벤트들은 토픽에 모여 저장되는데, 파일시스템의 폴더와 이벤트가 그 안의 파일이라고 비유하면 쉬워요.

첫 이벤트를 쓰기 전에 토픽을 만들어야 해요. 새 터미널을 열고 실행하세요.

$ bin/kafka-topics.sh --create --topic quickstart-events --bootstrap-server localhost:9092

Kafka의 커맨드라인 도구는 옵션이 많아요. 인자 없이 kafka-topics.sh를 실행하면 사용법이 나오고, 아래처럼 --describe로 새 토픽의 파티션 수 같은 상세 정보도 볼 수 있어요.

$ bin/kafka-topics.sh --describe --topic quickstart-events --bootstrap-server localhost:9092
Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1       ReplicationFactor: 1	Configs:
Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

기본으로 파티션 1개, 복제 팩터 1로 만들어져요. 파티션 수·복제 팩터를 바꾸고 싶으면 kafka-topics.sh --create에 --partitions와 --replication-factor 옵션을 주면 돼요.

4단계: 토픽에 이벤트 쓰기

Kafka 클라이언트는 네트워크로 브로커와 통신하며 이벤트를 쓰고 읽어요. 브로커는 받은 이벤트를 오래도록, 필요하다면 영원히 내구성 있고 내결함성 있게 저장해요.

콘솔 프로듀서 클라이언트로 토픽에 몇 개 이벤트를 써 볼게요. 기본적으로 엔터 한 줄이 하나의 이벤트가 되어 토픽에 쓰여요.

$ bin/kafka-console-producer.sh --topic quickstart-events --bootstrap-server localhost:9092
>This is my first event
>This is my second event

프로듀서는 언제든 Ctrl-C로 멈출 수 있어요.

5단계: 이벤트 읽기

새 터미널을 열고 콘솔 컨슈머 클라이언트를 실행해 방금 만든 이벤트를 읽어볼게요.

$ bin/kafka-console-consumer.sh --topic quickstart-events --from-beginning --bootstrap-server localhost:9092
This is my first event
This is my second event

--from-beginning은 토픽의 처음부터 다 읽겠다는 뜻이에요. 이 컨슈머도 Ctrl-C로 멈출 수 있어요.

이벤트는 Kafka에 내구성 있게 저장되므로 원하는 만큼 여러 번, 여러 컨슈머가 읽을 수 있어요. 또 다른 터미널을 열어 같은 명령을 다시 실행하면 같은 결과가 나오는 걸로 직접 확인할 수 있어요.

6단계: Kafka Connect로 데이터 흐르게 하기

관계형 데이터베이스나 전통적인 메시징 시스템 같은 기존 시스템에 데이터가 이미 많을 거예요. Kafka Connect는 외부 시스템에서 Kafka로, Kafka에서 외부 시스템으로 데이터를 계속 주입(import/export)하게 해 줘요. 커넥터(connector)를 운영하는 확장형 도구이고, 외부 시스템과 통신하는 커스텀 로직은 커넥터가 담당해요. 덕분에 기존 시스템과 Kafka를 통합하기가 아주 쉽고, 이미 만들어진 커넥터도 수백 개 있어요.

이 퀵스타트에서는 파일에서 토픽으로, 토픽에서 파일로 옮기는 간단한 커넥터를 돌려 볼게요. 먼저 Connect 워커 설정의 plugin.path에 connect-file-4.3.1.jar를 추가해야 해요. 퀵스타트에서는 상대 경로로 uber jar 하나를 쓰는데, 운영 배포에서는 절대 경로를 쓰는 편이 좋아요.

config/connect-standalone.properties에 아래 설정을 추가·수정해서 저장해요.

$ echo "plugin.path=libs/connect-file-4.3.1.jar" >> config/connect-standalone.properties

테스트용 시드 데이터를 만들고,

$ echo -e "foo
bar" > test.txt

Windows에서는 이렇게 해요.

$ echo foo > test.txt
$ echo bar >> test.txt

이제 두 커넥터를 standalone 모드로 실행해요. standalone 모드는 단일 로컬 전용 프로세스에서 도는 방식이에요. 설정 파일을 세 개 인자로 주는데, 첫 번째는 항상 Kafka Connect 프로세스 설정(연결할 브로커, 직렬화 포맷 등)이고, 나머지는 각각 만들 커넥터 하나씩의 설정이에요.

$ bin/connect-standalone.sh config/connect-standalone.properties config/connect-file-source.properties config/connect-file-sink.properties

이 샘플 설정은 두 커넥터를 만들어요. 첫 번째는 소스 커넥터로 입력 파일의 줄을 읽어 Kafka 토픽에 produce하고, 두 번째는 싱크 커넥터로 토픽의 메시지를 읽어 출력 파일에 줄로 작성해요.

시작 로그에 커넥터가 생성되는 메시지가 보일 거예요. 그다음 소스 커넥터가 test.txt의 줄을 읽어 connect-test 토픽으로, 싱크 커넥터가 connect-test의 메시지를 test.sink.txt 파일로 내보내요. 출력 파일 내용으로 파이프라인이 제대로 도는지 확인해요.

$ more test.sink.txt
foo
bar

데이터는 Kafka 토픽 connect-test에 저장되므로 콘솔 컨슈머로도 확인할 수 있어요.

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning
{"schema":{"type":"string","optional":false},"payload":"foo"}
{"schema":{"type":"string","optional":false},"payload":"bar"}

커넥터는 계속 데이터를 처리하므로 파일에 데이터를 추가해 보면 파이프라인을 따라 움직이는 게 보여요.

$ echo "Another line" >> test.txt

콘솔 컨슈머 출력과 싱크 파일에 해당 줄이 나타나는 걸 확인할 수 있어요.

7단계: Kafka Streams로 이벤트 처리하기

이벤트가 토픽에 저장됐다면, 이제 Kafka Streams 클라이언트 라이브러리(Java/Scala)로 데이터를 처리할 수 있어요. 미션 크리티컬한 실시간 애플리케이션·마이크로서비스를 구현하는 데 쓰이고, 입출력 데이터가 모두 Kafka 토픽에 저장돼요. Kafka Streams는 클라이언트 쪽에서 표준 Java/Scala 앱을 쓰고 배포하는 단순함과, Kafka의 서버-사이드 클러스터 기술의 장점을 결합해 확장성·탄력성·내결함성·분산성을 확보해요. 정확히-한-번 처리, 상태 저장 연산·집계, 윈도잉, 조인, 이벤트-시간 기반 처리를 지원해요.

대표적인 WordCount 알고리즘을 이런 식으로 구현해요.

KStream<String, String> textLines = builder.stream("quickstart-events");

KTable<String, Long> wordCounts = textLines
            .flatMapValues(line -> Arrays.asList(line.toLowerCase().split(" ")))
            .groupBy((keyIgnored, word) -> word)
            .count();

wordCounts.toStream().to("output-topic", Produced.with(Serdes.String(), Serdes.Long()));

8단계: 환경 종료하기

1. 프로듀서·컨슈머 클라이언트를 Ctrl-C로 멈추고,
2. Kafka 브로커를 Ctrl-C로 멈춰요.

로컬 환경 데이터를 지우고 싶다면 아래 명령을 실행해요.

$ rm -rf /tmp/kafka-logs /tmp/kraft-combined-logs

더 알아보기

• Kafka가 높은 수준에서 어떻게 동작하는지, 주요 개념과 다른 기술과의 비교는 Introduction을 읽어 보세요.
• 사용 사례에서 커뮤니티가 Kafka를 어떻게 활용하는지 확인할 수 있어요.