개발자 노트

[Spark] Structured Streaming checkpoints

1. checkpoints가 필요한 이유Structured Streaming에서 처리한 데이터의 신뢰성과 복구를 보장하기 위해 필요하다. Yarn Cluster의 장애로 인해 정상 동작 중이던 spark streaming이 비정상 종료되었다고 하자. 이 때, 어디부터 데이터 처리를 다시 해야하는지 판단해야 하는데 checkpoints에 기록된 메타 데이터를 확인한다. checkpoints를 지정하는 방법은 아래와 같이 writeStream에서 option으로 지정한다.df.writeStream .option("checkpointLocation", "/Volumes/catalog/schema/volume/path")microBatch 마다 처리한 결과를 해당 디렉토리(혹은 hdfs)에 저장한다.check..

이번 글에서는 Kafka Connect를 쿠버네티스(이하 K8s)에서 구축하기 위해 도커 이미지를 만드는 방법에 대해서 설명하려고 한다. 이 말을 들었을 때, "왜 이미지를 따로 만들지?"라고 생각할 수 있다. 왜냐하면 컨플루언트에서 제공하는 도커 이미지가 있기 때문이다. 하지만 필자는 직접 이미지를 만들기로 했다. 이유는 Kafka Connect 내에 따로 개발한 Connector들을 빌드할 때, 해당 Connector들을 포함한 Kafka Connect 도커 이미지를 만들고 싶었다. 그리고 현재 운영 중인 환경과 동일한 버전으로 구성된 도커 이미지를 만들어서 환경 차이에 따라 혹시 발생할지 모르는 문제를 피하고 싶었기 때문이다. 그럼 도커 이미지를 만드는 내용에 들어가기 앞서 왜 쿠버네티스 환경에서 ..

이번 글에서는 schema registry를 HA로 설정하는 방법에 대해서 설명한다. 하지만 HA를 구성하기 위한 설정은 생각보다 단순하기 때문에 schema registry에 대해서 대략적으로 소개한다. 그리고 설치가 완료된 schema registry를 사용해서 카프카에 avro 포맷으로 데이터를 보내고 읽는 kafka-avro-console-producer와 kafka-avro-console-consumer를 사용해서 테스트하는 방법까지 해보자. 1. 개요 먼저 schema registry에 대해서 간략히 설명한다. 한 문장으로 설명하면 카프카에 기록할 메시지의 데이터 스키마를 관리하고 검증하는 서비스이다. 스키마를 관리하고 검증한다고 표현했는데 관리는 스키마를 기록하고 유지하는 역할이며, 검증은 ..

이번 글에서는 producer에서 데이터를 보낼 때, 데이터의 토픽 파티션 분배에 대해서 살펴봅니다. 특히 Key값이 정해지지 않았을 때 파티션 분배가 어떤 로직으로 이루어지는지 분석합니다. 카프카의 버전이 올라가면서 파티션 분배의 단점들을 개선하는 시도들이 이루어지는데 초기 버전에서부터 시작하여 현재의 분배 로직이 이루어진 과정에 대해 소개합니다. 1. 개요 (파티션) 카프카를 써 보신 분들이라면 파티션의 존재 이유에 대해서 아실 것입니다. 카프카는 분산 시스템으로 대규모의 데이터 처리를 위해 여러 파티션을 두고 데이터들을 분산해서 저장합니다. 파티션 분배 로직이 카프카 브로커에 존재하지 않나?라고 생각하실 수도 있는데, 파티션 분배 로직은 producer 내부에 존재합니다. 그리고 producer의..

이번 글에서는 spark 관리자 역할에서 사용하고 있는 spark 버전 및 설정 정보를 수집하여 dashboard 형태로 모니터링하는 방법에 대해서 공유한다. 이런 아이디어를 생각하게 된 이유는 다른 사용자에 의해서 사용되는 spark 버전들을 관리하고 최신의 버전으로 가이드하기 위해서는 먼저 현황 파악이 우선되어야 했기 때문이다. 사실 spark는 event log를 적재하여 history server를 통해 spark job의 실행 결과를 분석할 수 있다. 문제는 history server의 UI에서는 전체적인 현황을 파악하기 보다는 한 개의 job에 대한 분석을 하는데 더 용이하다는데 있다. 그래서 일부 spark 옵션에 대해서만 수집해서 spark job들의 옵션들을 통합하고 통계화하면 spark..

ksqlDB 공식 문서를 보면 concepts 항목에 ksqlDB를 이해하기 위한 주요 개념들을 설명하고 있다. 이 글을 concepts 공식 문서의 내용을 간략하게 요약했다(글에 포함된 사진은 ksqlDB 공식 문서에서 가져왔다). 전체적인 개념만 파악하는 용도로 이 글을 활용하고 자세한 사항은 ksqlDB의 공식문서를 읽어보자. 1. ksqlDB란? 전통적인 RDB를 활용한 방식이 아닌 Event 처리 관점으로 어플리케이션을 개발하기 위한 목적으로 만들어졌다. Event는 시간의 흐름 안에서 특정 시점에 발생하고 기록된 데이터이다. ksqlDB는 Event를 쉽게 사용하게 한다는 목적을 지향한다. 그리고 Event의 저장 및 처리 기능을 카프카 플랫폼에 두고 그 위에 추상화하는 형태로 만들어졌다. E..

ksqlDB에서 HA(High Availability)를 무슨 작업을 위해서 존재할까? HA는 한글로 고가용성이라는 명칭으로 해석된다. 고가용성은 클러스터로 묶여 있는 서버들 중에 일부가 다운되더라도 서비스에 영향이 없어야 한다. ksqlDB에서의 고가용성은 pull query에서 필요하다. pull query에서의 고가용성을 위해 materialized state를 다른 서버에 복제해둬야 한다. materialized state의 복제를 위해서 ksqlDB의 기본 설정에서 변경해야할 값이 있다. 1. ksqlDB 클러스터 서버들의 내부 통신 listeners=http://0.0.0.0:8088 ksql.advertised.listener=http://host1.example.com:8088 기본 설정에..

이번에는 spark 빌드 관련해서 이야기를 하려고 합니다. spark은 대부분 hadoop yarn 기반에서 실행하는 경우가 많습니다. 하지만 hadoop cluster의 버전이 회사의 환경에 따라 다르므로 빌드하는 spark 버전에 따라 호환성을 잘 살펴야 합니다. 필자는 낮은 버전의 hadoop을 사용하고 있는데, 사용 하는 hadoop이 spark의 어느 버전까지 빌드해서 사용 가능한지 검토하게 됐습니다. 이 글은 그런 경험을 토대로 정리하는 목적으로 쓰게 됐습니다. 실행 가능한 spark 배포판을 빌드할 때, 사용하는 hadoop 버전을 명시해서 빌드를 합니다. 예를 들어 hadoop 3.3.0 버전을 사용할 때 아래와 같이 옵션을 넣어서 spark 배포판을 빌드합니다. ./build/mvn -P..

2022년 4월 14일. Kafka 한국 사용자 모임 Virtual Meetup에서 발표를 했습니다. 주제는 MirrorMaker2 소개 입니다. MirrorMaker2를 처음 들어보시는 분들에 맞춰서 기초적인 내용을 포함하여 자료를 준비했습니다. MirrorMaker2의 아키텍처, 활용, 모니터링, 운영 팀의 목차로 구성되었습니다. Kafka 한국 사용자 모임 Github에도 올라갈 예정이나 먼저 블로그에 자료 공유합니다.

이번 글에서는 KafkaAvroDeserializer를 사용한 컨슈머에 대해서 이야기하려고 한다. KafkaAvroDeserializer를 사용했다는 것은 Schema-Registry(이하 스키마 레지스트리)를 사용했다는 의미이다. 스키마 레지스트리를 간단히 설명하면 Confluent에서 제공한 Schema 관리 시스템이다. 카프카로 들어오는 레코드들의 버전 관리가 필요할 때 스키마 레지스트리를 사용한다. 도입부에 많은 용어들을 나열했는데 기본적인 내용을 먼저 정리해야 이 글에서 이야기할 내용이 전달될 수 있을 듯 하다. 다음 용어들의 개념을 간단히만 살펴보고(자세한 내용은 링크 추가) 이해를 돕고자 한다. AVRO(이하 에이브로) 스키마 스키마 레지스트리 1. 에이브로 에이브로는 스키마, 스키마 레지스..

MirrorMaker2(이하 MM2)는 띄우는데는 여러 방법이 있다. 분산 환경으로 한정하면 2가지 방법이 있다. 그것은 MM2 전용 클러스를 띄우거나 기존 Connect 클러스터에 MM2를 추가하는 것이다. 기존 Connect 클러스터에는 MM2를 추가한다는 표현을 썼는데, MM2가 Source Connector를 구현한 방식으로 개발되었기 때문이다. 이번 글에서는 기존 Connect 클러스터에 MM2 관련 Source Connector를 추가하는 방법에 대해서 설명한다. 1. Connect 클러스터를 활용하는 이유 본격적인 활용 방법에 대해서 설명을 하기 전에 기존 Connect 클러스터를 활용하는 이유에 대해서 설명하고자 한다. 사실 가장 큰 이유는 기존에 운영 중인 Conenct 클러스터가 있다면..

이번글에서는 Connect의 에러처리와 관련하여 정리한다. 에러처리와 관련하여 주요한 업데이트가 2.0과 2.6 버전에서 있었다. 2.6 버전에서의 추가 기능은 2.0의 기능 확장이라고 할 수 있는데, 자체적으로 개발한 plugin의 에러처리에 활용된다. 먼저 2.0 버전에서 처음 나온 개념인 dead queue letter에 대해 알아보자. 1. dead letter queue dead letter queue는 이름에서 유추할 수 있듯이 실패한 레코드를 보관하는 별도의 큐이다. 카프카는 dead letter queue로 원천 데이터가 보관된 토픽이 아닌 별도의 토픽으로 설정한다. 그리고 실패한 레코드의 메타정보도 포함시켜 저장한다. 아래 [그림 1]을 보면 쉽게 이해할 수 있다. 원천 토픽이 레코드 중..