DEFAULT_KEY_SERDE_CLASS_CONFIG in Kafka

The DEFAULT_KEY_SERDE_CLASS_CONFIG is a configuration property in Apache Kafka Streams. It specifies the default serializer/deserializer (Serde) to be used for the keys of messages in your Kafka Streams application.

What Is a Serde?

A Serde is a combination of a serializer and a deserializer:

  • Serializer: Converts objects into a byte array to send over Kafka.
  • Deserializer: Converts the byte array back into an object.

Kafka Streams requires Serdes to know how to handle data when reading from and writing to topics.

Usage of DEFAULT_KEY_SERDE_CLASS_CONFIG

This property allows you to specify the default Serde class for the key of messages processed by Kafka Streams. For example, if your keys are strings, you would set the default key Serde to Serdes.String().

Example Configuration in Java:

Properties props = new Properties();
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

Example Configuration in application.yml:

spring:
  kafka:
    streams:
      properties:
        default.key.serde: org.apache.kafka.common.serialization.Serdes$StringSerde
        default.value.serde: org.apache.kafka.common.serialization.Serdes$StringSerde

Common Serdes in Kafka Streams

Kafka Streams provides built-in Serdes for common data types:

  • StringSerde: For string keys/values.
    org.apache.kafka.common.serialization.Serdes.StringSerde
  • IntegerSerde: For integer keys/values.
    org.apache.kafka.common.serialization.Serdes.IntegerSerde
  • LongSerde: For long keys/values.
    org.apache.kafka.common.serialization.Serdes.LongSerde
  • DoubleSerde: For double keys/values.
    org.apache.kafka.common.serialization.Serdes.DoubleSerde
  • ByteArraySerde: For raw binary data.
    org.apache.kafka.common.serialization.Serdes.ByteArraySerde

Why Use DEFAULT_KEY_SERDE_CLASS_CONFIG?

  1. Convenience: Sets a global default, so you don't need to specify the Serde for every stream/table.
  2. Consistency: Ensures that all keys are handled the same way unless overridden.

Overriding the Default Serde

If you need to process data differently for a specific stream, you can override the default Serde when defining the stream:

KStream<Integer, String> stream = builder.stream(
    "source-topic",
    Consumed.with(Serdes.Integer(), Serdes.String()) // Custom key/value Serdes
);

Let me know if you'd like to explore custom Serdes!

댓글

댓글 쓰기

이 블로그의 인기 게시물

Spring Boot with Kafka Streams

Using Spring Boot with Kafka Streams simplifies the process further. Spring Boot provides excellent support for Kafka through Spring Kafka , allowing seamless integration and management. Here’s a step-by-step guide to building a Kafka Streams application with Spring Boot: 1. Setup Your Spring Boot Application Add Dependencies Include the necessary dependencies in your pom.xml : < dependency > < groupId > org.springframework.boot </ groupId > < artifactId > spring-boot-starter </ artifactId > </ dependency > < dependency > < groupId > org.springframework.kafka </ groupId > < artifactId > spring-kafka </ artifactId > </ dependency > < dependency > < groupId > com.google.code.gson </ groupId > < artifactId > gson </ artifactId > < version > 2.8.9 </ version > <!-- Version may vary --> </ dependency > 2. Configure Ka...
자세한 내용 보기

Install and run an FTP server using Docker

To install and run an FTP server using Docker, follow these steps. We’ll use the popular stilliard/pure-ftpd image, which is a lightweight and widely used FTP server. Step 1: Install Docker Make sure Docker is installed on your machine. If it isn’t, install it using the instructions below: Ubuntu/Debian: sudo apt update sudo apt install docker.io -y Mac: Install Docker Desktop from Docker's website . Windows: Install Docker Desktop from Docker's website . Verify Docker is installed: docker --version Step 2: Pull the FTP Server Docker Image Use the stilliard/pure-ftpd image, which is a simple and effective FTP server. docker pull stilliard/ pure -ftpd Step 3: Run the FTP Server Run the FTP server container using the following command: docker run -d --name ftp-server \ - p 21 : 21 -p 30000 - 30009 : 30000 - 30009 \ - e FTP_USER_NAME=testuser \ - e FTP_USER_PASS=testpass \ - e FTP_USER_HOME= /home/testuser \ stilliard/pure-ftpd Explanation...
자세한 내용 보기

kubeadm init

The kubeadm init command is used to bootstrap the first control plane node in a Kubernetes cluster. Here's how you can initialize a control plane node step-by-step: 1. Prerequisites Ensure the Following: Operating System : Use a supported Linux distribution (e.g., Ubuntu, CentOS, etc.). Container Runtime : Install a supported container runtime like containerd or CRI-O. kubeadm, kubelet, kubectl Installed : Install these tools using your package manager. apt update && apt install -y kubeadm kubelet kubectl Or on CentOS: yum install -y kubeadm kubelet kubectl Swap Disabled : swapoff - a sed - i '/swap/d' /etc/fstab Firewall Ports Open : Allow necessary Kubernetes ports such as 6443, 2379-2380, 10250, 10259, 10257. For example: sudo ufw allow 6443 /tcp sudo ufw allow 2379:2380 /tcp sudo ufw allow 10250:10259 /tcp 2. Run kubeadm init to Bootstrap the First Control Plane Basic Command Run the following command on the first control plane node: ...
자세한 내용 보기

Random numbers ranged 1 to 10 in Python

To get a random number between 1 and 10 in Python, you can use the randint function from the random module: import random # Generate a random integer between 1 and 10 (inclusive) random_number = random .randint( 1 , 10 ) print (random_number) Explanation random.randint(1, 10) returns a random integer in the range [1, 10], where both 1 and 10 are inclusive. This will generate a new random number each time you run the code.
자세한 내용 보기

Options for auto.offset.reset in Kafka

The default value of the auto.offset.reset configuration in Apache Kafka depends on the context in which it is used. This configuration determines what the consumer should do when there is no initial offset in Kafka or if the current offset does not exist anymore (e.g., due to log cleanup). Default Value latest is the default value for auto.offset.reset . Options for auto.offset.reset earliest : The consumer starts reading from the beginning of the log (the earliest offset). Useful for reprocessing data from the start. latest (default): The consumer starts reading from the end of the log (the latest offset). It reads only new messages produced after the consumer has started. none : Throws an exception if no offset is found for the consumer's group. Typically used when you want strict control over offsets. Example Configuration Java Code Properties props = new Properties(); props . put (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"...
자세한 내용 보기

경량 리눅스 배포판

경량 리눅스 배포판은 저사양 PC나 특정 목적에 최적화된 환경을 제공하기 위해 설계된 리눅스 배포판입니다. 이런 배포판은 하드웨어 자원을 최소로 사용하면서도 기본적인 기능을 유지하거나 특정 용도에 맞춘 효율성을 제공합니다. 아래는 주요 경량 리눅스 배포판과 그 특징을 정리한 목록입니다. 1. Puppy Linux 특징 : 설치 크기: 약 300MB RAM 사용량: 256MB 이상에서 실행 가능. Live USB로 실행 가능하며, 설치 없이 사용 가능. 빠른 부팅 속도와 간단한 사용자 인터페이스 제공. 용도 : 저사양 PC 복원용, 라이브 디스크로 활용. 홈페이지 : puppylinux.com 2. Tiny Core Linux 특징 : 설치 크기: 약 16MB~20MB. 매우 최소화된 운영체제, 필요한 구성 요소만 추가 설치 가능. GUI를 포함하나, 기본 애플리케이션은 거의 없음. 용도 : 초소형 환경, 임베디드 시스템, DIY 프로젝트. 홈페이지 : tinycorelinux.net 3. Lubuntu 특징 : 우분투 기반, LXQt 데스크톱 환경 사용. 가벼운 GUI와 함께 친숙한 사용자 경험 제공. 비교적 최신 소프트웨어와 우분투의 패키지 생태계를 활용 가능. 최소 요구 사항: 1GB RAM, 32비트 및 64비트 지원. 용도 : 저사양 PC, 오래된 노트북에서 일반적인 데스크톱 환경 제공. 홈페이지 : lubuntu.me 4. Linux Lite 특징 : 설치 크기: 약 1GB. XFCE 데스크톱 환경을 사용하여 가벼우면서도 친숙한 Windows 유사 환경 제공. 초보자 친화적이며 문서화가 잘 되어 있음. 최소 요구 사항: 1GHz CPU, 768MB RAM. 용도 : 저사양 PC에서 일반적인 데스크톱 작업. 홈페이지 : linuxliteos.com 5. MX Linux (Fluxbox Edition) 특징 : MX Linux의 경량 ...
자세한 내용 보기

Scan an HBase table with a prefix filter

To scan an HBase table with a prefix filter , you can either use the HBase shell , Java API , or Python via HappyBase . The prefix filter helps to retrieve only the rows whose row keys start with a specific prefix . Below are different ways to perform this operation. 1. Using HBase Shell In the HBase shell , you can use the scan command along with a filter expression to scan rows based on a prefix. Example: hbase (main) : 001 : 0 > scan 'table_name' , {FILTER => "PrefixFilter('prefix_value')" } Explanation: Replace 'table_name' with the name of your HBase table. Replace 'prefix_value' with the prefix you want to search for in the row keys. This filter ensures that only rows whose row keys start with the given prefix are retrieved. 2. Using Java API (HBase Client) If you're working in Java, you can use the PrefixFilter class from the HBase API to scan the table. Example: import org.apache.hadoop.hbase.client.*;...
자세한 내용 보기

MAX_POLL_RECORDS_CONFIG in Kafka

The default value of MAX_POLL_RECORDS_CONFIG in Apache Kafka is: 500 What is MAX_POLL_RECORDS_CONFIG ? The MAX_POLL_RECORDS_CONFIG configuration specifies the maximum number of records a Kafka consumer can retrieve in a single call to the poll() method. Key Details: Default Behavior : If not configured, the consumer will fetch up to 500 records per poll. This is a batch size limit and helps in controlling the volume of data processed at a time. Adjusting the Value : Smaller values (e.g., 10 ): Useful for low-latency processing or when you want to process fewer records at a time. Larger values (e.g., 1000 ): Useful for batch processing or applications designed to handle high throughput. Use Case Scenarios : High throughput: Increase the value to minimize the number of poll() calls. Low latency or real-time processing: Decrease the value to process smaller chunks faster. Example Configuration Java Configuration: Properties props = new Properties(); p...
자세한 내용 보기