ksql PRINT

Giriş

Belirtilen topic'i yazdırır

FROM BEGINNING

Örnek

Açıklaması şöyle

Please note that, by default KSQL is going to show us only the latest data arriving into the topic and not the historical data. 

Şöyle yaparız

ksql> print 'demo.demo.orders' from beginning; 

Key format: SESSION(KAFKA_STRING) or HOPPING(KAFKA_STRING) 
or TUMBLING(KAFKA_STRING) or KAFKA_STRING 

Value format: AVRO 

rowtime: 2020/07/13 14:36:53.207 Z, key: ..., value: ...

LIMIT

Normalde print komutu sonlanmaz, yeni girdi bekleyeme devam eder. Çıktıyı sınırlamak ve komutu sonlandırmak için LIMIT kullanılır

Örnek

Şöyle yaparız

ksql> print 'demo.demo.orders' from beginning lim1t 1; 
...
ksql>

ksql SHOW TOPICS

Örnek

Şöyle yaparız

ksql> show topics; 

Kafka Topic             | Partitions| Partition Replicas 
----------------------------------------------------------- 
demo.demo.orders        | 1         | 1 
orders                  | 1         | 1 
testksql_processing_log | 1         | 1 
-----------------------------------------------------------

ksql CREATE SOURCE CONNECTOR

Giriş

Debezium Connector yaratır

Örnek

Şöyle yaparız. demo.demo.orders diye yeni bir topic yaratır

ksql> CREATE SOURCE CONNECTOR SOURCE_MYSQL_02 WITH ( 
    'connector.class' = 'io.debezium.connector.MySQL.MySQLConnector', 
    'database.hostname' = 'MySQL', 
    'database.port' = '3306', 
    'database.user' = 'mysql_user', 
    'database.password' = 'mysql_pw', 
    'database.server.id' = '1', 
    'database.server.name' = 'demo', 
    'table.whitelist' = 'demo.orders', 
    'database.history.Kafka.bootstrap.servers' = 'Kafka:29092', 
    'database.history.Kafka.topic' = 'orders' , 
    'include.schema.changes' = 'false', 
    'key.converter'= 'org.apache.Kafka.connect.storage.StringConverter', 
    'value.converter'= 'io.confluent.connect.avro.AvroConverter', 
    'value.converter.schema.registry.url'= 'http://schema-registry:8081' 
    ); 
 
Message 
----------------------------------- 
Created connector SOURCE_MYSQL_02 

Debezium ve Kafka Connect İlişkisi - Debezium Connector Altta Kafka Connect Kullanır

Giriş

Açıklaması şöyle. Kısaca Debezium Connector Altta Kafka Connect Kullanır

It leverages the Kafka Connect framework to capture changes from various database systems and publish them as messages to Kafka topics.

Debezium provides a set of connectors for different databases, such as MySQL, PostgreSQL, MongoDB, and Oracle. These connectors read the transaction logs or other change streams of the source database and generate Kafka messages for each change. The messages are then consumed by downstream systems, such as data lakes, data warehouses, or other databases.

By leveraging Kafka, Debezium provides a scalable and fault-tolerant platform for capturing and processing database changes in real-time. Kafka's distributed architecture allows for parallel processing of messages across multiple nodes, and its built-in replication and fault-tolerance features ensure that data is reliably delivered to downstream systems.

Debezium also provides a set of features for data transformation, such as filtering, enrichment, and masking of sensitive data. These features can be applied to the Kafka messages before they are consumed by downstream systems.

Örnek

Debezium kendi içinde bir Kafka Connect bileşeni içeriyor. Kafka Connect'i direkt çalıştırmak istersek şöyle yaparız

cd /opt/kafka/bin
connect-standalone.sh \
  config/connect-standalone.properties \
  config/debezium-oracle-connector.properties

Loglama ayarlarını değiştirmek için şöyle yaparız

$KAFKA_HOME/config/connect-log4j.properties

log4j.rootLogger=DEBUG, stdout, connectAppender

kafka-console-producer komutu

Giriş

kafka-console-producer komutu çalıştırıldıktan sonra ekran göndermek istediğimiz mesajı yazarız ve Enter'a basarız

Bash Örnekleri

Örnek

Şöyle yaparız. Bu betik sürekli sensors isimle topic'e veri gönderir.

#!/bin/bash

topic="sensors"
kafka_server="<your-ip>:9092"
log_file="sensors.log"
KAFKA_HOME="/home/opc/kafka_build_323/kafka_2.13-3.2.3"

while true; do
  # Generate random sensor_id, temperature, and humidity
  sensor_id=$(printf "s%03d" $((RANDOM % 100 + 1)))
  temperature=$(echo "scale=2; 50 + (100 - 50) * $RANDOM / 32767" | bc)
  humidity=$(echo "scale=2; 20 + (80 - 20) * $RANDOM / 32767" | bc)
  timestamp=$(date -u +"%Y-%m-%dT%H:%M:%SZ")

  # Create single-line JSON payload
  payload="{\"schema\": {\"type\": \"struct\", \"fields\": [{\"type\": \"string\", \"optional\": false, \"field\": \"sensor_id\"}, {\"type\": \"float\", \"optional\": false, \"field\": \"temperature\"}, {\"type\": \"float\", \"optional\": false, \"field\": \"humidity\"}, {\"type\": \"string\", \"optional\": false, \"field\": \"timestamp\"}]}, \"payload\": {\"sensor_id\": \"$sensor_id\", \"temperature\": $temperature, \"humidity\": $humidity, \"timestamp\": \"$timestamp\"}}"

  # Send payload to Kafka topic
  echo "$payload" | $KAFKA_HOME/bin/kafka-console-producer.sh 
    --bootstrap-server $kafka_server \
    --topic $topic \

  # Append payload to log file
  echo "$payload" >> $log_file

  sleep 1
done

JSON şöyle

{
  "schema": {
    "type": "struct", 
    "fields": [
       {"type": "string", "optional": false, "field": "sensor_id"}, 
       {"type": "float", "optional": false, "field": "temperature"}, 
       {"type": "float", "optional": false, "field": "humidity"}, 
       {"type": "string", "optional": false, "field": "timestamp"}
    ]
  }, 
  "payload": {"sensor_id": "s076", "temperature": 54.23, "humidity": 65.45, "timestamp": "2023-05-10T12:26:47Z"}
}

--broker-list seçeneği

Örnek

Şöyle yaparız

kafka-console-producer \
--topic create-item \
--broker-list kafka:29092

{"id": "626bd1bd-c565-48ac-87b2-28f2247f6dea", "name": "my-new-item"}

producer.config  seçeneği

Örnek

Şöyle yaparız

bin/kafka-console-producer.sh \
  --broker-list $brokerssasl \
  --topic test \
  --producer.config /path/to/client.properties \
  > Hello world!

--producer-property seçeneği

Örnek

Şöyle yaparız

kafka-console-producer.sh \
  --bootstrap-server localhost:9092 \
  --topic my-topic \
  --producer-property acks=all

--property seçeneği

Örnek

Şöyle yaparız. Bu komut ile key ve value gönderilir.

bin/kafka-console-producer.sh --broker-list host1:9092,host2:9092,host3:9092 \
 --topic good_topic \
 --property parse.key=true \
 --property key.separator=,

Örnek

Açıklaması şöyle

The choice of the key.separator property is arbitrary. You can use any character. And now, you can send full key/value pairs from the command line!

Şöyle yaparız

kafka-console-producer --topic <topic-name> \
                       --broker-list <broker-host:port> \
                       --property parse.key=true \
                       --property key.separator=":"

--topic seçeneği

Örnek

test kuyruğuna mesaj yazmak için şöyle yaparız. Bu komut ile key göndermek mümkün değil, sadece value gönderilir.

kafka-console-producer.bat --broker-list localhost:9092 --topic test




Retry Queue

Bir örnek burada

ksql CREATE STREAM - Belirtilen Topic'ten Veri Okur

Giriş

CRATE STREAM ile kafka topic'ten okuyan bir stream yaratılır. Açıklaması şöyle

While we create a stream, we also have to supply the underlying topic, from where the stream shall be reading the data and be formed. We also need to specify, how the data is encoded. 

1. CREATE STREAM CLAUSE

Sütun tipi olarak

ARRAY<STRUCT<...>>,

DOUBLE

INTEGER

DECIMAL

VARCHAR

kullanılabilir

2. WITH CLAUSE

WITH kısmından sonra gelenler için açıklama şöyle

kafka_topic: Name of the Kafka topic underlying the stream.

value_format: Encoding of the messages stored in the Kafka topic.

partitions: Number of partitions to create for the locations topic. This is optional if the topic already exists.

kafka_topic ile okunacak topic belirtilir

value_format olarak şunlar kullanılabilir

AVRO

JSON 

DELIMITED

MATERIALIZED VIEW

Stream ile MATERIALIZED VIEW yaratılır. ksql MATERIALIZED VIEW yazısına taşıdım

Kullanım Örnekleri

Örnek - Stream + Materialized View

Şöyle yaparız

CREATE STREAM riderLocations (
  profileId VARCHAR,
  latitude DOUBLE,
  longitude DOUBLE)
WITH (
  kafka_topic='locations',
  value_format='json',
  partitions=1
);

// create Materialized view to get the latest data, which will keep track of the location 
// of the riders. 
// We will be using LATEST_BY_OFFSET which will return the latest value of the 
// specified column.
CREATE TABLE currentLocation AS
  SELECT
    profileId,
    LATEST_BY_OFFSET(latitude) AS la,
    LATEST_BY_OFFSET(longitude) AS lo   
  FROM riderlocations
  GROUP BY profileId
  EMIT CHANGES;

// create a materialized view which will give the location of the 
// rider within specified numbers
CREATE TABLE ridersNearLeiden AS
  SELECT 
    ROUND(GEO_DISTANCE(la, lo, 37.4133, -122.1162), -1) AS distanceInMiles,
    COLLECT_LIST(profileId) AS riders,   
    COUNT(*) AS count FROM currentLocation
    GROUP BY ROUND(GEO_DISTANCE(la, lo, 37.4133, -122.1162), -1);

Push query şöyle

// Push query is a continuous stream of updates to the ksqlDB. 
// The result of this statement isn’t persisted in a Kafka topic and is 
// printed out only in the console, or returned to the client.

SELECT * FROM riderLocations
 WHERE GEO_DISTANCE(latitude, longitude, 37.4133, -122.1162) <= 5 
 EMIT CHANGES;

Pull query şöyle

// Pull query fetch the current value from materialized view and terminates. 
// The result of this statement is not persisted in a Kafka topic and is 
// printed out only in the console.

// This query will return all the rider's profiles within a distance of 10miles.

SELECT * FROM ridersNearLeiden WHERE distanceInMiles <= 10;

Örnek

Şöyle yaparız. rabbit isimli bir stream oluşturur. rabbit-test-00 isimli topic'ten okur.

CREATE STREAM rabbit (transaction VARCHAR,
                      amount VARCHAR,
                      timestamp VARCHAR)
  WITH (KAFKA_TOPIC='rabbit-test-00',
        VALUE_FORMAT='JSON');

Bu stream'i süzen bir başka stream yaratmak için şöyle yaparız

CREATE STREAM transactions WITH (VALUE_FORMAT='AVRO') AS
  SELECT transaction AS tx_type,
         SUBSTRING(amount,1,1) AS CURRENCY,
         CAST(SUBSTRING(amount,2,LEN(amount)-1) AS DECIMAL(9,2)) AS tx_amount,
         TIMESTAMP AS tx_timestamp
    FROM rabbit
   WHERE timestamp IS NOT NULL
    EMIT CHANGES;

Örnek

Stream'i tablo haline getirip sorgulamak içindir şöyle yaparız

CREATE STREAM transactions (
  order_id INT KEY,
  customer_id INT,
  store_id INT,
  total DOUBLE,
  created_at VARCHAR
) WITH (
  kafka_topic = 'trasactions',
  partitions = 2,
  value_format = 'json'
);

Sorgulamak için şöyle yaparız. Önce  bir tablo yaratırız

CREATE TABLE sales_by_store AS
  SELECT store_id, SUM(total) AS total
  FROM TRANSACTIONS
  GROUP BY store_id
  EMIT CHANGES;

Açıklaması şöyle

What happens when you run this statement on ksqlDB?

The server creates a new persistent query that runs forever, processing data as it arrives. When each row is read from the transactions stream, the persistent query does two things.

1. Incrementally updates the materialized view to integrating the incoming row.

2. Emits a row to a changelog topic.

You can think of the changelog topic as an audit trail of all updates made to the materialized view. That will come in handy when we discuss the fault-tolerance. So let’s skip that for now.

Sonra sorgularız. Şöyle yaparız.

SELECT * FROM sales_by_store WHERE store_id=2000;

Sorguyu da push'lamak istersek. Şöyle yaparız.

SELECT * FROM sales_by_store WHERE store_id=2000 EMIT CHANGES;