람다 함수

람다 함수 또는 줄여서 "람다"를 사용하여 다양한 방식으로 입력 값에 적용할 수 있는 간단한 인라인 함수를 표현합니다. 

예를 들어 컬렉션의 각 요소에 람다 함수를 적용하여 결과 컬렉션을 변환할 수 있습니다. 또한 람다를 사용하여 컬렉션의 요소를 필터링하거나 컬렉션을 단일 값으로 줄일 수 있습니다. 

람다의 장점은 전체 UDF를 구현할 필요가 없는 방식으로 사용자 정의 기능을 표현할 수 있습니다.

가이드 에서 람다 함수를 사용하는 방법을 알아봅니다.

Confluent 플랫폼용 JDBC 커넥터(소스 및 싱크)

Kafka Connect JDBC 소스 커넥터를 사용하여 JDBC 드라이버가 있는 모든 관계형 데이터베이스에서 Apache Kafka® topic으로 데이터를 가져올 수 있습니다. JDBC 싱크 커넥터를 사용하여 Kafka topic에서 JDBC 드라이버가 있는 모든 관계형 데이터베이스로 데이터를 내보낼 수 있습니다. JDBC 커넥터는 다양한 데이터베이스를 지원합니다.

confluent-hub install confluentinc/kafka-connect-jdbc:latest

confluent-hub install confluentinc/kafka-connect-jdbc:5.5.1

connect-distributed: Caused by: javax.security.auth.login.LoginException: Unable to obtain Principal Name for authentication

export KAFKA_OPTS="-Djavax.security.auth.useSubjectCredsOnly=false"

bin/connect-distributed etc/kafka/connect-distributed.properties

사용자 정의 함수

ksqlDB는 내장 함수를 제공합니다.

ksqlDB에 없는 기능을 사용해야 하는 경우에 사용자 정의 함수를 사용하면 Java 후크를 사용하여 새 함수를 추가할 수 있습니다. 

방법 가이드 에서 사용법을 배우 거나 참조 문서 에서 중요한 정보를 찾아보십시오.

Time and Windows

In ksqlDB, a record is an immutable representation of an event in time. Each record carries a timestamp, which determines its position on the time axis.

This is the default timestamp that ksqlDB uses for processing the record. The timestamp is set either by the producer application or by the Apache Kafka® broker, depending on the topic's configuration. Records may be out-of-order within the stream.

Timestamps are used by time-dependent operations, like aggregations and joins.

Time semantics¶

Timestamps have different meanings, depending on the implementation. A record's timestamp can refer to the time when the event occurred, or when the record was ingested into Kafka, or when the record was processed. These times are event-time, ingestion-time, and processing-time.

Event-time¶

The time when a record is created by the data source. Achieving event-time semantics requires embedding timestamps in records when an event occurs and the record is produced.

For example, if the record is a geo-location change reported by a GPS sensor in a car, the associated event-time is the time when the GPS sensor captured the location change.

Ingestion-time¶

The time when a record is stored in a topic partition by a Kafka broker. Ingestion-time is similar to event-time, as a timestamp is embedded in the record, but the ingestion timestamp is generated when the Kafka broker appends the record to the target topic.

Ingestion-time can approximate event-time if the time difference between the creation of the record and its ingestion into Kafka is small.

For use cases where event-time semantics aren't possible, ingestion-time may be an alternative. Consider using ingestion-time when data producers don't embed timestamps in records, as in older versions of Kafka's Java producer client, or when the producer can't assign timestamps directly, like when it doesn't have access to a local clock.

Processing-time¶

The time when the record is consumed by a stream processing application. The processing-time can occur immediately after ingestion-time, or it may be delayed by milliseconds, hours, days, or longer.

For example, imagine an analytics application that reads and processes the geo-location data reported from car sensors, and presents it to a fleet-management dashboard. In this case, processing-time in the analytics application might be many minutes or hours after the event-time, as cars can move out of mobile reception for periods of time and have to buffer records locally.

Stream-time¶

The maximum timestamp seen over all processed records so far.

Timestamp assignment¶

A record's timestamp is set either by the record's producer or by the Kafka broker, depending on the topic's timestamp configuration. The topic's message.timestamp.type setting can be either CreateTime or LogAppendTime.

• CreateTime: The broker uses the the record's timestamp as set by the producer. This setting enforces event-time semantics.
• LogAppendTime: The broker overwrites the record's timestamp with the broker's local time when it appends the record to the topic's log. This setting enforces ingestion-time semantics. If LogAppendTime is configured, the producer has no control over the timestamp.

ksqlDB doesn't support processing-time operations directly, but you can implement user-defined functions (UDFs) that access the current time. For more information, see Functions.

By default, when ksqlDB imports a topic to create a stream, it uses the record's timestamp, but you can add the WITH(TIMESTAMP='some-field') clause to use a different field from the record's value as the timestamp. The optional TIMESTAMP_FORMAT property indicates how ksqlDB should parse the field. The field you specify can be an event-time or an ingestion-time. This approach implements payload-time semantics.

When working with time you should also make sure that additional aspects of time, like time zones and calendars, are correctly synchronized — or at least understood and traced — throughout your streaming data pipelines. It helps to agree on specifying time information in UTC or in Unix time, like seconds since the Unix epoch, everywhere in your system.

Timestamps of ksqlDB output streams¶

When a ksqlDB application writes new records to Kafka, timestamps are assigned to the records it creates. ksqlDB uses the underlying Kafka Streams implementation for computing timestamps. Timestamps are assigned based on context:

• When new output records are generated by processing an input record directly, output record timestamps are inherited from input record timestamps.
• When new output records are generated by a periodic function, the output record timestamp is defined as the current internal time of the stream task.
• For stateless operations, the input record timestamp is passed through. For flatMap and siblings that emit multiple records, all output records inherit the timestamp from the corresponding input record.

Timestamps for aggregations and joins¶

For aggregations and joins, timestamps are computed by using the following rules.

• For joins (stream-stream, table-table) that have left and right input records, the timestamp of the output record is assigned max(left.ts, right.ts).
• For stream-table joins, the output record is assigned the timestamp from the stream record.
• For aggregations, the max timestamp is computed over all records, per key, either globally (for non-windowed) or per-window.

Producers and timestamps¶

A producer application can set the timestamp on its records to any value, but usually, it choses a sensible event-time or the current wall-clock time.

If the topic's message.timestamp.type configuration is set to CreateTime, the following holds for the producer:

• When a producer record is created, it contains no timestamp, by default.
• The producer can set the timestamp on the record explicitly.
• If the timestamp isn't set when the producer application calls the producer.send() method, the current wall-clock time is set automatically.

In all three cases, the time semantics are considered to be event-time.

Timestamp extractors¶

When ksqlDB imports a topic to create a stream, it gets the timestamp from the topic's messages by using a timestamp extractor class. Timestamp extractors implement the TimestampExtractor interface.

Concrete implementations of timestamp extractors may retrieve or compute timestamps based on the actual contents of data records, like an embedded timestamp field, to provide event-time or ingestion-time semantics, or they may use any other approach, like returning the current wall-clock time at the time of processing to implement processing-time semantics.

By creating a custom timestamp extractor class, you can enforce different notions or semantics of time, depending on the requirements of your business logic. For more information see default.timestamp.extractor.

Windows in SQL queries¶

Representing time consistently enables aggregation operations on streams and tables, like SUM, that have distinct time boundaries. In ksqlDB, these boundaries are named windows.

A window has a start time and an end time, which you access in your queries by using the WINDOWSTART and WINDOWEND system columns.

Windowing lets you control how to group records that have the same key for stateful operations, like aggregations or joins, into time spans. ksqlDB tracks windows per record key.

When using windows in your SQL queries, aggregate functions are applied only to the records that occur within a specific time window. Records that arrive out-of-order are handled as you might expect: although the window end time has passed, the out-of-order records are still associated with the correct window.

Window types¶

There are three ways to define time windows in ksqlDB: hopping windows, tumbling windows, and session windows. Hopping and tumbling windows are time windows, because they're defined by fixed durations they you specify. Session windows are dynamically sized based on incoming data and defined by periods of activity separated by gaps of inactivity.

Hopping window¶

Hopping windows are based on time intervals. They model fixed-sized, possibly overlapping windows. A hopping window is defined by two properties: the window's duration and its advance, or "hop", interval. The advance interval specifies how far a window moves forward in time relative to the previous window. For example, you can configure a hopping window with a duration of five minutes and an advance interval of one minute. Because hopping windows can overlap, and usually they do, a record can belong to more than one such window.

All hopping windows have the same duration, but they might overlap, depending on the length of time specified in the ADVANCE BY property.

For example, if you want to count the pageviews for only Region_6 by female users for a hopping window of 30 seconds that advances by 10 seconds, you might run a query like this:

The hopping window's start time is inclusive, but the end time is exclusive. This is important for non-overlapping windows, in which each record must be contained in exactly one window.

Tumbling window¶

Tumbling windows are a special case of hopping windows. Like hopping windows, tumbling windows are based on time intervals. They model fixed-size, non-overlapping, gap-less windows. A tumbling window is defined by a single property: the window's duration. A tumbling window is a hopping window whose window duration is equal to its advance interval. Since tumbling windows never overlap, a record will belong to one and only one window.

All tumbling windows are the same size and adjacent to each other, which means that whenever a window ends, the next window starts.

For example, if you want to compute the the five highest-value orders per zip code per hour in an orders stream, you might run a query like this:

Here's another example: to detect potential credit card fraud in an authorization_attempts stream, you might run a query for the number of authorization attempts on a particular card that's greater than three, during a time interval of five seconds.

The tumbling window's start time is inclusive, but the end time is exclusive. This is important for non-overlapping windows, in which each record must be contained in exactly one window.

Session window¶

A session window aggregates records into a session, which represents a period of activity separated by a specified gap of inactivity, or "idleness". Any records with timestamps that occur within the inactivity gap of existing sessions are merged into the existing sessions. If a record's timestamp occurs outside of the session gap, a new session is created.

A new session window starts if the last record that arrived is further back in time than the specified inactivity gap.

Session windows are different from the other window types, because:

• ksqlDB tracks all session windows independently across keys, so windows of different keys typically have different start and end times.
• Session window durations vary. Even windows for the same key typically have different durations.

Session windows are especially useful for user behavior analysis. Session-based analyses range from simple metrics, like counting user visits on a news website or social platform, to more complex metrics, like customer-conversion funnel and event flows.

For example, to count the number of pageviews per region for session windows with a session inactivity gap of 60 seconds, you might run the following query, which sessionizes the input data and performs the counting/aggregation step per region:

The start and end times for a session window are both inclusive, in contrast to time windows.

A session window contains at least one record. It's not possible for a session window to have zero records.

If a session window contains exactly one record, the record's ROWTIME timestamp is identical to the window's own start and end times. Access these by using the WINDOWSTART and WINDOWEND system columns.

If a session window contains two or more records, then the earliest/oldest record's ROWTIME timestamp is identical to the window's start time, and the latest/newest record's ROWTIME timestamp is identical to the window's end time.

Windowed joins¶

ksqlDB supports using windows in JOIN queries by using the WITHIN clause.

For example, to find orders that have shipped within the last hour from an orders stream and a shipments stream, you might run a query like:

For more information on joins, see Join Event Streams with ksqlDB.

Out-of-order events¶

Frequently, events that belong to a window can arrive out-of-order, for example, over slow networks, and a grace period may be required to ensure the events are accepted into the window. ksqlDB enables configuring this behavior, for each of the window types.

For example, to allow events to be accepted for up to two hours after the window ends, you might run a query like:

Events that arrive after the grace period has passed are called late and aren't included in the aggregation result.

Window retention¶

For each window type, you can configure the number of windows in the past that ksqlDB retains. This capability is very useful for interactive applications that use ksqlDB as their primary serving data store.

For example, to retain the computed windowed aggregation results for a week, you might run the following query:

Note that the specified retention period should be larger than the sum of window size and any grace period.

Join Index

• Join Event Streams
• Partition Data to Enable Joins
• Synthetic key columns

Joining collections

Partitioning requirements

Synthetic key columns

쿼리

ksqlDB에는 영구(persistent), 푸시(push), 풀(pull)의 세 가지 쿼리가 있습니다. 

영구 쿼리

영구 쿼리는 이벤트 행을 무기한으로 실행하는 서버 측 쿼리입니다. 기존 스트림 또는 테이블에서 새 스트림 및 새 테이블 을 만들어 영구 쿼리를 실행합니다.

Push 쿼리

푸시 쿼리 는 실시간으로 변하는 이벤트를 구독하는  클라이언트에 의해 발행된 쿼리입니다.

푸시 쿼리의 예로 특정 사용자의 지리적 위치를 구독하는 것입니다. 쿼리는 지도 좌표를 요청하는 푸시 쿼리이기 때문에 위치 변경이 발생하는 즉시 클라이언트에 "푸시"됩니다. 이는 프로그래밍 방식으로 제어되는 마이크로서비스, 실시간 앱 또는 모든 종류의 비동기 제어 흐름을 빌드하는 데 유용합니다.

푸시 쿼리는 SQL과 유사한 언어를 사용하여 표현됩니다. 특정 키에 대한 스트림이나 테이블을 쿼리하는 데 사용할 수 있습니다. 또한 푸시 쿼리는 키 조회에만 국한되지 않습니다. 필터, 선택, 그룹화 기준, 파티션 기준 및 조인을 포함한 전체 SQL 세트를 지원합니다.

푸시 쿼리를 사용하면 결과에 대한 구독으로 스트림 또는 구체화된 테이블을 쿼리할 수 있습니다. 스트림을 반환하는 쿼리를 포함하여 모든 쿼리의 출력을 구독할 수 있습니다. 푸시 쿼리는 스트림 또는 구체화된 테이블로 실시간으로 새로운 정보에 조회할 수 있습니다. 비동기식 애플리케이션 흐름에 적합합니다. 요청/응답 흐름은 풀 쿼리를 참조하십시오 .

ksqlDB REST API에 HTTP 요청을 전송하여 푸시 쿼리를 실행하면 API가 무기한 길이의 청크 응답을 다시 보냅니다.

푸시 쿼리의 결과는 지원 Kafka topic에 유지되지 않습니다. Kafka 주제에 대한 쿼리 결과를 유지해야 하는 경우 CREATE TABLE AS SELECT 또는 CREATE STREAM AS SELECT 문을 사용합니다.

Pull 쿼리

풀 쿼리는 기존 RDBS에 대한 쿼리와 같이 "지금"의 결과를 검색하는 클라이언트가 수행한 쿼리입니다.

특정 사용자에 대한 지리적 위치에 대한 풀 쿼리는 현재 지도 좌표를 요청합니다. 풀 쿼리이기 때문에 현재까지의 데이터를 반환되고 연결이 닫힙니다. 이는 페이지 로드 시 사용자 인터페이스를 한 번 렌더링하는 데 이상적입니다. 일반적으로 모든 종류의 동기 제어 흐름에 적합합니다.

풀 쿼리를 사용하면 구체화된 뷰의 현재 상태를 가져올 수 있습니다. 구체화된 뷰는 새 이벤트가 도착하면 점진적으로 업데이트되기 때문에 풀 쿼리는 짧은 수행 시간으로 실행됩니다. 요청/응답 흐름에 매우 적합합니다. 비동기식 애플리케이션 흐름의 경우 푸시 쿼리를 참조하십시오 . 풀 쿼리는 ANSI SQL을 따릅니다.

ksqlDB REST API에 HTTP 요청을 전송하여 pull 쿼리를 실행하면 API가 단일 응답으로 응답합니다.

https://noti.st/rmoff/NrxcrM/slides 참고

테이블

테이블은 시간이 지남에 따라 데이터가 변경되는 컬렉션입니다. 

이벤트의 시간적 순서를 나타내는 스트림과 달리 테이블은 "지금" 현재의 사실을 나타냅니다.

예를 들어, 테이블을 사용하여 누군가가 살았던 장소를 모델링 하면,

 살았던 장소가 "서울 -> 제주도 -> 대전" 순으로 변경된 경우 테이블에는 최종 데이터인 "대전" 데이터를 가집니다.

테이블은 각 행의 키를 활용하여 작동합니다. 일련의 행이 키가 같은 경우,  데이터는 해당 키의 최신 정보를 나타냅니다. 

EVENT - STREAM - TABLE 관계

테이블 생성

스트림

스트림은 일련의 발생 사실을 나타내는 컬렉션입니다. 분할되고 변경 불가능하고 추가만 가능합니다.

이벤트의 집합으로 생각하면 됩니다.

예를 들어, 스트림의 행은 "Alice가 Bob에게 $100를 보냈습니다", "Charlie가 Bob에게 $50를 보냈습니다"와 같은 일련의 금융 거래를 모델링할 수 있습니다.

이벤트가 스트림에 입력되면 절대 변경할 수 없습니다. 스트림 끝에 새 행을 추가할 수 있지만 기존 행은 업데이트하거나 삭제할 수 없습니다.

각 행은 특정 파티션에 저장됩니다. 모든 행에는 암시적이든 명시적이든 해당 ID를 가르키는 키가 있습니다. 

동일한 키를 가진 모든 행은 동일한 파티션에 있습니다.