스파크 스트리밍과 같은 것을 이용하기 위해서는, 입력(Input) 도구로 아파치 카프카(Apache Kafka)를 많이 쓴다. 이번 포스팅에는 그 아파치 카프카를 설치하고, 간단한 예제를 실행시켜보는 것까지 하겠다.
시스템 환경은
Macbook Air OS X EI Capitan version 10.11.6 이다.
1. 카프카 다운로드
아파치 카프카 공식 홈페이지에 가면, 다운로드 페이지(https://kafka.apache.org/downloads)에서 다운로드 받을 수 있다.
나의 경우 Binary로 Scala 2.11 버전을 다운받았다.
다운로드 후, 압축을 풀면 끝.
2. 카프카 브로커(서버) 실행
카프카는 주키퍼(zookeeper)를 사용한다. 만약 사용하는 시스템에 현재 주키퍼가 따로 없다면, 내장된 주키퍼 서버를 실행시키면 된다. 설치한 카프카에는 패키지화된 단일 노드 주키퍼가 포함되어 있기 때문에 쉘스크립트 실행으로 주키퍼 인스턴스를 얻을 수 있다.
(주키퍼에 대한 설명은 여기를 참고하자)
2-1. 주키퍼 서버 실행
카프카를 설치한 홈에서 아래의 내용을 입력하면 주키퍼 서버가 실행된다.
bin/zookeeper-server-start.sh config/zookeeper.properties실행이 제대로 됐다면, 이런 저런 내용이 출력되다가 마지막에 아래와 같은 내용이 출력될 것이다.
INFO binding to port 0.0.0.0/0.0.0.0:2181 (org.apache.zookeeper.server.NIOServerCnxnFactory)
2-2. 카프카 서버 (Broker) 실행
새로운 터미널을 하나 더 띄워서, 아래와 같은 내용을 입력한다.
bin/kafka-server-start.sh config/server.properties마찬가지로, 이런 저런 내용이 출력되다가 마지막에 아래와 같은 내용이 출력되면 정상이다.
INFO [Kafka Server 0], started (kafka.server.KafkaServer)카프카 서버의 기본 포트는 9092번 이다. 만약에 필요에 따라서 변경을 하고 싶으면, config 디렉토리에 있는 설정 파일 중 일부를 수정하면 될 것이다.
3. 토픽(topic) 생성
아시다시피, 카프카는 기본적으로 토픽을 기반으로 작동한다.
그래서 카프카가 잘 동작하는지 확인하기 위해선, 일단 토픽을 생성해야한다.
새로운 터미널을 하나 더 생성해서, 아래와 같이 입력해보자.
bin/kafka-topics.sh -create -zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
정상적으로 토픽이 생성되었다면, Create topic "test". 라는 문구가 나타나는 것을 확인할 수 있다.
그리고, 생성한 토픽 리스트를 보거나, 토픽을 제거하기 위해서는 아래와 같이하면 된다.
*생성된 topic list를 확인 bin/kafka-topics.sh --list --zookeeper localhost:2181 test *생성된 topic 삭제 bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic test Topic test is marked for deletion. Note: This will have no impact if delete.topic.enable is not set to true.
토픽을 생성하는 과정에서 입력하는, replication과 partition에 대해서 조금 더 알아보자면,
카프카에서 토픽은 파티션(partition)이라고 하는 단위로 나누어지며, 클러스터의 각 서버들에 분산되어 저장된다.
복제(replication)는 고가용성을 위해 설정하는데, 데이터를 각 서버들(즉, 다른 파티션들)에 저장을 해놓고, 장애가 발생했을때 복제해놓은 데이터를 이용해서 fail over을 하게 된다. 이때 이 fail over는 파티션 단위로 이루어진다.
카프카 클러스터는 각 토픽에 대해 아래 그림과 같은 파티션 구성을 유지한다.
(아래 그림의 상황은 1개의 토픽이 3개의 파티션으로 분산되어 저장되는 것을 나타낸다)
각 파티션은 0부터 1씩 증가하는 오프셋(offset)을 각 데이터에 부여한다. 이 오프셋은 파티션별로 관리되며, 데이터를 식별하는 ID로 사용된다. 그래서, 토픽 내에서 데이터를 식별할 때 파티션 번호와 오프셋 값을 이용하여 식별할 수 있다.
4. 메세지 전송
카프카는 Producer와 Consumer를 통해 데이터를 주고 받는 방식이기 때문에,
Producer를 통해 파일이나 표준 입력에서 입력을 받고, 그것이 카프카 클러스터로 보내지면,
Consumer가 받은 내용을 처리한다.
참고로, 기본적으로 각 줄은 별도의 메세지로 보내진다.
이번 예제에서는 표준 입력을 받아서 메세지를 전달하는 과정을 보여주지만,
카프카에서는 파일 내용(log 파일과 같은..)을 전달할 수도 있다.
4-1. Producer 실행 및 메세지 전송
Producer는 다음과 같이 실행시킬 수 있다.
내용은 브로커로 메세지를 보내는데, test라는 토픽을 구독하는 Consumer에게 보내겠다는 것이다.
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
4-2. Consumer 실행
Consumer는 다음과 같이 실행시킬 수 있다.
내용은 test라는 토픽에 해당하는 데이터가 들어오면 받겠다는 것이다.
bin/kafka-console-consumer.sh -zookeeper localhost:2181 --topic test --from-beginning
이렇게, Producer와 Consumer를 실행시킨 후,
Producer에 내용을 입력하면, Consumer쪽에 그 내용이 들어온다는 것을 확인할 수 있다.
Producer 화면
Consumer 화면
끝!
출처
[1] http://hellowuniverse.com/2017/04/26/kafka-spark-streaming-with-python%EC%9C%BC%EB%A1%9C-%EC%8B%A4%EC%8B%9C%EA%B0%84-%EB%B6%84%EC%84%9D%EC%8B%9C%EC%8A%A4%ED%85%9C-%EB%A7%8C%EB%93%A4%EA%B8%B0/
댓글