Confluent KSQL&KSQLDB完全教程-專為流處理應用程序構建的數據庫

ksqlDB is an open source event streaming database that was released by Confluent in 2017 (a little more than a year after Kafka Streams was introduced into the Kafka ecosystem).

It simplifies the way stream processing applications are built, deployed, and maintained, by integrating two specialized components in the Kafka ecosystem (Kafka Connect and Kafka Streams) into a single system, and by giving us a high-level SQL interface for interacting with these components.

ksqlDB Features

• Build a complete streaming app with a few SQL statements

• Simple constructs for building streaming apps: ksqlDB enables you to build event streaming applications leveraging your familiarity with relational databases. Three categories are foundational to building an application: collections, stream processing, and queries.

• One mental model for the entire stack:With a lightweight, familiar SQL syntax, ksqlDB presents a single mental model for working with event streams across your entire stack: event capture, continuous event transformations, aggregations, and serving materialized views.

ksqlDB Benefits

• More interactive workflows

• Less code to maintain since stream processing topologies are expressed using SQL instead of a JVM language.

• Lower barrier to entry and fewer new concepts to learn, especially for those who are familiar with traditional SQL databases but new to stream processing.

• Simplified architecture, since the interface for managing connectors (which integrate external data sources into Kafka) and transforming data are combined into a single system.

• Increased developer productivity since it takes less code to express a stream processing application, and complexities of lower-level systems are hidden beneath new layers of abstraction.

• Better support for data exploration

• 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.

• I suggest that you use ksqlDB whenever your project could leverage any of the preceding benefits stated and when your stream processing applications can be expressed naturally and simply using SQL.

• if you need lower-level access to your application state, need to run periodic functions against your data, need to work with data formats that aren’t supported in ksqlDB, want more flexibility with application profiling/monitoring , or have a lot of business logic that isn’t easily expressed in SQL, then Kafka Streams is a better fit.

• ksqlDB isn't a good fit for BI reports, ad-hoc querying, or queries with random access patterns, because it's a continuous query system on data streams.

An architecture with ksqlDB is much simpler, as many of those external parts have been consolidated into the tool itself: ksqlDB has primitives for connectors, and it performs stream processing. It also features materialized views, so that your data can be queried just like a database table directly in ksqlDB, without needing to be sent to an external source.

ksqlDB Components

• ksqlDB engine -- processes SQL statements and queries

• REST interface -- enables client access to the engine

• ksqlDB CLI -- console that provides a command-line interface (CLI) to the engine

• ksqlDB UI -- enables developing ksqlDB applications in Confluent Control Center and Confluent Cloud

You can deploy your ksqlDB streaming applications using either Interactive or Headless mode.

In both deployment modes, ksqlDB enables distributing the processing load for your ksqlDB applications across all ksqlDB Server instances, and you can add more ksqlDB Server instances without restarting your applications. ksqlDB CLI -- console that provides a command-line interface (CLI) to the engine

Note: All servers that run in a ksqlDB cluster must use the same deployment mode.

Installation

• Install zookeeper Cluster

• Install Kafka Cluster

• Install ksqldb instance

Quick Start

• ksql CLI use

• create streams

• insert record into stream

• query message from stream

ksqldb query lifecycle

• explain ksqldb query lifecycle

--zookeeper configuration at zookeeper.properties

dataDir=/var/lib/zookeeper

clientPort=2181

maxClientCnxns=0

admin.enableServer=false

-- kafka configuration at server.properties

broker.id=88

listeners=PLAINTEXT://192.168.88.130:9092

num.network.threads=3

num.io.threads=8

socket.send.buffer.bytes=102400

socket.receive.buffer.bytes=102400

socket.request.max.bytes=104857600

log.dirs=/var/lib/kafka/kafka-logs

num.partitions=1

num.recovery.threads.per.data.dir=1

offsets.topic.replication.factor=1

transaction.state.log.replication.factor=1

transaction.state.log.min.isr=1

log.retention.hours=168

log.segment.bytes=1073741824

log.retention.check.interval.ms=300000

zookeeper.connect=192.168.88.130:2181

zookeeper.connection.timeout.ms=18000

group.initial.rebalance.delay.ms=0

-- start zookeeper

bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

-- start kafka

bin/kafka-server-start.sh -daemon config/server.properties

-- create topic

bin/kafka-topics.sh  --bootstrap-server 192.168.88.130:9092 --create --topic test --partitions 3 --replication-factor 1

-- list topics

bin/kafka-topics.sh  --bootstrap-server 192.168.88.130:9092 --list

--produce messages

bin/kafka-console-producer.sh --bootstrap-server 192.168.88.130:9092 --topic test

--consume messages

bin/kafka-console-consumer.sh --bootstrap-server 192.168.88.130:9092 --topic test --from-beginning

curl -sq http://ksqldb-packages.s3.amazonaws.com/archive/0.28/archive.key | gpg --import

curl http://ksqldb-packages.s3.amazonaws.com/archive/0.28/confluent-ksqldb-0.28.2.tar.gz --output confluent-ksqldb-0.28.2.tar.gz

curl http://ksqldb-packages.s3.amazonaws.com/archive/0.28/confluent-ksqldb-0.28.2.tar.gz.asc --output confluent-ksqldb-0.28.2.tar.gz.asc

gpg --verify confluent-ksqldb-0.28.2.tar.gz.asc confluent-ksqldb-0.28.2.tar.gz

sudo tar -xzvf confluent-ksqldb-0.28.2.tar.gz -C /opt/

sudo ln -svnf confluent-ksqldb-0.28.2 ksqldb

sudo chown -hR ubuntu:ubuntu confluent-ksqldb-0.28.2 ksqldb

## ksql-server.properties

ksql.service.id=ksql_dev_

listeners=http://0.0.0.0:8088

ksql.logging.processing.topic.auto.create=true

ksql.logging.processing.stream.auto.create=true

bootstrap.servers=192.168.88.130:9092

compression.type=snappy

## start ksqldb server

bin/ksql-server-start -daemon etc/ksqldb/ksql-server.properties

## ksql cli connect ksqldb

bin/ksql http://192.168.88.130:8088

Create stream and topic

create stream user(id int key, username string, age int) with(KAFKA_TOPIC=’user’, VALUE_FORMAT=’json’,partitions=1);

Insert records into streams

insert into user(id,username,age) values(1,’Alex’,38);

insert into user(id,username,age) values(2,’Ben’,28);

insert into user(id,username,age) values(3,’Jeff’,55);

Push query against stream

select 'Hello '+username from user emit changes;

EMIT CHANGES here telling ksqlDB to run a push query, which will automatically emit/push changes to the client.

SELECT query will continue to run even after the initial set of results have been emitted.

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

create stream if not exists user2(id int key, username string, age int) with(KAFKA_TOPIC='user', VALUE_FORMAT='json');

insert into user(id,username,age) values(4,'Jenny',35);

-- create_stream_query.sql

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

create stream if not exists user3(id int key, username string, age int) with(KAFKA_TOPIC='user', VALUE_FORMAT='json');

select * from user3 emit changes;

bin/ksql --file ./scripts/create_stream_query.sql http://192.168.88.130:8088

bin/ksql --execute "set 'auto.offset.reset'='earliest';select * from user3 emit changes;" http://192.168.88.130:8088

• Register a ksqlDB stream or table from an existing Kafka topic with a DDL statement, like CREATE STREAM <my-stream> WITH <topic-name>.

• Express your app by using a SQL statement, like CREATE TABLE AS SELECT FROM <my-stream>

• ksqlDB parses your statement into an abstract syntax tree (AST).

• ksqlDB uses the AST and creates the logical plan for your statement.

• ksqlDB uses the logical plan and creates the physical plan for your statement.

• ksqlDB generates and runs the Kafka Streams application.

• You manage the application as a STREAM or TABLE with its corresponding persistent query.

Data Types&Serialization

• Built-in Data Types

• Custom Data Type

• Schema Registry & Serialization

Collections

• Source Collections

• Work with Collections

• Collection basic queries

• Conditional Expressions

Persistent Queries

• Create Derived queries

• Explain persistent query

Streams App Define by SQL

• Word count application

• XMall application

In this lecture i will tutorial how to use lambda function during query statement applied