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!

댓글

댓글 쓰기

이 블로그의 인기 게시물

Fundamentals of English Grammar #1

That's a great textbook table of contents! It covers the fundamental tenses in English. Here is a summary of the highlighted sections, with example sentences and key learning tips. PRESENT TIME Item Topic Example Sentence(s) Learning Summary 1-1 The simple present and the present progressive Simple Present: The sun rises in the east. (Fact/Habit) / Present Progressive: She is studying for her exam right now. (Action in progress) Simple Present is for facts, habits, and routines . Present Progressive is for actions happening at the moment of speaking or temporary situations . 1-3 Frequency adverbs I always drink coffee in the morning. / He is never late for class. These adverbs ( always, usually, often, sometimes, rarely, never ) describe how often an action occurs. Their position is key: before the main verb, but after the verb to be or auxiliary verbs. 1-4, 1-5 Final -s / Spelling of final -s/-es He read s a book. / She watch es TV. The final -s (or -es ) is adde...
자세한 내용 보기

To switch to a specific tag in a Git repository

To switch to a specific tag in a Git repository, you can use the git checkout or git switch command. Here's how you can do it: 1. List Available Tags First, you may want to see a list of all available tags in the repository: git tag 2. Checkout a Specific Tag Once you know the tag you want to switch to, you can use either git checkout or git switch (if you're using a newer version of Git, which recommends git switch for checking out branches). Using git checkout : git checkout <tag- name > Using git switch (recommended for newer versions of Git): git switch -- detach <tag- name > The --detach flag is necessary because tags are not branches; they are just pointers to specific commits. Using --detach makes your HEAD point to the tag without modifying any branch. 3. Verify the Checkout After switching to the tag, verify that you're on the correct commit: git status This will show the tag you're currently on (you'll see somethi...
자세한 내용 보기

kafka polling vs listen

In Kafka, polling and listening are two different approaches to consuming messages, and they correspond to different APIs or frameworks. 1. Kafka Polling Polling refers to the use of Kafka's native poll() method in the Kafka Consumer API . It requires the consumer application to actively request messages from Kafka. Key Characteristics: Active Consumption : The application explicitly calls poll() in a loop to fetch messages. Manual Commit : Developers can choose when to commit offsets, giving fine-grained control over processing and checkpointing. Use Case : Recommended for applications where control over message consumption, offset management, or threading is critical. Example (Java): Properties props = new Properties(); props . put (ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092" ); props . put (ConsumerConfig.GROUP_ID_CONFIG, "group1" ); props . put (ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest" ); props . ...
자세한 내용 보기

Create topic on Kafka with partition count, 카프카 토픽 생성하기

To create a Kafka topic with 100 partitions, you can use the Kafka command-line tool ( kafka-topics.sh or kafka-topics.bat for Windows). When creating a topic, you can specify the number of partitions and the replication factor. Here’s how to create a topic with 100 partitions: Step-by-step instructions: Open a terminal on your machine where Kafka is installed. Run the kafka-topics.sh command with the necessary options to create the topic: kafka - topics . sh - - create - - bootstrap - server localhost:9092 - - replication - factor 1 - - partitions 100 - - topic < topic_name > --bootstrap-server localhost:9092 : Specifies the Kafka broker (or cluster) address. Replace localhost:9092 with your actual Kafka broker address if it’s different. --replication-factor 1 : Specifies the number of replicas for each partition. In this example, it's set to 1 , but in a production environment, you’ll want a higher replication factor for redundancy. --partitions 10...
자세한 내용 보기

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.*;...
자세한 내용 보기

To download a file from MinIO using Spring Boot, 스프링부트 Minio 사용하기

To download a file from MinIO using Spring Boot, you can utilize the MinIO Java SDK. Here's an example of how you can achieve this: Add the MinIO dependency to your pom.xml file: < dependency > < groupId > io.minio </ groupId > < artifactId > minio </ artifactId > < version > 7.1.0 </ version > </ dependency > Create a configuration file (e.g., MinioConfig.java ) to establish a connection with your MinIO server: ```java import io.minio.MinioClient; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @Configuration public class MinioConfig { @Value( " ${minio.endpoint} " ) private String endpoint; @Value( " ${minio.accessKey} " ) private String accessKey; @Value( " ${minio.secretKey} " ) private String secretKey; @Bean public MinioClient minioClient() { re...
자세한 내용 보기

Joining an additional control plane node to an existing Kubernetes cluster

Joining an additional control plane node to an existing Kubernetes cluster involves specific steps to ensure proper synchronization between the control plane components. Below is the detailed process: 1. Pre-Requisites Ensure the following on the new control plane node: Same Kubernetes Version : The new node must have the same Kubernetes version installed as the existing control plane. kubeadm version Kubeadm, Kubelet, and Kubectl Installed : Ensure these tools are installed and configured. Network Configuration : The new control plane must have network connectivity to the existing control plane nodes. 2. Retrieve the Join Command On an existing control plane node, generate the kubeadm join command with the --control-plane flag: kubeadm token create -- print -join-command You will get output similar to: kubeadm join <control-plane-endpoint> : 6443 --token <token> \ --discovery-token- ca -cert-hash sha256 : <hash> \ --control-plane 3. ...
자세한 내용 보기

Vespa vs Milvus

Vespa and Milvus are both powerful open-source vector search engines, but they differ significantly in terms of architecture, features, and use cases. Here's a comparison of the two: 1. Overview Vespa : Type : Distributed Search Engine Primary Focus : Full-text search, recommendation systems, machine learning, and vector search. Use Cases : eCommerce, news, social media, personalized recommendations, and general search. Key Strengths : Scalable, handles both structured and unstructured data, supports complex queries (e.g., multi-field search, ranking, aggregations). Milvus : Type : Vector Database and Search Engine Primary Focus : Efficient similarity search for high-dimensional vectors (commonly used in machine learning, AI, and computer vision tasks). Use Cases : AI-driven applications, image search, video search, recommendation engines based on embeddings, NLP, and other ML-based tasks. Key Strengths : Optimized for vector search, handles billions of vectors wi...
자세한 내용 보기

max_active_runs of Airflow

In Apache Airflow, if you want to ensure that a new run of a DAG doesn’t start before the previous one has completed, you can use the max_active_runs parameter in the DAG definition. Setting this parameter to 1 ensures that only one instance of the DAG is running at any given time. Setting max_active_runs for a DAG Here’s an example of how to set up a DAG with max_active_runs=1 : from airflow import DAG from airflow.operators.dummy_operator import DummyOperator from datetime import datetime # Define the DAG with DAG( 'my_dag' , description= 'A sample DAG' , schedule_interval= '@daily' , # Set your schedule start_date=datetime( 2023 , 1 , 1 ), catchup= False , max_active_runs= 1 # This prevents new runs before the previous run completes ) as dag: # Define tasks start = DummyOperator(task_id= 'start' ) end = DummyOperator(task_id= 'end' ) # Define task dependencies ...
자세한 내용 보기

urllib3 with proxy settings

To set a proxy using urllib3 , you typically configure it through a ProxyManager , which allows you to route requests through a specific proxy server. Here’s a quick guide: Step 1: Install urllib3 Make sure you have urllib3 installed: pip install urllib3 Step 2: Use ProxyManager Use the ProxyManager class to configure your proxy. Here’s an example: import urllib3 # Define the proxy URL proxy_url = "http://your_proxy_server:port" # Create a ProxyManager instance http = urllib3.ProxyManager(proxy_url) # Send a GET request through the proxy response = http.request( "GET" , "http://example.com" ) # Print the response print(response.data.decode( "utf-8" )) Step 3: Add Authentication (Optional) If your proxy server requires authentication, include it in the proxy URL: proxy_url = "http://username:password@your_proxy_server:port" http = urllib3.ProxyManager(proxy_url) Additional Tips For HTTPS URLs, ProxyManager ...
자세한 내용 보기