ksql LIST STREAMS veya SHOW STREAMS

Giriş

Not : LIST STREAMS veya SHOW STREAMS aynı şey.

Sütunlar şöyle

1. Stream Name

2. Kafka Topic

3. Key Format

4. Value Format

5. Windowed

ksql Push Queries - Stream İçin Tail Komutuna Benzer

Giriş

İsmi Push Queries çünkü Kafka sonuçları bize sürekli gönderiyor. Açıklaması şöyle.

A push query is a form of query issued by a client that subscribes to a result as it changes in real-time. It’s a continuous query that pushes the incremental result to the client in real-time. Push queries enable you to query a stream or materialized table with a subscription to the results

ksql 5.4'ten itibaren EMIT CHANGES kullanılması zorunlu oldu.

Örnek

Şöyle yaparız. SQL cümlesi çalıştırıldıktan sonra sadece yeni gelen veriyi göstermeye başlar.

SELECT name, countrycode FROM users_stream
EMIT CHANGES

Örnek

Şöyle yaparız. SQL cümlesi çalıştırıldıktan sonra sadece yeni gelen veriyi göstermeye başlar.

SELECT name, countrycode FROM users_stream EMIT CHANGES;

Eğer tüm veriyi görmek istersek şöyle yaparız

1. SET 'auto.offset.reset' = 'earliest';
2. SELECT name, countrycode FROM users_stream EMIT CHANGES;

Açıklaması şöyle

Please note here that, by default KSQL would only show us the new/future data and not the already-present data.

In order to view the historical data from the stream, we can modify the auto.offset.reset to ‘earliest’ value. Please note that, this property change is applicable to only this window. In case we exit and enter again to KSQL-CLI, we might have to again set this property.

ksql LIST TOPICS veya SHOW TOPICS

Giriş

Not : LIST TOPICS veya SHOW TOPICS aynı şey. Açıklaması şöyle

Kafka topic names are case-sensitive (“Ratings” and “ratings” are two different topics on a Kafka broker). All the KSQL constructs though, like Streams and Tables and everything else, are case-insensitive as you would expect from most database-like systems.

Sütunlar şöyle

1. Kafka Topics

2. Partitions

3. Partition Replicas

ksql Nedir

Giriş

Açıklaması şöyle

Purpose of KSQL :- At times, writing Streams-application can become complex task, another option is to use KSQL. With KSQL, we are going to write the SQL like queries and its going to generate a kafka-streams-application in the background. Thus, Using KSQL, it would allow us to write an Kafka Streams application, but in much simpler way.

KSQL CLI

Confluent CLI kullanılabilir. CLI bir KSQL sunucusu ile konuşur. Sunucu da SQL cümlelerini Kafka Stream uygulamasına çevirir. Şeklen şöyle

ksql MATERIALIZED VIEW - CREATE STREAM + CREATE TABLE

Giriş

1. CREATE STREAM ile bir stream yaratılır. Stream yaratılırken belirtilen Kafka topic'e veri yazmak için kullanılır. Yani Stream altta veriyi saklamak için bir topic kullanmak zorundadır

2. CREATE TABLE ile bir materialized view yaratılır. Materialized view sorgu için kullanılır

CREATE TABLE

MATERIALIZED VIEW İçin CREATE TABLE yazısına taşıdım

CREATE SOURCE TABLE

Açıklaması şöyle

Provide the SOURCE clause to enable running pull queries on the table.

The SOURCE clause runs an internal query for the table to create a materialized state that's used by pull queries. You can't terminate this query manually. Terminate it by dropping the table with the DROP TABLE statement.

When you create a SOURCE table, the table is created as read-only. For a read-only table, INSERT, DELETE TOPIC, and DROP TABLE statements aren't permitted.

To disable the SOURCE table feature, set ksql.source.table.materialization.enabled to false in the ksqlDB Server properties file.

MATERIALIZED VIEW TEST

1. Önce bir SELECT ile MATERIALIZED VIEW'un boş olduğu görülür

2. Sonra STREM'E veri eklenir

3. Sonra SELECT ile MATERIALIZED VIEW'un dolu olduğu görülür

Örnek - Stream + Materialized View

Şöyle yaparız. Burada stream'e yazılan veri company isimli topic'e yazılıyor

// Stream
CREATE STREAM company_stream (
  id VARCHAR KEY,  
  name VARCHAR,
  revenue DOUBLE
) WITH (
  kafka_topic = 'company',
  partitions = 2,
  value_format = 'json'
);

// Materialized view
SET 'auto.offset.reset' = 'earliest';

CREATE TABLE company_latest AS
SELECT 
  id, 
  LATEST_BY_OFFSET(name) AS name,
  LATEST_BY_OFFSET(revenue) AS revenue
FROM company_stream
GROUP BY id
EMIT CHANGES;

Test için şöyle yaparız

// Empty result set
SELECT * FROM companies_latest;

INSERT INTO company_stream (id, name, revenue) VALUES ('AMZ', 'Amazon', 100);

INSERT INTO company_stream (id, name, revenue) VALUES ('AMZ', 'Amazon', 450);

INSERT INTO company_stream (id, name, revenue) VALUES ('GOG', 'Google', 90);
INSERT INTO company_stream (id, name, revenue) VALUES ('APL', 'Apple', 130);

INSERT INTO company_stream (id, name, revenue) VALUES ('GOG', 'Google', 99);

INSERT INTO company_stream (id, name, revenue) VALUES ('APL', 'Apple', 139);

// Populated result set
SELECT *  FROM companies_latest;

DROP MATERIALIZED VIEW

Önce MATERIALIZED VIEW sonra STREAM drop edilir

Örnek

Şöyle yaparız

DROP TABLE IF EXISTS company_latest DELETE TOPIC;
DROP STREAM IF EXISTS company_stream DELETE TOPIC;

DELETE FROM MATERIALIZED VIEW

Açıklaması şöyle

To remove a record from the Kafka topic, we need to execute an INSERT statement with a NULL value. The NULL value records are called tombstones in Kafka.

Açıklaması şöyle

We can only insert NULL value into a stream that has VALUE_FORMAT set to KAFKA.

Açıklaması şöyle

There is a feature request to implement some kind of DELETE statement that would insert NULL value into a stream. See and up vote the following feature if you agree with the following feature request: https://github.com/confluentinc/ksql/issues/7073le

Örnek

Önce bir stream yaratırız

SET 'auto.offset.reset' = 'earliest';

CREATE STREAM companies (
  id VARCHAR KEY,  
  name VARCHAR,
  revenue DOUBLE,
  deleted BOOLEAN
) WITH (
  kafka_topic = 'companies',
  partitions = 2,
  value_format = 'AVRO'
);

Bu stream'i okuyan iki tane stream daha yaratırız. Şöyle yaparız. Burada companies stream deleted alanına göre ikiye ayrılıyor. companies_deleted stream'i value_format = 'KAFKA' kullanıyor. Çünkü value değeri NULL olarak atanıyor

CREATE STREAM companies_existing 
WITH (
  kafka_topic = 'companies_latest',
  partitions = 2,
  value_format = 'AVRO'
) AS 
SELECT *
FROM companies
WHERE deleted = FALSE;

CREATE STREAM companies_deleted 
WITH (
  kafka_topic = 'companies_latest',
  partitions = 2,
  value_format = 'KAFKA'
) AS 
SELECT ID, CAST(NULL AS VARCHAR) 
FROM companies
WHERE deleted = TRUE;

Daha sonra bir materialized view yaratırız. Şöyle yaparız. Burada Source Table yaratılıyor

CREATE SOURCE TABLE companies_latest (
  id VARCHAR PRIMARY KEY,  
  name VARCHAR,
  revenue DOUBLE
) WITH (
  kafka_topic = 'companies_latest',
  partitions = 2,
  value_format = 'AVRO'
);

Test etmek için biraz veri ekleriz. Şöyle yaparız

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('AMZ', 'Amazon', 100, false);

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('AMZ', 'Amazon', 450, false);

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('GOG', 'Google', 90, false);

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('APL', 'Apple', 130, false);

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('GOG', 'Google', 99, false);

INSERT INTO companies (id, name, revenue, deleted) 
VALUES ('APL', 'Apple', 139, false);

Şöyle yaparız

SELECT * FROM companies_latest;

Test etmek için bir satırı sileriz

INSERT INTO companies (id, deleted) VALUES ('APL', true);

Materialized view'a bakınca çıktı şöyle. En son satırda REVENUE alanı null

ksql> print company_latest;
Key format: KAFKA_STRING
Value format: AVRO
rowtime: 2023/01/25 18:12:25.355 Z, key: AMZ, value: {"NAME": "Amazon", "REVENUE": 100.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:25.422 Z, key: AMZ, value: {"NAME": "Amazon", "REVENUE": 450.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:25.502 Z, key: GOG, value: {"NAME": "Google", "REVENUE": 90.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:25.530 Z, key: APL, value: {"NAME": "Apple", "REVENUE": 130.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:25.562 Z, key: GOG, value: {"NAME": "Google", "REVENUE": 99.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:25.621 Z, key: APL, value: {"NAME": "Apple", "REVENUE": 139.0, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:12:30.311 Z, key: APL, value: {"NAME": null, "REVENUE": null, "DELETED": false}, partition: 0
rowtime: 2023/01/25 18:16:20.989 Z, key: APL, value: <null>, partition: 0

Her şeyi silmek için şöyle yaparız. Önce MATERIALIZED VIEW sonra STREAM'ler drop edilir

DROP TABLE IF EXISTS companies_latest;
DROP STREAM IF EXISTS companies_deleted;
DROP STREAM IF EXISTS companies_existing DELETE TOPIC;
DROP STREAM IF EXISTS companies DELETE TOPIC;

Docker Compose ve Schema Registry

Giriş

confluentinc/cp-zookeeper

confluentinc/cp-server

confluentinc/cp-schema-registry

kullanılır

Örnek

Şöyle yaparız

---
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.2
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-server:7.3.2
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_VALUE_SCHEMA_VALIDATION: "true"
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081

  schema-registry:
    image: confluentinc/cp-schema-registry:7.2.2
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

- 

Kafka Producer KafkaProducer.send metodu - Avro Gönderme - Schema Registry Mesaj Gönderme

Giriş

Açıklaması şöyle

Kafka client library automatically handles data serialization. For the client library to serialize data into AVRO, configure the Kafka producer to use KafkaAvroSerializer with the url of schema registry, it will automatically register schema with the register and convert data into AVRO format as part of message transmission.

Maven

Örnek

Şu satırı dahil ederiz

<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-schema-registry-client</artifactId>
  <version>6.2.0</version>
</dependency>
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>6.2.0</version>
</dependency>

Örnek

Maven şu satırı dahil ederiz

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
</dependency>

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-schema-registry-client</artifactId>
</dependency>

<dependency>
    <groupId>io.confluent</groupId>
    <artifactId>kafka-avro-serializer</artifactId>
</dependency>

<dependency>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro</artifactId>
</dependency>

Örnek

Şöyle yaparız. Toplam 4 tane özelliği atamak gerekiyor. 

VALUE_SERIALIZER_CLASS_CONFIG : KafkaAvroSerializer

SCHEMA_REGISTRY_URL_CONFIG : Schema Registry Sunucusu Adresi

@Configuration
public class KafkaConfig {
  @Value("${spring.kafka.bootstrap-servers}")
  private String bootstrapServers;
    
  @Value("${spring.kafka.schema-registry-url}")
  private String schemaRegistryUrl;
    
  @Bean
  public Map<String, Object> producerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
    props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
    return props;
  }
  // Define your producer factory, Kafka template, etc.
}

Örnek

Şöyle yaparız. Burada key olarak "schema.registry.url" kullanılıyor ama zaten bunun için bir sabit var.

@PostMapping("/insurance-claims")
public ResponseEntity<Void> generateClaimRequest() {
  KafkaProducer producer = createKafkaProducer();

  InsuranceClaimKey key = generateAvroClaimRequestKey();
  InsuranceClaim value = generateAvroClaimRequest();
  ProducerRecord<InsuranceClaimKey, InsuranceClaim> producerRecord = 
                            new ProducerRecord<>("claim-submitted", key, value);
  kafkaProducer.send(producerRecord).get();
    
  return ResponseEntity.ok().build();
}

private KafkaProducer<InsuranceClaimKey, InsuranceClaim> createKafkaProducer() {
  Properties props = new Properties();
  props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
  props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
  props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
  props.put("schema.registry.url", "http://localhost:8081");
  return new KafkaProducer<>(props);
}

Örnek

Şöyle yaparız. Burada key olarak "schema.registry.url" kullanılıyor ama zaten bunun için bir sabit var.

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

Company comp = new Company();
...
Producer<String, Company> producer = new KafkaProducer<>(props);
sendData(producer, new ProducerRecord<>("company", "12345", comp));
producer.close();

void sendData(Producer producer, ProducerRecord record) {
  try {
    RecordMetadata meta = (RecordMetadata) producer.send(record).get();
    System.out.printf("key=%s, value=%s => partition=%d, offset=%d\n", 
      record.key(), record.value(), meta.partition(), meta.offset());

  } catch (InterruptedException | ExecutionException e) {
    System.out.printf("Exception %s\n", e.getMessage());
  }
}