KSQL DB 学习笔记1

Summary

官网：ksqldb.io

ksqlDB and Kafka Streams
ksqlDB is built on Kafka Streams, a robust stream processing framework that is part of Apache Kafka®. You can use ksqlDB and Kafka Streams together in your event streaming applications.

Stream ，Table，Window这些核心概念在KSQL也一样存在。

PS :UI 好像没有开源，也找不到对应docker image。

可以将Kafka Connect(比如连接到PG、ElasticSearch）与ksql DB直接结合起来，通过sql模式读取、写入数据到外部。

通过JMX metrics对外提供度量数据：https://docs.ksqldb.io/en/latest/reference/metrics/

启动命令：
java -cp /usr/share/java/ksqldb-rest-app/: -Xmx3g -server -XX:+UseCo
ncMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+ExplicitGCInvokesConcurrent -XX:NewRatio=1 -Djava.awt.headless=true -D
com.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dksql.log.dir=/usr/logs -Dl
og4j.configuration=file:/etc/ksqldb/log4j.properties -Dksql.server.install.dir=/usr -Xlog:gc:file=/usr/logs/ksql-server-gc.log:time,tags:filecount=
10,filesize=102400 io.confluent.ksql.rest.server.KsqlServerMain ksql-server.properties

KsqlDB Server 提供rest接口提供对外访问

核心对象

Streams
A stream is a partitioned, immutable, append-only collection that represents a series of historical facts.

Tables
A table is a mutable, partitioned collection that models change over time. In contrast with a stream, which represents a historical sequence of events, a table represents what is true as of “now”.

Keys
You can mark a column with the KEY keyword to indicate that it’s a key column.

Headers
Starting in ksqlDB 0.24, you can mark a column with HEADERS or HEADER(‘’) to indicate that it is populated by the header field of the underlying Kafka record.

Pseudocolumns
A pseudocolumn is a column that’s automatically populated by ksqlDB and contains meta-information that can be inferred about the row at creation time.

Primary key constraints(table 必须有PK,不能有KEY)
Only tables can have primary keys. Streams do not support them.
Adding multiple rows to a table with the same primary key doesn’t cause the subsequent rows to be rejected.

Custom Types
The CREATE TYPE statement registers a type alias directly in KSQL.

Materialized Views
The benefit of a materialized view is that it evaluates a query on the changes only (the delta), instead of evaluating the query on the entire table.
If a table is created directly on top of a Kafka topic, it’s not materialized. Non-materialized tables can’t be queried, because they would be highly inefficient. On the other hand, if a table is derived from another collection, ksqlDB materializes its results, and you can make queries against it.

STREAM /TABLE 关键属性

KAFKA_TOPIC (required)¶
The name of the Kafka topic that backs the table.

KEY_FORMAT¶
The serialization format of the message key in the topic. For supported formats, see Serialization Formats.

If not supplied, the system default is used, defined by ksql.persistence.default.format.key. If the default is also not set, the statement is rejected as invalid.

You can’t use the KEY_FORMAT property with the FORMAT property in the same CREATE TABLE statement.

KEY_SCHEMA_ID¶
The schema ID of the key schema in Schema Registry.

TIMESTAMP ;By
default, the ROWTIME pseudo column is the timestamp of the message in the Kafka topic.
You can use the TIMESTAMP property to override ROWTIME with the contents of the specified column within the Kafka message, similar to timestamp extractors in the Kafka Streams API.
Time-based operations, like windowing, process a record according to the timestamp in ROWTIME.
Timestamps have an accuracy of milliseconds.

TIMESTAMP_FORMAT The format of the timestamp. Any format supported by
java.time.format.DateTimeFormatter is valid (e.g.,
yyyy-MM-dd HH:mm:ss).

VALUE_DELIMITER The character that acts as the field delimiter when VALUE_FOR
MAT=‘DELIMITED’. Commas (,) are the default delimiter, but
‘SPACE’ and ‘TAB’ are also valid values.

VALUE_FORMAT¶
The serialization format of the message value in the topic. For supported formats, see Serialization Formats.

If VALUE_FORMAT isn’t provided, the system default is used, defined by ksql.persistence.default.format.value. If the default is also not set, the statement is rejected as invalid.