Streaming aggregations only work correctly when you can tell the system when a time window is truly "done." This lesson teaches you how watermarks, tumbling, sliding, and session windows work together — including late data strategies and production-grade Flink code.
Picture this: your e-commerce platform processes clickstream events from millions of users. Your analytics team wants a real-time dashboard showing revenue per product category, updated every five minutes. Simple enough — until you realize that mobile events from users on spotty connections can arrive 90 seconds late, your payment confirmation events sometimes lag the original purchase by several minutes, and during flash sales your Kafka consumers fall behind and replay events out of order. Now what does "every five minutes" even mean?
This is the central problem of streaming windowing. In batch processing, you have all the data before you compute anything. In streaming, data arrives continuously and often out of order, and you need to decide when to close a window and emit a result — knowing full well that some events that belong in that window haven't shown up yet. Get this wrong in one direction and your aggregations are perpetually incomplete; get it wrong in the other and you're waiting so long that your "real-time" dashboard is essentially batch processing with extra steps.
By the end of this lesson, you'll know how to implement all three major window types in Apache Flink (with patterns transferable to Spark Structured Streaming and Beam), configure watermarks to handle out-of-order data gracefully, decide what to do with genuinely late records, and build a production-grade pipeline that emits correct aggregations without sacrificing timeliness.
What you'll learn:
You should be comfortable with the core concepts covered in Batch vs. Stream Processing: Choosing the Right Ingestion Pattern for Your Pipeline and have some familiarity with incremental loading patterns including watermarks. Basic Python or Java/Scala knowledge is assumed. The code examples here use Python with Apache Flink's PyFlink API, but the concepts map directly to Flink's Java API and to Spark Structured Streaming.
Before diving into window types, you need to internalize the distinction between event time and processing time — because everything else in this lesson depends on it.
"timestamp": "2024-01-15T14:23:47Z").In a perfect world, these are milliseconds apart. In production, the gap can be minutes or hours. Network partitions, mobile clients going offline, IoT devices with clock skew, retry storms after downstream failures — all of these create events that arrive at your pipeline long after they occurred.
Processing time windowing is tempting because it's simple: just use the wall clock. But it produces results that are fundamentally wrong for any use case where correctness matters. A five-minute revenue window that closes based on processing time might capture events that happened at 14:18 alongside events that happened at 14:25, purely because of ingestion delays. Your finance team will not appreciate that.
Event time windowing gives you what actually happened, when it actually happened — but it requires a mechanism to know when you've seen "enough" of the events for a given time period. That mechanism is the watermark.
Key insight: Always use event time when the correctness of aggregations matters. Processing time is only appropriate for metrics about your infrastructure itself (e.g., "how many events did we process per second?") not about the business events flowing through it.
A watermark is a signal in the event stream that says: "I am confident that no event with a timestamp earlier than T will arrive from now on." When your streaming system receives a watermark for time T, it can safely close any windows that end at or before T and emit their results.
Watermarks are generated by your pipeline, typically based on the maximum event timestamp seen so far, minus a configured lag. That lag represents your tolerance for out-of-order data.
# PyFlink watermark strategy configuration
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.connectors.kafka import KafkaSource
from pyflink.common import WatermarkStrategy, Duration
from pyflink.common.watermark_strategy import TimestampAssigner
env = StreamExecutionEnvironment.get_execution_environment()
# BoundedOutOfOrderness watermark strategy
# Tells Flink: events can arrive up to 30 seconds late
watermark_strategy = (
WatermarkStrategy
.for_bounded_out_of_orderness(Duration.of_seconds(30))
.with_timestamp_assigner(
# Extract event timestamp from your payload
lambda event, record_timestamp: event["event_ts_ms"]
)
)
kafka_source = (
KafkaSource.builder()
.set_bootstrap_servers("kafka-broker:9092")
.set_topics("ecommerce.clickstream")
.set_group_id("windowing-pipeline")
.set_value_only_deserializer(...)
.build()
)
stream = env.from_source(
kafka_source,
watermark_strategy,
"Clickstream Kafka Source"
)
The for_bounded_out_of_orderness(Duration.of_seconds(30)) strategy generates a watermark equal to max_seen_event_time - 30 seconds. So when Flink has seen an event at 14:30:00, it emits a watermark at 14:29:30, signaling that windows ending before 14:29:30 can be safely closed.
How do you choose the lag value? Look at your data. In production, instrument your pipeline to track the difference between event time and ingestion time across a representative sample of traffic. The 99th percentile of that distribution is a reasonable starting point. If 99% of your events arrive within 45 seconds of their event time, use 45–60 seconds as your bounded out-of-orderness.
Warning: Setting your watermark lag too low means late events miss their windows and get dropped or routed to side outputs. Setting it too high increases end-to-end latency because windows won't close until the watermark catches up. This is a genuine tradeoff — there's no free lunch.
One subtlety that trips people up in production: in a parallel pipeline, each parallel subtask maintains its own watermark. The effective global watermark is the minimum across all parallel instances. This means if one of your Kafka partitions goes quiet (no events), its watermark stops advancing, and your entire pipeline's effective watermark stalls.
# Configure idle timeout to handle quiet partitions
watermark_strategy = (
WatermarkStrategy
.for_bounded_out_of_orderness(Duration.of_seconds(30))
.with_idleness(Duration.of_minutes(1)) # Mark idle sources after 1 minute
.with_timestamp_assigner(
lambda event, record_timestamp: event["event_ts_ms"]
)
)
The with_idleness setting tells Flink to exclude idle sources from the minimum watermark calculation, preventing a stalled partition from freezing all your windows.
Tumbling windows are fixed-size, non-overlapping intervals. Every event belongs to exactly one window. Think of them as dividing the event time axis into equal-length buckets: 00:00–05:00, 05:00–10:00, 10:00–15:00, and so on.
They're the right choice when you want a clean, non-overlapping summary — revenue per five-minute interval, error counts per hour, API call volume per minute.
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.common import Time
# Aggregate clickstream events into 5-minute revenue buckets
# grouped by product category
revenue_stream = (
stream
.map(lambda event: (
event["product_category"],
event["purchase_amount"],
event["event_ts_ms"]
))
.key_by(lambda x: x[0]) # Key by product category
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.apply(CategoryRevenueWindowFunction())
)
from pyflink.datastream import WindowFunction
from pyflink.datastream.window import TimeWindow
class CategoryRevenueWindowFunction(WindowFunction):
def apply(self, key, window: TimeWindow, inputs, out):
total_revenue = sum(event[1] for event in inputs)
event_count = len(list(inputs))
# Emit one record per window per category
out.collect({
"category": key,
"window_start": window.start,
"window_end": window.end,
"total_revenue_usd": total_revenue,
"order_count": event_count,
"avg_order_value": total_revenue / event_count if event_count > 0 else 0
})
The window fires when the watermark passes the window's end time. If your five-minute window ends at 14:25:00 and your watermark lag is 30 seconds, the window fires when Flink sees an event with timestamp 14:25:30 or later (which advances the watermark to 14:25:00+).
Tip: Tumbling windows align to the epoch by default. A 5-minute window starting the pipeline at 14:23 will create windows 14:20–14:25, 14:25–14:30, etc. — not 14:23–14:28. This is usually what you want for reporting consistency, but you can specify an offset:
TumblingEventTimeWindows.of(Time.hours(1), Time.hours(-8))for UTC-8 alignment.
If your downstream consumer needs to answer "how many orders in the last 5 minutes from right now?" with right now being any arbitrary point in time, a tumbling window gives you a stale answer. You'd get the result for the most recently closed window, which could be almost 5 minutes ago. That's when you reach for sliding windows.
Sliding windows have a fixed size but advance at a shorter interval, creating overlapping windows. You define two parameters: the window size and the slide (how frequently a new window starts). Each event belongs to size/slide windows simultaneously.
A sliding window of size 10 minutes with a slide of 1 minute means every minute you get a fresh 10-minute aggregate. If a purchase event occurs at 14:23:47, it will appear in the windows covering 14:14–14:24, 14:15–14:25, 14:16–14:26... all the way to 14:23–14:33.
from pyflink.datastream.window import SlidingEventTimeWindows
# 10-minute revenue rolling window, updated every 1 minute
# Useful for "revenue in the last 10 minutes" on a live dashboard
rolling_revenue = (
stream
.filter(lambda event: event.get("event_type") == "purchase")
.map(lambda event: (
event["product_category"],
float(event["purchase_amount"])
))
.key_by(lambda x: x[0])
.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
.apply(RollingRevenueFunction())
)
Warning: Sliding windows are memory-hungry. Because each event belongs to multiple windows, Flink must maintain state for each overlapping window per key. A 1-hour window with a 1-minute slide means each event is tracked in 60 concurrent windows. For high-cardinality keys (e.g., per-user metrics with millions of users), this can exhaust your job manager's heap. Profile your state backend before deploying to production.
The computational cost scales as size/slide. A 60-minute window sliding every minute has a 60× overhead compared to a 60-minute tumbling window. For many use cases, you can approximate a rolling window using a tumbling window + downstream aggregation, which is significantly cheaper.
The slide interval determines your dashboard's "freshness." For a real-time fraud detection system checking transaction velocity (e.g., more than 5 transactions per card in 10 minutes), a 1-minute slide gives you a result that's at most 1 minute stale. For a less time-sensitive report like hourly active users updated every 15 minutes, a 15-minute slide against a 1-hour window is perfectly adequate and much cheaper.
Session windows are the most interesting and the most misunderstood of the three types. Unlike tumbling and sliding windows, session windows have no fixed size. Instead, they group events that are temporally close together into a single session, with the window closing when there's a gap in activity longer than a configured timeout.
This maps directly to user behavior modeling. A user browsing your site might generate 15 events over 8 minutes, go to lunch, then return 45 minutes later and generate 10 more events. Those are two distinct sessions with different intent, and you'd want to analyze them separately.
from pyflink.datastream.window import EventTimeSessionWindows
# Group user interactions into sessions
# A new session starts if the user is inactive for more than 20 minutes
user_sessions = (
stream
.filter(lambda event: event["event_type"] in [
"page_view", "product_click", "add_to_cart", "purchase"
])
.map(lambda event: (
event["user_id"],
event["event_type"],
event["product_id"],
event["event_ts_ms"]
))
.key_by(lambda x: x[0]) # Key by user_id
.window(EventTimeSessionWindows.with_gap(Time.minutes(20)))
.apply(UserSessionAnalyticsFunction())
)
class UserSessionAnalyticsFunction(WindowFunction):
def apply(self, user_id, window: TimeWindow, inputs, out):
events = sorted(inputs, key=lambda x: x[3]) # Sort by timestamp
event_sequence = [e[1] for e in events]
converted = any(e[1] == "purchase" for e in events)
products_viewed = list(set(e[2] for e in events if e[2]))
session_duration_seconds = (window.end - window.start) / 1000
out.collect({
"user_id": user_id,
"session_start": window.start,
"session_end": window.end,
"session_duration_seconds": session_duration_seconds,
"event_count": len(events),
"converted": converted,
"products_viewed": products_viewed,
"funnel_path": " -> ".join(event_sequence)
})
Session windows shine for:
Key insight: Session windows are dynamic — Flink merges windows as new events arrive. If a user generates an event at 14:00, then at 14:15, then at 14:30, with a 20-minute gap, all three events belong to the same session (each new event is within 20 minutes of the previous). But if the next event arrives at 14:51, that starts a new session. Flink handles this merging automatically using its state backend.
Because session windows can grow unboundedly (a single non-stop user session could technically last hours), they can accumulate significant state per key. For applications with millions of users, configure your state backend with TTL to prevent state from growing indefinitely:
from pyflink.datastream import StateTtlConfig
from pyflink.common import Time as FlinkTime
# Configure state TTL for session window state
# State older than 24 hours will be cleaned up
ttl_config = (
StateTtlConfig
.new_builder(FlinkTime.hours(24))
.set_update_type(StateTtlConfig.UpdateType.OnCreateAndWrite)
.set_state_visibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
.build()
)
Even with a well-chosen watermark lag, some events will arrive after their window has closed. Your watermark is a tradeoff, not a guarantee. The question is: what should your pipeline do with those stragglers?
Flink gives you three options:
In most production scenarios, option 1 is unacceptable for anything business-critical, and option 2 alone is insufficient because you don't know what happened to truly late data. Option 3 combined with option 2 gives you the most flexibility.
from pyflink.datastream import OutputTag
# Define a side output tag for late events
late_events_tag = OutputTag("late-clickstream-events")
# Main stream with allowed lateness and side output for late events
windowed_stream = (
stream
.key_by(lambda event: event["product_category"])
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.allowed_lateness(Time.minutes(2)) # Accept events up to 2 min after window closes
.side_output_late_data(late_events_tag) # Route truly late events to side output
.apply(CategoryRevenueWindowFunction())
)
# The main stream: on-time results (and updated results within allowed lateness)
windowed_stream.add_sink(revenue_sink)
# The side output: events that arrived after allowed lateness expired
late_stream = windowed_stream.get_side_output(late_events_tag)
late_stream.add_sink(late_events_sink) # Route to DLQ, audit table, or correction pipeline
With allowed_lateness(Time.minutes(2)), after the watermark passes the window end time (triggering the initial fire), Flink keeps the window state alive for an additional 2 minutes. Any late event arriving within that grace period triggers a refire of the window with an updated result. After 2 minutes, the window state is purged and any subsequent events go to the side output.
Tip: Think of allowed lateness as the difference between your P99 and P99.9 latency. Your watermark lag handles the 99th percentile of late data. Allowed lateness is the safety net for the remaining 0.9%. Keep it small enough that your state doesn't balloon, but large enough to catch meaningful stragglers.
The late events side output deserves special attention. This stream is invaluable for debugging pipeline quality issues — if you're seeing high volumes of late events, it's a signal that your watermark strategy needs recalibration. Route these to a separate Kafka topic or database table and monitor their volume as an operational metric. For ideas on how to handle these problematic records systematically, see the patterns covered in implementing dead letter queues and poison message handling.
The default trigger fires a window exactly once when the watermark passes the window end. But production dashboards often need lower latency than your watermark lag allows. If your lag is 30 seconds, your five-minute tumbling windows won't fire until at least 5 minutes 30 seconds of event time has passed. For a live dashboard, that's acceptable. For a fraud alert system, it might not be.
Flink's trigger system lets you fire windows early based on either event count or processing time:
from pyflink.datastream.window import TumblingEventTimeWindows
from pyflink.datastream.triggers import (
EventTimeTrigger,
ContinuousProcessingTimeTrigger,
PurgingTrigger
)
# Fire early every 30 seconds of processing time (for dashboard freshness)
# Fire finally when the watermark passes the window end (for accuracy)
windowed_stream = (
stream
.key_by(lambda event: event["product_category"])
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.trigger(
ContinuousProcessingTimeTrigger.of(Time.seconds(30))
)
.apply(CategoryRevenueWithFiringTypeFunction())
)
When using early triggers, your window function fires multiple times for the same window. You need to communicate to downstream systems whether a given result is preliminary or final. A common pattern is to include a is_final flag and a firing_type field in your output schema:
class CategoryRevenueWithFiringTypeFunction(WindowFunction):
def apply(self, key, window: TimeWindow, inputs, out):
total_revenue = sum(event[1] for event in inputs)
event_count = len(list(inputs))
import time
current_watermark_estimate = time.time() * 1000
is_final = current_watermark_estimate >= window.end
out.collect({
"category": key,
"window_start": window.start,
"window_end": window.end,
"total_revenue_usd": total_revenue,
"order_count": event_count,
"is_final": is_final,
"result_type": "FINAL" if is_final else "EARLY",
"emitted_at_ms": int(time.time() * 1000)
})
Warning: Early triggers mean your consumers will receive multiple results for the same window key. Your sink must be designed to handle this — typically using an upsert pattern (update the row if it already exists) rather than an append. If your downstream database doesn't support upserts, you'll need to implement deduplication logic. See designing idempotent data pipelines for patterns that handle exactly this scenario.
Windowed streaming pipelines are stateful — Flink must remember the accumulated events in each window between checkpoints. Without proper checkpointing configuration, a job restart means losing all in-flight window state, which causes incorrect final results (or none at all).
from pyflink.datastream import CheckpointingMode
# Configure checkpointing for fault tolerance
env.enable_checkpointing(60000) # Checkpoint every 60 seconds
env.get_checkpoint_config().set_checkpointing_mode(
CheckpointingMode.EXACTLY_ONCE
)
env.get_checkpoint_config().set_min_pause_between_checkpoints(30000)
env.get_checkpoint_config().set_checkpoint_timeout(120000)
# Use RocksDB state backend for large state (many keys/long windows)
from pyflink.datastream.state_backend import EmbeddedRocksDBStateBackend
env.set_state_backend(EmbeddedRocksDBStateBackend())
For windowed pipelines with long window durations or high key cardinality (like per-user session windows with millions of users), the RocksDB state backend is almost always the right choice over the in-memory HashMapStateBackend. RocksDB spills state to local disk, trading some latency for dramatically higher state capacity.
The checkpointing and state management topic runs deep — for a thorough treatment, see the dedicated lesson on checkpointing and state management in long-running data pipelines.
Let's put everything together in a realistic scenario. You're building a streaming analytics pipeline for an e-commerce platform with these requirements:
Here's the complete pipeline structure:
from pyflink.datastream import StreamExecutionEnvironment, OutputTag
from pyflink.datastream.connectors.kafka import KafkaSource, KafkaSink
from pyflink.common import WatermarkStrategy, Duration, Time
from pyflink.datastream.window import (
TumblingEventTimeWindows,
SlidingEventTimeWindows,
EventTimeSessionWindows
)
import json
def build_ecommerce_analytics_pipeline():
env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(4)
env.enable_checkpointing(60_000)
# Watermark strategy: tolerate 45 seconds of out-of-order events
# Based on P99 measurement of our event ingestion latency
watermark_strategy = (
WatermarkStrategy
.for_bounded_out_of_orderness(Duration.of_seconds(45))
.with_idleness(Duration.of_minutes(2))
.with_timestamp_assigner(
lambda event, _: int(event.get("event_ts_ms", 0))
)
)
# Source: raw ecommerce events from Kafka
raw_stream = env.from_source(
build_kafka_source("ecommerce.events"),
watermark_strategy,
"Ecommerce Events Source"
).map(lambda bytes_val: json.loads(bytes_val.decode("utf-8")))
# Filter to purchase events only for revenue calculations
purchase_stream = raw_stream.filter(
lambda e: e.get("event_type") == "purchase"
and e.get("purchase_amount") is not None
)
# --- BRANCH 1: 5-minute tumbling window, revenue by category ---
late_purchases_tag = OutputTag("late-purchases")
category_revenue_5min = (
purchase_stream
.map(lambda e: (e["product_category"], float(e["purchase_amount"])))
.key_by(lambda x: x[0])
.window(TumblingEventTimeWindows.of(Time.minutes(5)))
.allowed_lateness(Time.minutes(2))
.side_output_late_data(late_purchases_tag)
.apply(CategoryRevenueTumblingFunction())
)
# Route late purchases to an audit topic
category_revenue_5min.get_side_output(late_purchases_tag) \
.map(lambda e: json.dumps({"late_event": e, "reason": "past_allowed_lateness"})) \
.add_sink(build_kafka_sink("ecommerce.late_events.audit"))
# Write primary results to revenue topic
category_revenue_5min \
.map(lambda r: json.dumps(r)) \
.add_sink(build_kafka_sink("ecommerce.revenue.5min"))
# --- BRANCH 2: 10-min sliding window (1-min slide) for live dashboard ---
rolling_revenue_10min = (
purchase_stream
.map(lambda e: (e["product_category"], float(e["purchase_amount"])))
.key_by(lambda x: x[0])
.window(SlidingEventTimeWindows.of(Time.minutes(10), Time.minutes(1)))
.apply(CategoryRevenueSlidingFunction())
)
rolling_revenue_10min \
.map(lambda r: json.dumps(r)) \
.add_sink(build_kafka_sink("ecommerce.revenue.rolling10min"))
# --- BRANCH 3: Session windows for user funnel analysis ---
user_sessions = (
raw_stream
.filter(lambda e: e.get("event_type") in [
"page_view", "product_click", "add_to_cart", "purchase"
])
.map(lambda e: (
e.get("user_id"),
e.get("event_type"),
e.get("product_id"),
int(e.get("event_ts_ms", 0))
))
.filter(lambda x: x[0] is not None) # Drop events without user_id
.key_by(lambda x: x[0])
.window(EventTimeSessionWindows.with_gap(Time.minutes(20)))
.apply(UserSessionFunction())
)
user_sessions \
.map(lambda r: json.dumps(r)) \
.add_sink(build_kafka_sink("ecommerce.user_sessions"))
env.execute("Ecommerce Analytics Windowing Pipeline")
if __name__ == "__main__":
build_ecommerce_analytics_pipeline()
The pipeline fans out from a single source into three branches, each with different windowing semantics. This is a common pattern — you read the stream once and derive multiple aggregated views simultaneously, which is far more efficient than running three separate pipelines against the same Kafka topic. For more on these branching patterns, see data pipeline design patterns: fan-out, fan-in, and branching workflows.
Symptom: Windows never fire. Your job is running but producing no output.
Cause: All events are arriving with timestamps that are too far behind the current time. Your watermark lag is larger than the difference between event time and the current time, so the watermark never advances past any window boundary. Alternatively, a single idle Kafka partition is holding back the global watermark.
Fix: Enable with_idleness() on your watermark strategy. Verify your timestamp extractor is returning milliseconds (not seconds — a Unix timestamp in seconds looks 1000× too small to Flink).
# Wrong: returning seconds
lambda event, _: event["event_ts"] # 1705324800
# Right: returning milliseconds
lambda event, _: event["event_ts"] * 1000 # 1705324800000
Symptom: Your five-minute windows are firing every few seconds with empty or minimal data.
Cause: Your timestamp extractor is returning 0 or a garbage value (e.g., None coerced to 0). All events get assigned timestamp 0, which means they're all in the same window that closed in 1970, and every new batch of events triggers an immediate fire.
Fix: Add validation in your timestamp extractor and log or filter events with invalid timestamps.
def extract_timestamp(event, record_timestamp):
ts = event.get("event_ts_ms")
if ts is None or ts <= 0:
# Return record_timestamp as fallback, or filter upstream
return record_timestamp
return int(ts)
Symptom: Your job's memory usage grows continuously. Eventually the job fails with an OutOfMemoryError or RocksDB write stalls.
Cause: High-cardinality keys (e.g., per-user-ID windows) combined with long window durations or long allowed lateness. If you have 5 million users and a 20-minute session window gap, you could have millions of concurrent open windows in state.
Fix: Switch to RocksDB state backend, configure state TTL, and reconsider your key cardinality. Do you really need per-user windows, or can you aggregate at a coarser level (e.g., per-cohort)?
Symptom: Your downstream database or dashboard shows doubled values, especially after a job restart.
Cause: Your sink isn't idempotent. Flink's at-least-once semantics (or failed exactly-once transactions) can cause windows to be re-emitted after recovery from a checkpoint.
Fix: Use upsert sinks with a composite key of (window_start, window_end, group_key). Or implement exactly-once delivery guarantees end-to-end using Flink's transactional Kafka sink.
Symptom: Your aggregations look correct but don't match what the business expects when replaying historical data.
Cause: You're using TumblingProcessingTimeWindows instead of TumblingEventTimeWindows. Processing time windows are determined by when the job runs, not when the events happened. When you replay yesterday's data today, everything lands in today's processing time windows.
Fix: Always double-check your import and constructor:
# Wrong for event-time windowing:
from pyflink.datastream.window import TumblingProcessingTimeWindows
# Right:
from pyflink.datastream.window import TumblingEventTimeWindows
You've now seen how all three pieces fit together: watermarks tell Flink when it's safe to close a window; window types (tumbling, sliding, session) define the shape of your aggregations; and allowed lateness plus side outputs give you a coherent strategy for data that doesn't play by the rules.
The key decisions you'll make for any windowed streaming pipeline are:
From here, the natural next topics to dig into are:
Windowing is one of those topics where the gap between "I understand it conceptually" and "I can run it reliably in production" is genuinely wide. The exercises and patterns in this lesson are the bridge. Build the e-commerce pipeline, break it deliberately, and watch how the late event counts and watermark metrics respond. That's when the mental model really locks in.