You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ja...@apache.org on 2018/11/14 02:17:24 UTC
[1/2] samza git commit: Updated API documentation for high and low
level APIs.
Repository: samza
Updated Branches:
refs/heads/master 1d6f693c8 -> 4e5907090
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/api/programming-model.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/api/programming-model.md b/docs/learn/documentation/versioned/api/programming-model.md
index 1c9bd1c..efdcfa3 100644
--- a/docs/learn/documentation/versioned/api/programming-model.md
+++ b/docs/learn/documentation/versioned/api/programming-model.md
@@ -18,74 +18,81 @@ title: Programming Model
See the License for the specific language governing permissions and
limitations under the License.
-->
-# Introduction
-Samza provides different sets of programming APIs to meet requirements from different sets of users. The APIs are listed below:
+### Introduction
+Samza provides multiple programming APIs to fit your use case:
-1. Java programming APIs: Samza provides Java programming APIs for users who are familiar with imperative programming languages. The overall programming model to create a Samza application in Java will be described here. Samza also provides two sets of APIs to describe user processing logic:
- 1. [High-level API](high-level-api.md): this API allows users to describe the end-to-end stream processing pipeline in a connected DAG (Directional Acyclic Graph). It also provides a rich set of build-in operators to help users implementing common transformation logic, such as filter, map, join, and window.
- 2. [Task API](low-level-api.md): this is low-level Java API which provides “bare-metal” programming interfaces to the users. Task API allows users to explicitly access physical implementation details in the system, such as accessing the physical system stream partition of an incoming message and explicitly controlling the thread pool to execute asynchronous processing method.
-2. [Samza SQL](samza-sql.md): Samza provides SQL for users who are familiar with declarative query languages, which allows the users to focus on data manipulation via SQL predicates and UDFs, not the physical implementation details.
-3. Beam API: Samza also provides a [Beam runner](https://beam.apache.org/documentation/runners/capability-matrix/) to run applications written in Beam API. This is considered as an extension to existing operators supported by the high-level API in Samza.
+1. Java APIs: Samza's provides two Java programming APIs that are ideal for building advanced Stream Processing applications.
+ 1. [High Level Streams API](high-level-api.md): Samza's flexible High Level Streams API lets you describe your complex stream processing pipeline in the form of a Directional Acyclic Graph (DAG) of operations on message streams. It provides a rich set of built-in operators that simplify common stream processing operations such as filtering, projection, repartitioning, joins, and windows.
+ 2. [Low Level Task API](low-level-api.md): Samza's powerful Low Level Task API lets you write your application in terms of processing logic for each incoming message.
+2. [Samza SQL](samza-sql.md): Samza SQL provides a declarative query language for describing your stream processing logic. It lets you manipulate streams using SQL predicates and UDFs instead of working with the physical implementation details.
+3. Apache Beam API: Samza also provides a [Apache Beam runner](https://beam.apache.org/documentation/runners/capability-matrix/) to run applications written using the Apache Beam API. This is considered as an extension to the operators supported by the High Level Streams API in Samza.
-The following sections will be focused on Java programming APIs.
-
-# Key Concepts for a Samza Java Application
-To write a Samza Java application, you will typically follow the steps below:
-1. Define your input and output streams and tables
-2. Define your main processing logic
+### Key Concepts
The following sections will talk about key concepts in writing your Samza applications in Java.
-## Samza Applications
-When writing your stream processing application using Java API in Samza, you implement either a [StreamApplication](javadocs/org/apache/samza/application/StreamApplication.html) or [TaskApplication](javadocs/org/apache/samza/application/TaskApplication.html) and define your processing logic in the describe method.
-- For StreamApplication:
+#### Samza Applications
+A [SamzaApplication](javadocs/org/apache/samza/application/SamzaApplication.html) describes the inputs, outputs, state, configuration and the logic for processing data from one or more streaming sources.
+
+You can implement a
+[StreamApplication](javadocs/org/apache/samza/application/StreamApplication.html) and use the provided [StreamApplicationDescriptor](javadocs/org/apache/samza/application/descriptors/StreamApplicationDescriptor) to describe the processing logic using Samza's High Level Streams API in terms of [MessageStream](javadocs/org/apache/samza/operators/MessageStream.html) operators.
{% highlight java %}
-
- public void describe(StreamApplicationDescriptor appDesc) { … }
+
+ public class MyStreamApplication implements StreamApplication {
+ @Override
+ public void describe(StreamApplicationDescriptor appDesc) {
+ // Describe your application here
+ }
+ }
{% endhighlight %}
+
+Alternatively, you can implement a [TaskApplication](javadocs/org/apache/samza/application/TaskApplication.html) and use the provided [TaskApplicationDescriptor](javadocs/org/apache/samza/application/descriptors/TaskApplicationDescriptor) to describe it using Samza's Low Level API in terms of per-message processing logic.
+
+
- For TaskApplication:
{% highlight java %}
- public void describe(TaskApplicationDescriptor appDesc) { … }
+ public class MyTaskApplication implements TaskApplication {
+ @Override
+ public void describe(TaskApplicationDescriptor appDesc) {
+ // Describe your application here
+ }
+ }
{% endhighlight %}
-## Descriptors for Data Streams and Tables
-There are three different types of descriptors in Samza: [InputDescriptor](javadocs/org/apache/samza/system/descriptors/InputDescriptor.html), [OutputDescriptor](javadocs/org/apache/samza/system/descriptors/OutputDescriptor.html), and [TableDescriptor](javadocs/org/apache/samza/table/descriptors/TableDescriptor.html). The InputDescriptor and OutputDescriptor are used to describe the physical sources and destinations of a stream, while a TableDescriptor is used to describe the physical dataset and IO functions for a table.
-Usually, you will obtain InputDescriptor and OutputDescriptor from a [SystemDescriptor](javadocs/org/apache/samza/system/descriptors/SystemDescriptor.html), which include all information about producer and consumers to a physical system. The following code snippet illustrate how you will obtain InputDescriptor and OutputDescriptor from a SystemDescriptor.
-{% highlight java %}
-
- public class BadPageViewFilter implements StreamApplication {
- @Override
- public void describe(StreamApplicationDescriptor appDesc) {
- KafkaSystemDescriptor kafka = new KafkaSystemDescriptor();
- InputDescriptor<PageView> pageViewInput = kafka.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
- OutputDescriptor<DecoratedPageView> pageViewOutput = kafka.getOutputDescriptor(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
+#### Streams and Table Descriptors
+Descriptors let you specify the properties of various aspects of your application from within it.
- // Now, implement your main processing logic
- }
- }
-
-{% endhighlight %}
+[InputDescriptor](javadocs/org/apache/samza/system/descriptors/InputDescriptor.html)s and [OutputDescriptor](javadocs/org/apache/samza/system/descriptors/OutputDescriptor.html)s can be used for specifying Samza and implementation-specific properties of the streaming inputs and outputs for your application. You can obtain InputDescriptors and OutputDescriptors using a [SystemDescriptor](javadocs/org/apache/samza/system/descriptors/SystemDescriptor.html) for your system. This SystemDescriptor can be used for specify Samza and implementation-specific properties of the producer and consumers for your I/O system. Most Samza system implementations come with their own SystemDescriptors, but if one isn't available, you
+can use the [GenericSystemDescriptor](javadocs/org/apache/samza/system/descriptors/GenericSystemDescriptor.html).
-You can also add a TableDescriptor to your application.
+A [TableDescriptor](javadocs/org/apache/samza/table/descriptors/TableDescriptor.html) can be used for specifying Samza and implementation-specific properties of a [Table](javadocs/org/apache/samza/table/Table.html). You can use a Local TableDescriptor (e.g. [RocksDbTableDescriptor](javadocs/org/apache/samza/storage/kv/descriptors/RocksDbTableDescriptor.html) or a [RemoteTableDescriptor](javadocs/org/apache/samza/table/descriptors/RemoteTableDescriptor).
+
+
+The following example illustrates how you can use input and output descriptors for a Kafka system, and a table descriptor for a local RocksDB table within your application:
{% highlight java %}
-
- public class BadPageViewFilter implements StreamApplication {
+
+ public class MyStreamApplication implements StreamApplication {
@Override
- public void describe(StreamApplicationDescriptor appDesc) {
- KafkaSystemDescriptor kafka = new KafkaSystemDescriptor();
- InputDescriptor<PageView> pageViewInput = kafka.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
- OutputDescriptor<DecoratedPageView> pageViewOutput = kafka.getOutputDescriptor(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
- TableDescriptor<String, Integer> viewCountTable = new RocksDBTableDescriptor(
- “pageViewCountTable”, KVSerde.of(new StringSerde(), new IntegerSerde()));
-
- // Now, implement your main processing logic
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("kafka")
+ .withConsumerZkConnect(ImmutableList.of("..."))
+ .withProducerBootstrapServers(ImmutableList.of("...", "..."));
+ KafkaInputDescriptor<PageView> kid =
+ ksd.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
+ KafkaOutputDescriptor<DecoratedPageView> kod =
+ ksd.getOutputDescriptor(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
+
+ RocksDbTableDescriptor<String, Integer> td =
+ new RocksDbTableDescriptor(“viewCounts”, KVSerde.of(new StringSerde(), new IntegerSerde()));
+
+ // Implement your processing logic here
}
}
@@ -93,21 +100,21 @@ You can also add a TableDescriptor to your application.
The same code in the above describe method applies to TaskApplication as well.
-## Stream Processing Logic
+#### Stream Processing Logic
-Samza provides two sets of APIs to define the main stream processing logic, high-level API and Task API, via StreamApplication and TaskApplication, respectively.
+Samza provides two sets of APIs to define the main stream processing logic, High Level Streams API and Low Level Task API, via StreamApplication and TaskApplication, respectively.
-High-level API allows you to describe the processing logic in a connected DAG of transformation operators, like the example below:
+High Level Streams API allows you to describe the processing logic in a connected DAG of transformation operators, like the example below:
{% highlight java %}
public class BadPageViewFilter implements StreamApplication {
@Override
public void describe(StreamApplicationDescriptor appDesc) {
- KafkaSystemDescriptor kafka = new KafkaSystemDescriptor();
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor();
InputDescriptor<PageView> pageViewInput = kafka.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
OutputDescriptor<DecoratedPageView> pageViewOutput = kafka.getOutputDescriptor(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
- TableDescriptor<String, Integer> viewCountTable = new RocksDBTableDescriptor(
+ RocksDbTableDescriptor<String, Integer> viewCountTable = new RocksDbTableDescriptor(
“pageViewCountTable”, KVSerde.of(new StringSerde(), new IntegerSerde()));
// Now, implement your main processing logic
@@ -120,7 +127,7 @@ High-level API allows you to describe the processing logic in a connected DAG of
{% endhighlight %}
-Task API allows you to describe the processing logic in a customized StreamTaskFactory or AsyncStreamTaskFactory, like the example below:
+Low Level Task API allows you to describe the processing logic in a customized StreamTaskFactory or AsyncStreamTaskFactory, like the example below:
{% highlight java %}
@@ -130,7 +137,7 @@ Task API allows you to describe the processing logic in a customized StreamTaskF
KafkaSystemDescriptor kafka = new KafkaSystemDescriptor();
InputDescriptor<PageView> pageViewInput = kafka.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
OutputDescriptor<DecoratedPageView> pageViewOutput = kafka.getOutputDescriptor(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
- TableDescriptor<String, Integer> viewCountTable = new RocksDBTableDescriptor(
+ RocksDbTableDescriptor<String, Integer> viewCountTable = new RocksDbTableDescriptor(
“pageViewCountTable”, KVSerde.of(new StringSerde(), new IntegerSerde()));
// Now, implement your main processing logic
@@ -142,11 +149,10 @@ Task API allows you to describe the processing logic in a customized StreamTaskF
{% endhighlight %}
-Details for [high-level API](high-level-api.md) and [Task API](low-level-api.md) are explained later.
+#### Configuration for a Samza Application
-## Configuration for a Samza Application
+To deploy a Samza application, you need to specify the implementation class for your application and the ApplicationRunner to launch your application. The following is an incomplete example of minimum required configuration to set up the Samza application and the runner. For additional configuration, see the Configuration Reference.
-To deploy a Samza application, you will need to specify the implementation class for your application and the ApplicationRunner to launch your application. The following is an incomplete example of minimum required configuration to set up the Samza application and the runner:
{% highlight jproperties %}
# This is the class implementing StreamApplication
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/api/samza-sql.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/api/samza-sql.md b/docs/learn/documentation/versioned/api/samza-sql.md
index 13b059f..7412f6c 100644
--- a/docs/learn/documentation/versioned/api/samza-sql.md
+++ b/docs/learn/documentation/versioned/api/samza-sql.md
@@ -87,7 +87,7 @@ Note: Samza sql console right now doesn’t support queries that need state, for
# Running Samza SQL on YARN
-The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the low level task API, high level API as well as Samza SQL.
+The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the Low Level Task API, High Level Streams API as well as Samza SQL.
This tutorial demonstrates a simple Samza application that uses SQL to perform stream processing.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/api/table-api.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/api/table-api.md b/docs/learn/documentation/versioned/api/table-api.md
index f5efa88..0a9c33c 100644
--- a/docs/learn/documentation/versioned/api/table-api.md
+++ b/docs/learn/documentation/versioned/api/table-api.md
@@ -181,7 +181,7 @@ join with a table and finally write the output to another table.
# Using Table with Samza Low Level API
-The code snippet below illustrates the usage of table in Samza low level API.
+The code snippet below illustrates the usage of table in Samza Low Level Task API.
{% highlight java %}
1 class SamzaTaskApplication implements TaskApplication {
@@ -273,8 +273,7 @@ The table below summarizes table metrics:
[`RemoteTable`](https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java)
provides a unified abstraction for Samza applications to access any remote data
-store through stream-table join in high-level API or direct access in low-level
-API. Remote Table is a store-agnostic abstraction that can be customized to
+store through stream-table join in High Level Streams API or direct access in Low Level Task API. Remote Table is a store-agnostic abstraction that can be customized to
access new types of stores by writing pluggable I/O "Read/Write" functions,
implementations of
[`TableReadFunction`](https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/TableReadFunction.java) and
@@ -283,7 +282,7 @@ interfaces. Remote Table also provides common functionality, eg. rate limiting
(built-in) and caching (hybrid).
The async APIs in Remote Table are recommended over the sync versions for higher
-throughput. They can be used with Samza with low-level API to achieve the maximum
+throughput. They can be used with Samza with Low Level Task API to achieve the maximum
throughput.
Remote Tables are represented by class
@@ -420,7 +419,7 @@ created during instantiation of Samza container.
The life of a table goes through a few phases
1. **Declaration** - at first one declares the table by creating a `TableDescriptor`. In both
- Samza high level and low level API, the `TableDescriptor` is registered with stream
+ Samza High Level Streams API and Low Level Task API, the `TableDescriptor` is registered with stream
graph, internally converted to `TableSpec` and in return a reference to a `Table`
object is obtained that can participate in the building of the DAG.
2. **Instantiation** - during planning stage, configuration is
@@ -436,7 +435,7 @@ The life of a table goes through a few phases
* In Samza high level API, all table instances can be retrieved from `TaskContext` using
table-id during initialization of a
[`InitableFunction`] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/operators/functions/InitableFunction.java).
- * In Samza low level API, all table instances can be retrieved from `TaskContext` using
+ * In Samza Low Level Task API, all table instances can be retrieved from `TaskContext` using
table-id during initialization of a
[`InitableTask`] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/task/InitableTask.java).
4. **Cleanup** -
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/connectors/eventhubs.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/connectors/eventhubs.md b/docs/learn/documentation/versioned/connectors/eventhubs.md
index 0be288b..9fdc861 100644
--- a/docs/learn/documentation/versioned/connectors/eventhubs.md
+++ b/docs/learn/documentation/versioned/connectors/eventhubs.md
@@ -126,7 +126,7 @@ In this section, we will walk through a simple pipeline that reads from one Even
4 MessageStream<KV<String, String>> eventhubInput = appDescriptor.getInputStream(inputDescriptor);
5 OutputStream<KV<String, String>> eventhubOutput = appDescriptor.getOutputStream(outputDescriptor);
- // Define the execution flow with the high-level API
+ // Define the execution flow with the High Level Streams API
6 eventhubInput
7 .map((message) -> {
8 System.out.println("Received Key: " + message.getKey());
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/connectors/kafka.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/connectors/kafka.md b/docs/learn/documentation/versioned/connectors/kafka.md
index b71c736..9628130 100644
--- a/docs/learn/documentation/versioned/connectors/kafka.md
+++ b/docs/learn/documentation/versioned/connectors/kafka.md
@@ -24,9 +24,9 @@ Samza offers built-in integration with Apache Kafka for stream processing. A com
The `hello-samza` project includes multiple examples on interacting with Kafka from your Samza jobs. Each example also includes instructions on how to run them and view results.
-- [High-level API Example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/FilterExample.java) with a corresponding [tutorial](/learn/documentation/{{site.version}}/deployment/yarn.html#starting-your-application-on-yarn)
+- [High Level Streams API Example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/FilterExample.java) with a corresponding [tutorial](/learn/documentation/{{site.version}}/deployment/yarn.html#starting-your-application-on-yarn)
-- [Low-level API Example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/wikipedia/task/application/WikipediaParserTaskApplication.java) with a corresponding [tutorial](https://github.com/apache/samza-hello-samza#hello-samza)
+- [Low Level Task API Example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/wikipedia/task/application/WikipediaParserTaskApplication.java) with a corresponding [tutorial](https://github.com/apache/samza-hello-samza#hello-samza)
### Concepts
@@ -105,7 +105,7 @@ The above example configures Samza to ignore checkpointed offsets for `page-view
-### Code walkthrough: High-level API
+### Code walkthrough: High Level Streams API
In this section, we walk through a complete example that reads from a Kafka topic, filters a few messages and writes them to another topic.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/core-concepts/core-concepts.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/core-concepts/core-concepts.md b/docs/learn/documentation/versioned/core-concepts/core-concepts.md
index 479ebcb..b69de3d 100644
--- a/docs/learn/documentation/versioned/core-concepts/core-concepts.md
+++ b/docs/learn/documentation/versioned/core-concepts/core-concepts.md
@@ -55,7 +55,7 @@ A stream can have multiple producers that write data to it and multiple consumer
A stream is sharded into multiple partitions for scaling how its data is processed. Each _partition_ is an ordered, replayable sequence of records. When a message is written to a stream, it ends up in one its partitions. Each message in a partition is uniquely identified by an _offset_.
-Samza supports for pluggable systems that can implement the stream abstraction. As an example, Kafka implements a stream as a topic while another database might implement a stream as a sequence of updates to its tables.
+Samza supports pluggable systems that can implement the stream abstraction. As an example, Kafka implements a stream as a topic while a database might implement a stream as a sequence of updates to its tables.
## Stream Application
A _stream application_ processes messages from input streams, transforms them and emits results to an output stream or a database. It is built by chaining multiple operators, each of which take in one or more streams and transform them.
@@ -63,19 +63,19 @@ A _stream application_ processes messages from input streams, transforms them an
Samza offers three top-level APIs to help you build your stream applications: <br/>
-1. The [Samza Streams DSL](/learn/documentation/{{site.version}}/api/high-level-api.html), which offers several built-in operators like map, filter etc. This is the recommended API for most use-cases. <br/>
-2. The [low-level API](/learn/documentation/{{site.version}}/api/low-level-api.html), which allows greater flexibility to define your processing-logic and offers for greater control <br/>
+1. The [High Level Streams API](/learn/documentation/{{site.version}}/api/high-level-api.html), which offers several built-in operators like map, filter etc. This is the recommended API for most use-cases. <br/>
+2. The [Low Level Task API](/learn/documentation/{{site.version}}/api/low-level-api.html), which allows greater flexibility to define your processing-logic and offers greater control <br/>
3. [Samza SQL](/learn/documentation/{{site.version}}/api/samza-sql.html), which offers a declarative SQL interface to create your applications <br/>
## State
-Samza supports for both stateless and stateful stream processing. _Stateless processing_, as the name implies, does not retain any state associated with the current message after it has been processed. A good example of this is to filter an incoming stream of user-records by a field (eg:userId) and write the filtered messages to their own stream.
+Samza supports for both stateless and stateful stream processing. _Stateless processing_, as the name implies, does not retain any state associated with the current message after it has been processed. A good example of this is filtering an incoming stream of user-records by a field (eg:userId) and writing the filtered messages to their own stream.
-In contrast, _stateful processing_ requires you to record some state about a message even after processing it. Consider the example of counting the number of unique users to a website every five minutes. This requires you to record state about each user seen thus far, for deduping later. Samza offers a fault-tolerant, scalable state-store for this purpose.
+In contrast, _stateful processing_ requires you to record some state about a message even after processing it. Consider the example of counting the number of unique users to a website every five minutes. This requires you to store information about each user seen thus far for de-duplication. Samza offers a fault-tolerant, scalable state-store for this purpose.
## Time
-Time is a fundamental concept in stream processing, especially how it is modeled and interpreted by the system. Samza supports two notions of dealing with time. By default, all built-in Samza operators use processing time. In processing time, the timestamp of a message is determined by when it is processed by the system. For example, an event generated by a sensor could be processed by Samza several milliseconds later.
+Time is a fundamental concept in stream processing, especially in how it is modeled and interpreted by the system. Samza supports two notions of time. By default, all built-in Samza operators use processing time. In processing time, the timestamp of a message is determined by when it is processed by the system. For example, an event generated by a sensor could be processed by Samza several milliseconds later.
-On the other hand, in event time, the timestamp of an event is determined by when it actually occurred in the source. For example, a sensor which generates an event could embed the time of occurrence as a part of the event itself. Samza provides event-time based processing by its integration with [Apache BEAM](https://beam.apache.org/documentation/runners/samza/).
+On the other hand, in event time, the timestamp of an event is determined by when it actually occurred at the source. For example, a sensor which generates an event could embed the time of occurrence as a part of the event itself. Samza provides event-time based processing by its integration with [Apache BEAM](https://beam.apache.org/documentation/runners/samza/).
## Processing guarantee
Samza supports at-least once processing. As the name implies, this ensures that each message in the input stream is processed by the system at-least once. This guarantees no data-loss even when there are failures, thereby making Samza a practical choice for building fault-tolerant applications.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/hadoop/overview.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/hadoop/overview.md b/docs/learn/documentation/versioned/hadoop/overview.md
index 0820127..3d8caaa 100644
--- a/docs/learn/documentation/versioned/hadoop/overview.md
+++ b/docs/learn/documentation/versioned/hadoop/overview.md
@@ -29,7 +29,7 @@ Samza provides a single set of APIs for both batch and stream processing. This u
### Multi-stage Batch Pipeline
-Complex data pipelines usually consist multiple stages, with data shuffled (repartitioned) between stages to enable key-based operations such as windowing, aggregation, and join. Samza [high-level API](/startup/preview/index.html) provides an operator named `partitionBy` to create such multi-stage pipelines. Internally, Samza creates a physical stream, called an “intermediate stream”, based on the system configured as in `job.default.system`. Samza repartitions the output of the previous stage by sending it to the intermediate stream with the appropriate partition count and partition key. It then feeds it to the next stage of the pipeline. The lifecycle of intermediate streams is completely managed by Samza so from the user perspective the data shuffling is automatic.
+Complex data pipelines usually consist multiple stages, with data shuffled (repartitioned) between stages to enable key-based operations such as windowing, aggregation, and join. Samza [High Level Streams API](/startup/preview/index.html) provides an operator named `partitionBy` to create such multi-stage pipelines. Internally, Samza creates a physical stream, called an “intermediate stream”, based on the system configured as in `job.default.system`. Samza repartitions the output of the previous stage by sending it to the intermediate stream with the appropriate partition count and partition key. It then feeds it to the next stage of the pipeline. The lifecycle of intermediate streams is completely managed by Samza so from the user perspective the data shuffling is automatic.
For a single-stage pipeline, dealing with bounded data sets is straightforward: the system consumer “knows” the end of a particular partition, and it will emit end-of-stream token once a partition is complete. Samza will shut down the container when all its input partitions are complete.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/index.html
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/index.html b/docs/learn/documentation/versioned/index.html
index 893e428..2d5446b 100644
--- a/docs/learn/documentation/versioned/index.html
+++ b/docs/learn/documentation/versioned/index.html
@@ -26,19 +26,15 @@ title: Documentation
<h4>API</h4>
<ul class="documentation-list">
- <li><a href="api/low-level-api.html">Low-level API</a></li>
- <li><a href="api/high-level-api.html">Streams DSL</a></li>
+ <li><a href="api/programming-model.html">API overview</a></li>
+ <li><a href="api/high-level-api.html">High Level Streams API</a></li>
+ <li><a href="api/low-level-api.html">Low Level Task API</a></li>
<li><a href="api/table-api.html">Table API</a></li>
<li><a href="api/samza-sql.html">Samza SQL</a></li>
<li><a href="https://beam.apache.org/documentation/runners/samza/">Apache BEAM</a></li>
-<!-- TODO comparisons pages
- <li><a href="comparisons/aurora.html">Aurora</a></li>
- <li><a href="comparisons/jms.html">JMS</a></li>
- <li><a href="comparisons/s4.html">S4</a></li>
--->
</ul>
-<h4>Deployment</h4>
+<h4>DEPLOYMENT</h4>
<ul class="documentation-list">
<li><a href="deployment/deployment-model.html">Deployment options</a></li>
@@ -46,7 +42,7 @@ title: Documentation
<li><a href="deployment/standalone.html">Run as an embedded library</a></li>
</ul>
-<h4>Connectors</h4>
+<h4>CONNECTORS</h4>
<ul class="documentation-list">
<li><a href="connectors/overview.html">Connectors overview</a></li>
@@ -56,7 +52,7 @@ title: Documentation
<li><a href="connectors/kinesis.html">AWS Kinesis</a></li>
</ul>
-<h4>Operations</h4>
+<h4>OPERATIONS</h4>
<ul class="documentation-list">
<li><a href="operations/monitoring.html">Monitoring</a></li>
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/jobs/samza-configurations.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/jobs/samza-configurations.md b/docs/learn/documentation/versioned/jobs/samza-configurations.md
index 5828969..e55446a 100644
--- a/docs/learn/documentation/versioned/jobs/samza-configurations.md
+++ b/docs/learn/documentation/versioned/jobs/samza-configurations.md
@@ -249,7 +249,7 @@ These properties define Samza's storage mechanism for efficient [stateful stream
|Name|Default|Description|
|--- |--- |--- |
-|stores.**_store-name_**.factory| |You can give a store any **_store-name_** except `default` (`default` is reserved for defining default store parameters), and use that name to get a reference to the store in your Samza application (call [TaskContext.getStore()](../api/javadocs/org/apache/samza/task/TaskContext.html#getStore(java.lang.String)) in your task's [`init()`](../api/javadocs/org/apache/samza/task/InitableTask.html#init) method for the low-level API and in your application function's [`init()`](..api/javadocs/org/apache/samza/operators/functions/InitableFunction.html#init) method for the high level API. The value of this property is the fully-qualified name of a Java class that implements [`StorageEngineFactory`](../api/javadocs/org/apache/samza/storage/StorageEngineFactory.html). Samza currently ships with two storage engine implementations: <br><br>`org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory` <br>An on-disk storage engine with a key-value interface,
implemented using [RocksDB](http://rocksdb.org/). It supports fast random-access reads and writes, as well as range queries on keys. RocksDB can be configured with various additional tuning parameters.<br><br> `org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory`<br> In memory implementation of a key-value store. Uses `util.concurrent.ConcurrentSkipListMap` to store the keys in order.|
+|stores.**_store-name_**.factory| |You can give a store any **_store-name_** except `default` (`default` is reserved for defining default store parameters), and use that name to get a reference to the store in your Samza application (call [TaskContext.getStore()](../api/javadocs/org/apache/samza/task/TaskContext.html#getStore(java.lang.String)) in your task's [`init()`](../api/javadocs/org/apache/samza/task/InitableTask.html#init) method for the Low Level Task API and in your application function's [`init()`](..api/javadocs/org/apache/samza/operators/functions/InitableFunction.html#init) method for the high level API. The value of this property is the fully-qualified name of a Java class that implements [`StorageEngineFactory`](../api/javadocs/org/apache/samza/storage/StorageEngineFactory.html). Samza currently ships with two storage engine implementations: <br><br>`org.apache.samza.storage.kv.RocksDbKeyValueStorageEngineFactory` <br>An on-disk storage engine with a key-value interf
ace, implemented using [RocksDB](http://rocksdb.org/). It supports fast random-access reads and writes, as well as range queries on keys. RocksDB can be configured with various additional tuning parameters.<br><br> `org.apache.samza.storage.kv.inmemory.InMemoryKeyValueStorageEngineFactory`<br> In memory implementation of a key-value store. Uses `util.concurrent.ConcurrentSkipListMap` to store the keys in order.|
|stores.**_store-name_**.key.serde| |If the storage engine expects keys in the store to be simple byte arrays, this [serde](../container/serialization.html) allows the stream task to access the store using another object type as key. The value of this property must be a serde-name that is registered with serializers.registry.*.class. If this property is not set, keys are passed unmodified to the storage engine (and the changelog stream, if appropriate).|
|stores.**_store-name_**.msg.serde| |If the storage engine expects values in the store to be simple byte arrays, this [serde](../container/serialization.html) allows the stream task to access the store using another object type as value. The value of this property must be a serde-name that is registered with serializers.registry.*.class. If this property is not set, values are passed unmodified to the storage engine (and the changelog stream, if appropriate).|
|stores.**_store-name_**.changelog| |Samza stores are local to a container. If the container fails, the contents of the store are lost. To prevent loss of data, you need to set this property to configure a changelog stream: Samza then ensures that writes to the store are replicated to this stream, and the store is restored from this stream after a failure. The value of this property is given in the form system-name.stream-name. The "system-name" part is optional. If it is omitted you must specify the system in job.changelog.system config. Any output stream can be used as changelog, but you must ensure that only one job ever writes to a given changelog stream (each instance of a job and each store needs its own changelog stream).|
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/operations/monitoring.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/operations/monitoring.md b/docs/learn/documentation/versioned/operations/monitoring.md
index d93e497..ec8430c 100644
--- a/docs/learn/documentation/versioned/operations/monitoring.md
+++ b/docs/learn/documentation/versioned/operations/monitoring.md
@@ -37,8 +37,8 @@ Like any other production software, it is critical to monitor the health of our
+ [A.3 Creating a Custom MetricsReporter](#customreporter)
* [B. Metric Types in Samza](#metrictypes)
* [C. Adding User-Defined Metrics](#userdefinedmetrics)
- + [Low-level API](#lowlevelapi)
- + [High-Level API](#highlevelapi)
+ + [Low Level Task API](#lowlevelapi)
+ + [High Level Streams API](#highlevelapi)
* [D. Key Internal Samza Metrics](#keyinternalsamzametrics)
+ [D.1 Vital Metrics](#vitalmetrics)
+ [D.2 Store Metrics](#storemetrics)
@@ -232,7 +232,7 @@ To add a new metric, you can simply use the _MetricsRegistry_ in the provided Ta
the init() method to register new metrics. The code snippets below show examples of registering and updating a user-defined
Counter metric. Timers and gauges can similarly be used from within your task class.
-### <a name="lowlevelapi"></a> Low-level API
+### <a name="lowlevelapi"></a> Low Level Task API
Simply have your task implement the InitableTask interface and access the MetricsRegistry from the TaskContext.
@@ -252,9 +252,9 @@ public class MyJavaStreamTask implements StreamTask, InitableTask {
}
```
-### <a name="highlevelapi"></a> High-Level API
+### <a name="highlevelapi"></a> High Level Streams API
-In the high-level API, you can define a ContextManager and access the MetricsRegistry from the TaskContext, using which you can add and update your metrics.
+In the High Level Streams API, you can define a ContextManager and access the MetricsRegistry from the TaskContext, using which you can add and update your metrics.
```
public class MyJavaStreamApp implements StreamApplication {
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/tutorials/versioned/hello-samza-high-level-code.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/hello-samza-high-level-code.md b/docs/learn/tutorials/versioned/hello-samza-high-level-code.md
index 1c06116..a881b51 100644
--- a/docs/learn/tutorials/versioned/hello-samza-high-level-code.md
+++ b/docs/learn/tutorials/versioned/hello-samza-high-level-code.md
@@ -35,7 +35,7 @@ cd hello-samza
git checkout latest
{% endhighlight %}
-This project already contains implementations of the wikipedia application using both the low-level task API and the high-level API. The low-level task implementations are in the `samza.examples.wikipedia.task` package. The high-level application implementation is in the `samza.examples.wikipedia.application` package.
+This project already contains implementations of the wikipedia application using both the Low Level Task API and the High Level Streams API. The Low Level Task API implementations are in the `samza.examples.wikipedia.task` package. The High Level Streams API implementation is in the `samza.examples.wikipedia.application` package.
This tutorial will provide step by step instructions to recreate the existing wikipedia application.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/tutorials/versioned/hello-samza-high-level-yarn.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/hello-samza-high-level-yarn.md b/docs/learn/tutorials/versioned/hello-samza-high-level-yarn.md
index cfe589a..5127fd6 100644
--- a/docs/learn/tutorials/versioned/hello-samza-high-level-yarn.md
+++ b/docs/learn/tutorials/versioned/hello-samza-high-level-yarn.md
@@ -18,9 +18,9 @@ title: Hello Samza High Level API - YARN Deployment
See the License for the specific language governing permissions and
limitations under the License.
-->
-The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the low level task API as well as the high level API.
+The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the Low Level Task API as well as the High Level Streams API.
-This tutorial demonstrates a simple wikipedia application created with the high level API. The [Hello Samza tutorial] (/startup/hello-samza/{{site.version}}/index.html) is the low-level analog to this tutorial. It demonstrates the same logic but is created with the task API. The tutorials are designed to be as similar as possible. The primary differences are that with the high level API we accomplish the equivalent of 3 separate low-level jobs with a single application, we skip the intermediate topics for simplicity, and we can visualize the execution plan after we start the application.
+This tutorial demonstrates a simple wikipedia application created with the High Level Streams API. The [Hello Samza tutorial] (/startup/hello-samza/{{site.version}}/index.html) is the Low Level Task API analog to this tutorial. It demonstrates the same logic but is created with the Low Level Task API. The tutorials are designed to be as similar as possible. The primary differences are that with the High Level Streams API we accomplish the equivalent of 3 separate Low Level Task API jobs with a single application, we skip the intermediate topics for simplicity, and we can visualize the execution plan after we start the application.
### Get the Code
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md b/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
index 53dbec3..b9fe533 100644
--- a/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
+++ b/docs/learn/tutorials/versioned/samza-event-hubs-standalone.md
@@ -18,7 +18,7 @@ title: Samza Event Hubs Connectors Example
See the License for the specific language governing permissions and
limitations under the License.
-->
-The [hello-samza](https://github.com/apache/samza-hello-samza) project has an example that uses the Samza high-level API to consume and produce from [Event Hubs](../../documentation/versioned/connectors/eventhubs.html) using the Zookeeper deployment model.
+The [hello-samza](https://github.com/apache/samza-hello-samza) project has an example that uses the Samza High Level Streams API to consume and produce from [Event Hubs](../../documentation/versioned/connectors/eventhubs.html) using the Zookeeper deployment model.
#### Get the Code
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/tutorials/versioned/samza-sql.md
----------------------------------------------------------------------
diff --git a/docs/learn/tutorials/versioned/samza-sql.md b/docs/learn/tutorials/versioned/samza-sql.md
index f64aa06..71cfcc6 100644
--- a/docs/learn/tutorials/versioned/samza-sql.md
+++ b/docs/learn/tutorials/versioned/samza-sql.md
@@ -68,7 +68,7 @@ Below are some of the sql queries that you can execute using the samza-sql-conso
# Running Samza SQL on YARN
-The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the low level task API, high level API as well as Samza SQL.
+The [hello-samza](https://github.com/apache/samza-hello-samza) project is an example project designed to help you run your first Samza application. It has examples of applications using the Low Level Task API, High Level Streams API as well as Samza SQL.
This tutorial demonstrates a simple Samza application that uses SQL to perform stream processing.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/startup/code-examples/versioned/index.md
----------------------------------------------------------------------
diff --git a/docs/startup/code-examples/versioned/index.md b/docs/startup/code-examples/versioned/index.md
index ba1cc3e..384419d 100644
--- a/docs/startup/code-examples/versioned/index.md
+++ b/docs/startup/code-examples/versioned/index.md
@@ -36,7 +36,7 @@ These include:
- The [Join example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/JoinExample.java]) demonstrates how you can join a Kafka stream of page-views with a stream of ad-clicks
-- The [Stream-Table Join example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java) demonstrates how the Samza Table API. It joins a Kafka stream with a remote dataset accessed through a REST service.
+- The [Stream-Table Join example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java) demonstrates how to use the Samza Table API. It joins a Kafka stream with a remote dataset accessed through a REST service.
- The [SessionWindow](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/SessionWindowExample.java) and [TumblingWindow](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/TumblingWindowExample.java) examples illustrate Samza's rich windowing and triggering capabilities.
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/samza-api/src/main/java/org/apache/samza/task/ClosableTask.java
----------------------------------------------------------------------
diff --git a/samza-api/src/main/java/org/apache/samza/task/ClosableTask.java b/samza-api/src/main/java/org/apache/samza/task/ClosableTask.java
index 36003b3..148773f 100644
--- a/samza-api/src/main/java/org/apache/samza/task/ClosableTask.java
+++ b/samza-api/src/main/java/org/apache/samza/task/ClosableTask.java
@@ -19,12 +19,19 @@
package org.apache.samza.task;
+import org.apache.samza.context.ApplicationContainerContext;
+import org.apache.samza.context.ApplicationTaskContext;
+
/**
* A ClosableTask augments {@link org.apache.samza.task.StreamTask}, allowing the method implementer to specify
* code that will be called when the StreamTask is being shut down by the framework, providing to emit final metrics,
* clean or close resources, etc. The close method is not guaranteed to be called in event of crash or hard kill
* of the process.
+ *
+ * Deprecated: It's recommended to manage the lifecycle of any runtime objects using
+ * {@link ApplicationContainerContext} and {@link ApplicationTaskContext} instead.
*/
+@Deprecated
public interface ClosableTask {
void close() throws Exception;
}
[2/2] samza git commit: Updated API documentation for high and low
level APIs.
Posted by ja...@apache.org.
Updated API documentation for high and low level APIs.
vjagadish1989 nickpan47 Please take a look.
Author: Prateek Maheshwari <pm...@apache.org>
Reviewers: Jagadish<ja...@apache.org>
Closes #802 from prateekm/api-docs
Project: http://git-wip-us.apache.org/repos/asf/samza/repo
Commit: http://git-wip-us.apache.org/repos/asf/samza/commit/4e590709
Tree: http://git-wip-us.apache.org/repos/asf/samza/tree/4e590709
Diff: http://git-wip-us.apache.org/repos/asf/samza/diff/4e590709
Branch: refs/heads/master
Commit: 4e590709061fa8442dc0b0366a1f9976d04db9a0
Parents: 1d6f693
Author: Prateek Maheshwari <pm...@apache.org>
Authored: Tue Nov 13 18:17:17 2018 -0800
Committer: Jagadish <jv...@linkedin.com>
Committed: Tue Nov 13 18:17:17 2018 -0800
----------------------------------------------------------------------
.../versioned/api/high-level-api.md | 411 ++++++++++---------
.../versioned/api/low-level-api.md | 351 ++++++++--------
.../versioned/api/programming-model.md | 118 +++---
.../documentation/versioned/api/samza-sql.md | 2 +-
.../documentation/versioned/api/table-api.md | 11 +-
.../versioned/connectors/eventhubs.md | 2 +-
.../documentation/versioned/connectors/kafka.md | 6 +-
.../versioned/core-concepts/core-concepts.md | 14 +-
.../documentation/versioned/hadoop/overview.md | 2 +-
docs/learn/documentation/versioned/index.html | 16 +-
.../versioned/jobs/samza-configurations.md | 2 +-
.../versioned/operations/monitoring.md | 10 +-
.../versioned/hello-samza-high-level-code.md | 2 +-
.../versioned/hello-samza-high-level-yarn.md | 4 +-
.../versioned/samza-event-hubs-standalone.md | 2 +-
docs/learn/tutorials/versioned/samza-sql.md | 2 +-
docs/startup/code-examples/versioned/index.md | 2 +-
.../org/apache/samza/task/ClosableTask.java | 7 +
18 files changed, 483 insertions(+), 481 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/api/high-level-api.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/api/high-level-api.md b/docs/learn/documentation/versioned/api/high-level-api.md
index 9c13f24..43fd727 100644
--- a/docs/learn/documentation/versioned/api/high-level-api.md
+++ b/docs/learn/documentation/versioned/api/high-level-api.md
@@ -1,6 +1,6 @@
---
layout: page
-title: High-level API
+title: High Level Streams API
---
<!--
Licensed to the Apache Software Foundation (ASF) under one or more
@@ -19,155 +19,131 @@ title: High-level API
limitations under the License.
-->
-# Introduction
+### Table Of Contents
+- [Introduction](#introduction)
+- [Code Examples](#code-examples)
+- [Key Concepts](#key-concepts)
+ - [StreamApplication](#streamapplication)
+ - [MessageStream](#messagestream)
+ - [Table](#table)
+- [Operators](#operators)
+ - [Map](#map)
+ - [FlatMap](#flatmap)
+ - [Filter](#filter)
+ - [PartitionBy](#partitionby)
+ - [Merge](#merge)
+ - [Broadcast](#broadcast)
+ - [SendTo (Stream)](#sendto-stream)
+ - [SendTo (Table)](#sendto-table)
+ - [Sink](#sink)
+ - [Join (Stream-Stream)](#join-stream-stream)
+ - [Join (Stream-Table)](#join-stream-table)
+ - [Window](#window)
+ - [Windowing Concepts](#windowing-concepts)
+ - [Window Types](#window-types)
+- [Operator IDs](#operator-ids)
+- [Data Serialization](#data-serialization)
+- [Application Serialization](#application-serialization)
-The high level API provides the libraries to define your application logic. The StreamApplication is the central abstraction which your application must implement. You start by declaring your inputs as instances of MessageStream. Then you can apply operators on each MessageStream like map, filter, window, and join to define the whole end-to-end data processing in a single program.
+### Introduction
-Since the 0.13.0 release, Samza provides a new high level API that simplifies your applications. This API supports operations like re-partitioning, windowing, and joining on streams. You can now express your application logic concisely in few lines of code and accomplish what previously required multiple jobs.
-# Code Examples
+Samza's flexible High Level Streams API lets you describe your complex stream processing pipeline in the form of a Directional Acyclic Graph (DAG) of operations on [MessageStream](javadocs/org/apache/samza/operators/MessageStream). It provides a rich set of built-in operators that simplify common stream processing operations such as filtering, projection, repartitioning, stream-stream and stream-table joins, and windowing.
-Check out some examples to see the high-level API in action.
-1. PageView AdClick Joiner demonstrates joining a stream of PageViews with a stream of AdClicks, e.g. to analyze which pages get the most ad clicks.
-2. PageView Repartitioner illustrates re-partitioning the incoming stream of PageViews.
-3. PageView Sessionizer groups the incoming stream of events into sessions based on user activity.
-4. PageView by Region counts the number of views per-region over tumbling time intervals.
+### Code Examples
-# Key Concepts
-## StreamApplication
-When writing your stream processing application using the Samza high-level API, you implement a StreamApplication and define your processing logic in the describe method.
+[The Samza Cookbook](https://github.com/apache/samza-hello-samza/tree/master/src/main/java/samza/examples/cookbook) contains various recipes using the Samza High Level Streams API. These include:
-{% highlight java %}
-
- public void describe(StreamApplicationDescriptor appDesc) { … }
-
-{% endhighlight %}
-
-For example, here is a StreamApplication that validates and decorates page views with viewer’s profile information.
-
-{% highlight java %}
-
- public class BadPageViewFilter implements StreamApplication {
- @Override
- public void describe(StreamApplicationDescriptor appDesc) {
- KafkaSystemDescriptor kafka = new KafkaSystemDescriptor();
- InputDescriptor<PageView> pageViewInput = kafka.getInputDescriptor(“page-views”, new JsonSerdeV2<>(PageView.class));
- OutputDescriptor<DecoratedPageView> outputPageViews = kafka.getOutputDescriptor( “decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class));
- MessageStream<PageView> pageViews = appDesc.getInputStream(pageViewInput);
- pageViews.filter(this::isValidPageView)
- .map(this::addProfileInformation)
- .sendTo(appDesc.getOutputStream(outputPageViews));
- }
- }
-
-{% endhighlight %}
+- The [Filter example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/FilterExample.java) demonstrates how to perform stateless operations on a stream.
-## MessageStream
-A MessageStream, as the name implies, represents a stream of messages. A StreamApplication is described as a series of transformations on MessageStreams. You can get a MessageStream in two ways:
-1. Using StreamApplicationDescriptor.getInputStream to get the MessageStream for a given input stream (e.g., a Kafka topic).
-2. By transforming an existing MessageStream using operations like map, filter, window, join etc.
-## Table
-A Table represents a dataset that can be accessed by keys, and is one of the building blocks of the Samza high level API; the main motivation behind it is to support stream-table joins. The current K/V store is leveraged to provide backing store for local tables. More variations such as direct access and composite tables will be supported in the future. The usage of a table typically follows three steps:
-1. Create a table
-2. Populate the table using the sendTo() operator
-3. Join a stream with the table using the join() operator
+- The [Join example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/JoinExample.java]) demonstrates how you can join a Kafka stream of page-views with a stream of ad-clicks
-{% highlight java %}
+- The [Stream-Table Join example](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/RemoteTableJoinExample.java) demonstrates how to use the Samza Table API. It joins a Kafka stream with a remote dataset accessed through a REST service.
- final StreamApplication app = (streamAppDesc) -> {
- Table<KV<Integer, Profile>> table = streamAppDesc.getTable(new InMemoryTableDescriptor("t1")
- .withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde())));
- ...
- };
+- The [SessionWindow](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/SessionWindowExample.java) and [TumblingWindow](https://github.com/apache/samza-hello-samza/blob/latest/src/main/java/samza/examples/cookbook/TumblingWindowExample.java) examples illustrate Samza's rich windowing and triggering capabilities.
-{% endhighlight %}
-
-Example above creates a TableDescriptor object, which contains all information about a table. The currently supported table types are InMemoryTableDescriptor and RocksDbTableDescriptor. Notice the type of records in a table is KV, and Serdes for both key and value of records needs to be defined (line 4). Additional parameters can be added based on individual table types.
+### Key Concepts
+#### StreamApplication
+A [StreamApplication](javadocs/org/apache/samza/application/StreamApplication) describes the inputs, outputs, state, configuration and the processing logic for an application written using Samza's High Level Streams API.
-More details about step 2 and 3 can be found at [operator section](#operators).
+A typical StreamApplication implementation consists of the following stages:
-# Anatomy of a typical StreamApplication
-There are 3 simple steps to write your stream processing logic using the Samza high-level API.
-## Step 1: Obtain the input streams
-You can obtain the MessageStream for your input stream ID (“page-views”) using StreamApplicationDescriptor.getInputStream.
+ 1. Configuring the inputs, outputs and state (tables) using the appropriate [SystemDescriptor](javadocs/org/apache/samza/system/descriptors/SystemDescriptor)s, [InputDescriptor](javadocs/org/apache/samza/descriptors/InputDescriptor)s, [OutputDescriptor](javadocs/org/apache/samza/system/descriptors/OutputDescriptor)s and [TableDescriptor](javadocs/org/apache/samza/table/descriptors/TableDescriptor)s.
+ 2. Obtaining the corresponding [MessageStream](javadocs/org/apache/samza/operators/MessageStream)s, [OutputStream](javadocs/org/apache/samza/operators/OutputStream)s and [Table](javadocs/org/apache/samza/table/Table)s from the provided [StreamApplicationDescriptor](javadocs/org/apache/samza/application/descriptors/StreamApplicationDescriptor)
+ 3. Defining the processing logic using operators and functions on the streams and tables thus obtained.
+The following example StreamApplication removes page views older than 1 hour from the input stream:
+
{% highlight java %}
-
- KafkaSystemDescriptor sd = new KafkaSystemDescriptor("kafka")
- .withConsumerZkConnect(ImmutableList.of("localhost:2181"))
- .withProducerBootstrapServers(ImmutableList.of("localhost:9092"));
-
- KafkaInputDescriptor<KV<String, Integer>> pageViewInput =
- sd.getInputDescriptor("page-views", KVSerde.of(new StringSerde(), new JsonSerdeV2(PageView.class)));
-
- MessageStream<PageView> pageViews = streamAppDesc.getInputStream(pageViewInput);
-
+
+ public class PageViewFilter implements StreamApplication {
+ public void describe(StreamApplicationDescriptor appDescriptor) {
+ // Step 1: configure the inputs and outputs using descriptors
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("kafka")
+ .withConsumerZkConnect(ImmutableList.of("..."))
+ .withProducerBootstrapServers(ImmutableList.of("...", "..."));
+ KafkaInputDescriptor<PageViewEvent> kid =
+ ksd.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaOutputDescriptor<PageViewEvent>> kod =
+ ksd.getOutputDescriptor("recentPageViewEvent", new JsonSerdeV2<>(PageViewEvent.class)));
+
+ // Step 2: obtain the message strems and output streams
+ MessageStream<PageViewEvent> pageViewEvents = appDescriptor.getInputStream(kid);
+ OutputStream<PageViewEvent> recentPageViewEvents = appDescriptor.getOutputStream(kod);
+
+ // Step 3: define the processing logic
+ pageViewEvents
+ .filter(m -> m.getCreationTime() >
+ System.currentTimeMillis() - Duration.ofHours(1).toMillis())
+ .sendTo(recentPageViewEvents);
+ }
+ }
+
{% endhighlight %}
-The parameter {% highlight java %}pageViewInput{% endhighlight %} is the [InputDescriptor](javadocs/org/apache/samza/system/descriptors/InputDescriptor.html). Each InputDescriptor includes the full information of an input stream, including the stream ID, the serde to deserialize the input messages, and the system. By default, Samza uses the stream ID as the physical stream name and accesses the stream on the default system which is specified with the property “job.default.system”. However, the physical name and system properties can be overridden in configuration. For example, the following configuration defines the stream ID “page-views” as an alias for the PageViewEvent topic in a local Kafka cluster.
-{% highlight jproperties %}
+#### MessageStream
+A [MessageStream](javadocs/org/apache/samza/operators/MessageStream), as the name implies, represents a stream of messages. A StreamApplication is described as a Directed Acyclic Graph (DAG) of transformations on MessageStreams. You can get a MessageStream in two ways:
- streams.page-views.samza.physical.name=PageViewEvent
+1. Calling StreamApplicationDescriptor#getInputStream() with an [InputDescriptor](javadocs/org/apache/samza/system/descriptors/InputDescriptor) obtained from a [SystemDescriptor](javadocs/org/apache/samza/system/descriptors/SystemDescriptor).
+2. By transforming an existing MessageStream using operators like map, filter, window, join etc.
-{% endhighlight %}
+#### Table
+A [Table](javadocs/org/apache/samza/table/Table) is an abstraction for data sources that support random access by key. It is an evolution of the older [KeyValueStore](javadocs/org/apache/samza/storage/kv/KeyValueStore) API. It offers support for both local and remote data sources and composition through hybrid tables. For remote data sources, a [RemoteTable] provides optimized access with caching, rate-limiting, and retry support. Depending on the implementation, a Table can be a [ReadableTable](javadocs/org/apache/samza/table/ReadableTable) or a [ReadWriteTable](javadocs/org/apache/samza/table/ReadWriteTable).
+
+In the High Level Streams API, you can obtain and use a Table as follows:
-## Step 2: Define your transformation logic
-You are now ready to define your StreamApplication logic as a series of transformations on MessageStreams.
+1. Use the appropriate TableDescriptor to specify the table properties.
+2. Register the TableDescriptor with the StreamApplicationDescriptor. This returns a Table reference, which can be used for populate the table using the [Send To Table](#sendto-table) operator, or for joining a stream with the table using the [Stream-Table Join](#join-stream-table) operator.
+3. Alternatively, you can obtain a Table reference within an operator's [InitableFunction](javadocs/org/apache/samza/operators/functions/InitableFunction) using the provided [TaskContext](javadocs/org/apache/samza/context/TaskContext).
-{% highlight java %}
-
- MessageStream<DecoratedPageViews> decoratedPageViews = pageViews.filter(this::isValidPageView)
- .map(this::addProfileInformation);
-
-{% endhighlight %}
+### Operators
+The High Level Streams API provides common operations like map, flatmap, filter, merge, broadcast, joins, and windows on MessageStreams. Most of these operators accept their corresponding Functions as an argument.
-## Step 3: Write to output streams
-
-Finally, you can create an OutputStream using StreamApplicationDescriptor.getOutputStream and send the transformed messages through it.
+#### Map
+Applies the provided 1:1 [MapFunction](javadocs/org/apache/samza/operators/functions/MapFunction) to each element in the MessageStream and returns the transformed MessageStream. The MapFunction takes in a single message and returns a single message (potentially of a different type).
{% highlight java %}
- KafkaOutputDescriptor<DecoratedPageViews> outputPageViews =
- sd.getInputDescriptor("page-views", new JsonSerdeV2(DecoratedPageViews.class));
-
- // Send messages with userId as the key to “decorated-page-views”.
- decoratedPageViews.sendTo(streamAppDesc.getOutputStream(outputPageViews));
-
-{% endhighlight %}
-
-The parameter {% highlight java %}outputPageViews{% endhighlight %} is the [OutputDescriptor](javadocs/org/apache/samza/system/descriptors/OutputDescriptor.html), which includes the stream ID, the serde to serialize the outgoing messages, the physical name and the system. Similarly, the properties for this stream can be overridden just like the stream IDs for input streams. For example:
-
-{% highlight jproperties %}
-
- streams.decorated-page-views.samza.physical.name=DecoratedPageViewEvent
-
-{% endhighlight %}
-
-# Operators
-The high level API supports common operators like map, flatmap, filter, merge, joins, and windowing on streams. Most of these operators accept corresponding Functions, which are Initable and Closable.
-## Map
-Applies the provided 1:1 MapFunction to each element in the MessageStream and returns the transformed MessageStream. The MapFunction takes in a single message and returns a single message (potentially of a different type).
-
-{% highlight java %}
-
MessageStream<Integer> numbers = ...
MessageStream<Integer> tripled = numbers.map(m -> m * 3);
MessageStream<String> stringified = numbers.map(m -> String.valueOf(m));
{% endhighlight %}
-## FlatMap
-Applies the provided 1:n FlatMapFunction to each element in the MessageStream and returns the transformed MessageStream. The FlatMapFunction takes in a single message and returns zero or more messages.
+
+#### FlatMap
+Applies the provided 1:n [FlatMapFunction](javadocs/org/apache/samza/operators/functions/FlatMapFunction) to each element in the MessageStream and returns the transformed MessageStream. The FlatMapFunction takes in a single message and returns zero or more messages.
{% highlight java %}
MessageStream<String> sentence = ...
- // Parse the sentence into its individual words splitting by space
+ // Parse the sentence into its individual words splitting on space
MessageStream<String> words = sentence.flatMap(sentence ->
Arrays.asList(sentence.split(“ ”))
{% endhighlight %}
-## Filter
-Applies the provided FilterFunction to the MessageStream and returns the filtered MessageStream. The FilterFunction is a predicate that specifies whether a message should be retained in the filtered stream. Messages for which the FilterFunction returns false are filtered out.
+
+#### Filter
+Applies the provided [FilterFunction](javadocs/org/apache/samza/operators/functions/FilterFunction) to the MessageStream and returns the filtered MessageStream. The FilterFunction is a predicate that specifies whether a message should be retained in the filtered stream. Messages for which the FilterFunction returns false are filtered out.
{% highlight java %}
@@ -178,91 +154,109 @@ Applies the provided FilterFunction to the MessageStream and returns the filtere
MessageStream<String> shortWords = words.filter(word -> word.size() < 3);
{% endhighlight %}
-## PartitionBy
+
+#### PartitionBy
Re-partitions this MessageStream using the key returned by the provided keyExtractor and returns the transformed MessageStream. Messages are sent through an intermediate stream during repartitioning.
+
{% highlight java %}
MessageStream<PageView> pageViews = ...
- // Repartition pageView by userId.
- MessageStream<KV<String, PageView>> partitionedPageViews = pageViews.partitionBy(
- pageView -> pageView.getUserId(), // key extractor
- pageView -> pageView, // value extractor
- KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageView.class)), // serdes
- "partitioned-page-views"); // operator ID
+
+ // Repartition PageViews by userId.
+ MessageStream<KV<String, PageView>> partitionedPageViews =
+ pageViews.partitionBy(
+ pageView -> pageView.getUserId(), // key extractor
+ pageView -> pageView, // value extractor
+ KVSerde.of(new StringSerde(), new JsonSerdeV2<>(PageView.class)), // serdes
+ "partitioned-page-views"); // operator ID
{% endhighlight %}
-The operator ID should be unique for an operator within the application and is used to identify the streams and stores created by the operator.
+The operator ID should be unique for each operator within the application and is used to identify the streams and stores created by the operator.
-## Merge
+#### Merge
Merges the MessageStream with all the provided MessageStreams and returns the merged stream.
{% highlight java %}
- MessageStream<ServiceCall> serviceCall1 = ...
- MessageStream<ServiceCall> serviceCall2 = ...
- // Merge individual “ServiceCall” streams and create a new merged MessageStream
- MessageStream<ServiceCall> serviceCallMerged = serviceCall1.merge(serviceCall2);
+ MessageStream<LogEvent> log1 = ...
+ MessageStream<LogEvent> log2 = ...
+
+ // Merge individual “LogEvent” streams and create a new merged MessageStream
+ MessageStream<LogEvent> mergedLogs = log1.merge(log2);
+
+ // Alternatively, use mergeAll to merge multiple streams
+ MessageStream<LogEvent> mergedLogs = MessageStream.mergeAll(ImmutableList.of(log1, log2, ...));
{% endhighlight %}
-The merge transform preserves the order of each MessageStream, so if message {% highlight java %}m1{% endhighlight %} appears before {% highlight java %}m2{% endhighlight %} in any provided stream, then, {% highlight java %}m1{% endhighlight %} also appears before {% highlight java %}m2{% endhighlight %} in the merged stream.
-As an alternative to the merge instance method, you also can use the [MessageStream#mergeAll](javadocs/org/apache/samza/operators/MessageStream.html#mergeAll-java.util.Collection-) static method to merge MessageStreams without operating on an initial stream.
+The merge transform preserves the order of messages within each MessageStream. If message <code>m1</code> appears before <code>m2</code> in any provided stream, then, <code>m1</code> will also appears before <code>m2</code> in the merged stream.
+
+#### Broadcast
+Broadcasts the contents of the MessageStream to every *instance* of downstream operators via an intermediate stream.
-## Broadcast
-Broadcasts the MessageStream to all instances of down-stream transformation operators via the intermediate stream.
{% highlight java %}
- MessageStream<VersionChange> verChanges = ...
- // Broadcast input data version change event to all operator instances.
- MessageStream<VersionChange> broadcastVersionChanges =
- verChanges.broadcast(new JsonSerdeV2<>(VersionChange.class), // serde
- "broadcast-version-changes"); // operator ID
+ MessageStream<VersionChange> versionChanges = ...
+
+ // Broadcast version change event to all downstream operator instances.
+ versionChanges
+ .broadcast(
+ new JsonSerdeV2<>(VersionChange.class), // serde
+ "version-change-broadcast"); // operator ID
+ .map(vce -> /* act on version change event in each instance */ );
+
{% endhighlight %}
-## SendTo(Stream)
-Sends all messages from this MessageStream to the provided OutputStream. You can specify the key and the value to be used for the outgoing message.
+#### SendTo (Stream)
+Sends all messages in this MessageStream to the provided OutputStream. You can specify the key and the value to be used for the outgoing messages.
{% highlight java %}
- // Output a new message with userId as the key and region as the value to the “user-region” stream.
- OutputDescriptor<KV<String, String>> outputRegions =
- kafka.getOutputDescriptor(“user-region”, KVSerde.of(new StringSerde(), new StringSerde());
+ // Obtain the OutputStream using an OutputDescriptor
+ KafkaOutputDescriptor<KV<String, String>> kod =
+ ksd.getOutputDescriptor(“user-country”, KVSerde.of(new StringSerde(), new StringSerde());
+ OutputStream<KV<String, String>> userCountries = appDescriptor.getOutputStream(od)
+
MessageStream<PageView> pageViews = ...
- MessageStream<KV<String, PageView>> keyedPageViews = pageViews.map(KV.of(pageView.getUserId(), pageView.getRegion()));
- keyedPageViews.sendTo(appDesc.getOutputStream(outputRegions));
+ // Send a new message with userId as the key and their country as the value to the “user-country” stream.
+ pageViews
+ .map(pageView -> KV.of(pageView.getUserId(), pageView.getCountry()));
+ .sendTo(userCountries);
{% endhighlight %}
-## SendTo(Table)
-Sends all messages from this MessageStream to the provided table, the expected message type is KV.
+
+#### SendTo (Table)
+Sends all messages in this MessageStream to the provided Table. The expected message type is [KV](javadocs/org/apache/samza/operators/KV).
{% highlight java %}
+
+ MessageStream<Profile> profilesStream = ...
+ Table<KV<Long, Profile>> profilesTable =
- // Write a new message with memberId as the key and profile as the value to a table.
- appDesc.getInputStream(kafka.getInputDescriptor("Profile", new NoOpSerde<Profile>()))
- .map(m -> KV.of(m.getMemberId(), m))
- .sendTo(table);
+ profilesStream
+ .map(profile -> KV.of(profile.getMemberId(), profile))
+ .sendTo(profilesTable);
{% endhighlight %}
-## Sink
+#### Sink
Allows sending messages from this MessageStream to an output system using the provided [SinkFunction](javadocs/org/apache/samza/operators/functions/SinkFunction.html).
-This offers more control than {% highlight java %}sendTo{% endhighlight %} since the SinkFunction has access to the MessageCollector and the TaskCoordinator. For instance, you can choose to manually commit offsets, or shut-down the job using the TaskCoordinator APIs. This operator can also be used to send messages to non-Samza systems (e.g. remote databases, REST services, etc.)
+This offers more control than [SendTo (Stream)](#sendto-stream) since the SinkFunction has access to the MessageCollector and the TaskCoordinator. For example, you can choose to manually commit offsets, or shut-down the job using the TaskCoordinator APIs. This operator can also be used to send messages to non-Samza systems (e.g. a remote databases, REST services, etc.)
{% highlight java %}
- // Repartition pageView by userId.
MessageStream<PageView> pageViews = ...
- pageViews.sink( (msg, collector, coordinator) -> {
+
+ pageViews.sink((msg, collector, coordinator) -> {
// Construct a new outgoing message, and send it to a kafka topic named TransformedPageViewEvent.
- collector.send(new OutgoingMessageEnvelope(new SystemStream(“kafka”,
- “TransformedPageViewEvent”), msg));
+ collector.send(new OutgoingMessageEnvelope(new SystemStream(“kafka”, “TransformedPageViewEvent”), msg));
} );
{% endhighlight %}
-## Join(Stream-Stream)
-The stream-stream Join operator joins messages from two MessageStreams using the provided pairwise [JoinFunction](javadocs/org/apache/samza/operators/functions/JoinFunction.html). Messages are joined when the keys extracted from messages from the first stream match keys extracted from messages in the second stream. Messages in each stream are retained for the provided ttl duration and join results are emitted as matches are found.
+#### Join (Stream-Stream)
+The Stream-Stream Join operator joins messages from two MessageStreams using the provided pairwise [JoinFunction](javadocs/org/apache/samza/operators/functions/JoinFunction.html). Messages are joined when the key extracted from a message from the first stream matches the key extracted from a message in the second stream. Messages in each stream are retained for the provided ttl duration and join results are emitted as matches are found. Join only retains the latest message for each input stream.
{% highlight java %}
@@ -271,7 +265,9 @@ The stream-stream Join operator joins messages from two MessageStreams using the
MessageStream<OrderRecord> orders = …
MessageStream<ShipmentRecord> shipments = …
- MessageStream<FulfilledOrderRecord> shippedOrders = orders.join(shipments, new OrderShipmentJoiner(),
+ MessageStream<FulfilledOrderRecord> shippedOrders = orders.join(
+ shipments, // other stream
+ new OrderShipmentJoiner(), // join function
new StringSerde(), // serde for the join key
new JsonSerdeV2<>(OrderRecord.class), new JsonSerdeV2<>(ShipmentRecord.class), // serde for both streams
Duration.ofMinutes(20), // join TTL
@@ -297,18 +293,18 @@ The stream-stream Join operator joins messages from two MessageStreams using the
{% endhighlight %}
-## Join(Stream-Table)
-The stream-table Join operator joins messages from a MessageStream using the provided [StreamTableJoinFunction](javadocs/org/apache/samza/operators/functions/StreamTableJoinFunction.html). Messages from the input stream are joined with record in table using key extracted from input messages. The join function is invoked with both the message and the record. If a record is not found in the table, a null value is provided; the join function can choose to return null (inner join) or an output message (left outer join). For join to function properly, it is important to ensure the input stream and table are partitioned using the same key as this impacts the physical placement of data.
+#### Join (Stream-Table)
+The Stream-Table Join operator joins messages from a MessageStream with messages in a Table using the provided [StreamTableJoinFunction](javadocs/org/apache/samza/operators/functions/StreamTableJoinFunction.html). Messages are joined when the key extracted from a message in the stream matches the key for a record in the table. The join function is invoked with both the message and the record. If a record is not found in the table, a null value is provided. The join function can choose to return null for an inner join, or an output message for a left outer join. For join correctness, it is important to ensure the input stream and table are partitioned using the same key (e.g., using the partitionBy operator) as this impacts the physical placement of data.
{% highlight java %}
-
- streamAppDesc.getInputStream(kafk.getInputDescriptor("PageView", new NoOpSerde<PageView>()))
- .partitionBy(PageView::getMemberId, v -> v, "p1")
- .join(table, new PageViewToProfileJoinFunction())
+
+ pageViews
+ .partitionBy(pv -> pv.getMemberId, pv -> pv, "page-views-by-memberid")
+ .join(profiles, new PageViewToProfileTableJoiner())
...
- public class PageViewToProfileJoinFunction implements StreamTableJoinFunction
- <Integer, KV<Integer, PageView>, KV<Integer, Profile>, EnrichedPageView> {
+ public class PageViewToProfileTableJoiner implements
+ StreamTableJoinFunction<Integer, KV<Integer, PageView>, KV<Integer, Profile>, EnrichedPageView> {
@Override
public EnrichedPageView apply(KV<Integer, PageView> m, KV<Integer, Profile> r) {
@@ -325,10 +321,11 @@ The stream-table Join operator joins messages from a MessageStream using the pro
return record.getKey();
}
}
+
{% endhighlight %}
-## Window
-### Windowing Concepts
+### Window
+#### Windowing Concepts
**Windows, Triggers, and WindowPanes**: The window operator groups incoming messages in the MessageStream into finite windows. Each emitted result contains one or more messages in the window and is called a WindowPane.
A window can have one or more associated triggers which determine when results from the window are emitted. Triggers can be either [early triggers](javadocs/org/apache/samza/operators/windows/Window.html#setEarlyTrigger-org.apache.samza.operators.triggers.Trigger-) that allow emitting results speculatively before all data for the window has arrived, or late triggers that allow handling late messages for the window.
@@ -341,8 +338,8 @@ A discarding window clears all state for the window at every emission. Each emis
An accumulating window retains window results from previous emissions. Each emission will contain all messages that arrived since the beginning of the window.
-### Window Types
-The Samza high-level API currently supports tumbling and session windows.
+#### Window Types
+The Samza High Level Streams API currently supports tumbling and session windows.
**Tumbling Window**: A tumbling window defines a series of contiguous, fixed size time intervals in the stream.
@@ -350,20 +347,26 @@ Examples:
{% highlight java %}
- // Group the pageView stream into 3 second tumbling windows keyed by the userId.
+ // Group the pageView stream into 30 second tumbling windows keyed by the userId.
MessageStream<PageView> pageViews = ...
- MessageStream<WindowPane<String, Collection<PageView>>> =
- pageViews.window(
- Windows.keyedTumblingWindow(pageView -> pageView.getUserId(), // key extractor
+ MessageStream<WindowPane<String, Collection<PageView>>> = pageViews.window(
+ Windows.keyedTumblingWindow(
+ pageView -> pageView.getUserId(), // key extractor
Duration.ofSeconds(30), // window duration
new StringSerde(), new JsonSerdeV2<>(PageView.class)));
- // Compute the maximum value over tumbling windows of 3 seconds.
+ // Compute the maximum value over tumbling windows of 30 seconds.
MessageStream<Integer> integers = …
Supplier<Integer> initialValue = () -> Integer.MIN_VALUE;
- FoldLeftFunction<Integer, Integer> aggregateFunction = (msg, oldValue) -> Math.max(msg, oldValue);
- MessageStream<WindowPane<Void, Integer>> windowedStream =
- integers.window(Windows.tumblingWindow(Duration.ofSeconds(30), initialValue, aggregateFunction, new IntegerSerde()));
+ FoldLeftFunction<Integer, Integer> aggregateFunction =
+ (msg, oldValue) -> Math.max(msg, oldValue);
+
+ MessageStream<WindowPane<Void, Integer>> windowedStream = integers.window(
+ Windows.tumblingWindow(
+ Duration.ofSeconds(30),
+ initialValue,
+ aggregateFunction,
+ new IntegerSerde()));
{% endhighlight %}
@@ -378,39 +381,53 @@ Examples:
Supplier<Integer> initialValue = () -> 0
FoldLeftFunction<PageView, Integer> countAggregator = (pageView, oldCount) -> oldCount + 1;
Duration sessionGap = Duration.ofMinutes(3);
- MessageStream<WindowPane<String, Integer> sessionCounts = pageViews.window(Windows.keyedSessionWindow(
- pageView -> pageView.getUserId(), sessionGap, initialValue, countAggregator,
+
+ MessageStream<WindowPane<String, Integer> sessionCounts = pageViews.window(
+ Windows.keyedSessionWindow(
+ pageView -> pageView.getUserId(),
+ sessionGap,
+ initialValue,
+ countAggregator,
new StringSerde(), new IntegerSerde()));
- // Compute the maximum value over tumbling windows of 3 seconds.
+ // Compute the maximum value over tumbling windows of 30 seconds.
MessageStream<Integer> integers = …
Supplier<Integer> initialValue = () -> Integer.MAX_INT
-
FoldLeftFunction<Integer, Integer> aggregateFunction = (msg, oldValue) -> Math.max(msg, oldValue)
- MessageStream<WindowPane<Void, Integer>> windowedStream =
- integers.window(Windows.tumblingWindow(Duration.ofSeconds(3), initialValue, aggregateFunction,
- new IntegerSerde()));
+
+ MessageStream<WindowPane<Void, Integer>> windowedStream = integers.window(
+ Windows.tumblingWindow(
+ Duration.ofSeconds(30),
+ initialValue,
+ aggregateFunction,
+ new IntegerSerde()));
{% endhighlight %}
-# Operator IDs
-Each operator in your application is associated with a globally unique identifier. By default, each operator is assigned an ID based on its usage in the application. Some operators that create and use external resources (e.g., intermediate streams for partitionBy and broadcast, stores and changelogs for joins and windows, etc.) require you to provide an explicit ID for them. It's highly recommended to provide meaningful IDs for such operators. These IDs help you control the underlying resources when you make changes to the application logic that change the position of the operator within the DAG, and
+### Operator IDs
+Each operator in the StreamApplication is associated with a globally unique identifier. By default, each operator is assigned an ID by the framework based on its position in the operator DAG for the application. Some operators that create and use external resources require you to provide an explicit ID for them. Examples of such operators are partitionBy and broadcast with their intermediate streams, and window and join with their local stores and changelogs. It's strongly recommended to provide meaningful IDs for such operators.
+
+These IDs help you manage the underlying resources when you make changes to the application logic that change the position of the operator within the DAG and:
+
1. You wish to retain the previous state for the operator, since the changes to the DAG don't affect the operator semantics. For example, you added a map operator before a partitionBy operator to log the incoming message. In this case, you can retain previous the operator ID.
+
2. You wish to discard the previous state for the operator, since the changes to the DAG change the operator semantics. For example, you added a filter operator before a partitionBy operator that discards some of the messages. In this case, you should change the operator ID. Note that by doing so you will lose any previously checkpointed messages that haven't been completely processed by the downstream operators yet.
An operator ID is of the format: **jobName-jobId-opCode-opId**
-- **jobName** is the name of your job, as specified using the configuration "job.name"
-- **jobId** is the name of your job, as specified using the configuration "job.id"
-- **opCode** is an identifier for the type of the operator, e.g. map/filter/join
+
+- **jobName** is the name of your application, as specified using the configuration "app.name"
+- **jobId** is the id of your application, as specified using the configuration "app.id"
+- **opCode** is a pre-defined identifier for the type of the operator, e.g. map/filter/join
- **opId** is either auto-generated by the framework based on the position of the operator within the DAG, or can be provided by you for operators that manage external resources.
-# Application Serialization
-Samza relies on Java Serialization to distribute your application logic to the processors. For this to work, all of your custom application logic needs to be Serializable. For example, all the Function interfaces implement Serializable, and your implementations need to be serializable as well. It's recommended to use the Context APIs to set up any non-serializable context that your Application needs at Runtime.
+### Data Serialization
+Producing data to and consuming data from streams and tables require serializing and de-serializing it. In addition, some stateful operators like joins and windows store data locally for durability across restarts. Such operations require you to provide a [Serde](javadocs/org/apache/samza/serializers/Serde) implementation when using them. This also helps Samza infer the type of the data in your application, thus allowing the operator transforms to be checked for type safety at compile time. Samza provides the following Serde implementations that you can use out of the box:
-# Data Serialization
-Producing and consuming from streams and tables require serializing and deserializing data. In addition, some operators like joins and windows store data in a local store for durability across restarts. Such operations require you to provide a Serde implementation when using them. This also helps Samza infer the type of the data in your application, thus allowing the operator transforms to be type safe. Samza provides the following Serde implementations that you can use out of the box:
+- Common Types: Serdes for common Java data types, such as ByteBuffer, Double, Long, Integer, Byte, String.
+- [SerializableSerde](javadocs/org/apache/samza/serializers/SerializableSerde): A Serde for Java classes that implement the java.io.Serializable interface.
+- [JsonSerdeV2](javadocs/org/apache/samza/serializers/JsonSerdeV2): a Jackson based type safe JSON Serde that allows serializing from and deserializing to a POJO.
+- [KVSerde](javadocs/org/apache/samza/serializers/KVSerde): A pair of Serdes, first for the keys, and the second for the values in the incoming/outgoing message, a table record, or a [KV](javadocs/org/apache/samza/operators/KV) object.
+- [NoOpSerde](javadocs/org/apache/samza/serializers/NoOpSerde): A marker serde that indicates that the framework should not attempt any serialization/deserialization of the data. This is useful in some cases where the SystemProducer or SystemConsumer handles serialization and deserialization of the data itself.
-- KVSerde: A pair of Serdes, first for the keys, and the second for the values in the incoming/outgoing message or a table record.
-- NoOpSerde: A serde implementation that indicates that the framework should not attempt any serialization/deserialization of the data. Useful in some cases when the SystemProducer/SystemConsumer handle serialization/deserialization themselves.
-- JsonSerdeV2: a type-specific Json serde that allows directly deserializing the Json bytes into to specific POJO type.
-- Serdes for primitive types: serdes for primitive types, such as ByteBuffer, Double, Long, Integer, Byte, String, etc.
+### Application Serialization
+Samza uses Java Serialization to distribute an application's processing logic to the processors. For this to work, all application logic, including any Function implementations passed to operators, needs to be serializable. If you need to use any non-serializable objects at runtime, you can use the [ApplicationContainerContext](javadocs/org/apache/samza/context/ApplicationContainerContext) and [ApplicationTaskContext](javadocs/org/apache/samza/context/ApplicationContainerContext) APIs to manage their lifecycle.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/samza/blob/4e590709/docs/learn/documentation/versioned/api/low-level-api.md
----------------------------------------------------------------------
diff --git a/docs/learn/documentation/versioned/api/low-level-api.md b/docs/learn/documentation/versioned/api/low-level-api.md
index e91f74e..34d2b06 100644
--- a/docs/learn/documentation/versioned/api/low-level-api.md
+++ b/docs/learn/documentation/versioned/api/low-level-api.md
@@ -19,291 +19,268 @@ title: Low level Task API
limitations under the License.
-->
+### Table Of Contents
+- [Introduction](#introduction)
+- [Code Examples](#code-examples)
+- [Key Concepts](#key-concepts)
+ - [TaskApplication](#taskapplication)
+ - [TaskFactory](#taskfactory)
+ - [Task Interfaces](#task-interfaces)
+ - [StreamTask](#streamtask)
+ - [AsyncStreamTask](#asyncstreamtask)
+ - [Additional Task Interfaces](#additional-task-interfaces)
+ - [InitableTask](#initabletask)
+ - [ClosableTask](#closabletask)
+ - [WindowableTask](#windowabletask)
+ - [EndOfStreamListenerTask](#endofstreamlistenertask)
+- [Common Operations](#common-operations)
+ - [Receiving Messages from Input Streams](#receiving-messages-from-input-streams)
+ - [Sending Messages to Output Streams](#sending-messages-to-output-streams)
+ - [Accessing Tables](#accessing-tables)
+- [Legacy Applications](#legacy-applications)
-# Introduction
-Task APIs (i.e. [StreamTask](javadocs/org/apache/samza/task/StreamTask.html) or [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask.html)) are bare-metal interfaces that exposes the system implementation details in Samza. When using Task APIs, you will implement your application as a [TaskApplication](javadocs/org/apache/samza/application/TaskApplication.html). The main difference between a TaskApplication and a StreamApplication is the APIs used to describe the processing logic. In TaskApplication, the processing logic is defined via StreamTask and AsyncStreamTask.
+### Introduction
+Samza's powerful Low Level Task API lets you write your application in terms of processing logic for each incoming message. When using the Low Level Task API, you implement a [TaskApplication](javadocs/org/apache/samza/application/TaskApplication). The processing logic is defined as either a [StreamTask](javadocs/org/apache/samza/task/StreamTask) or an [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask).
-# Key Concepts
-## TaskApplication
-Here is an example of a user implemented TaskApplication:
+### Code Examples
-{% highlight java %}
-
- package com.example.samza;
+The [Hello Samza](https://github.com/apache/samza-hello-samza/tree/master/src/main/java/samza/examples/wikipedia/task/application) Wikipedia applications demonstrate how to use Samza's Low Level Task API. These applications consume various events from Wikipedia, transform them, and calculates several statistics about them.
- public class BadPageViewFilter implements TaskApplication {
- @Override
- public void describe(TaskApplicationDescriptor appDesc) {
- // Add input, output streams and tables
- KafkaSystemDescriptor<String, PageViewEvent> kafkaSystem =
- new KafkaSystemDescriptor(“kafka”)
- .withConsumerZkConnect(myZkServers)
- .withProducerBootstrapServers(myBrokers);
- KVSerde<String, PageViewEvent> serde =
- KVSerde.of(new StringSerde(), new JsonSerdeV2<PageViewEvent>());
- // Add input, output streams and tables
- appDesc.withInputStream(kafkaSystem.getInputDescriptor(“pageViewEvent”, serde))
- .withOutputStream(kafkaSystem.getOutputDescriptor(“goodPageViewEvent”, serde))
- .withTable(new RocksDBTableDescriptor(
- “badPageUrlTable”, KVSerde.of(new StringSerde(), new IntegerSerde())
- .withTaskFactory(new BadPageViewTaskFactory());
- }
- }
+- The [WikipediaFeedTaskApplication](https://github.com/apache/samza-hello-samza/blob/master/src/main/java/samza/examples/wikipedia/task/application/WikipediaFeedTaskApplication.java) demonstrates how to consume multiple Wikipedia event streams and merge them to an Apache Kafka topic.
-{% endhighlight %}
+- The [WikipediaParserTaskApplication](https://github.com/apache/samza-hello-samza/blob/master/src/main/java/samza/examples/wikipedia/task/application/WikipediaParserTaskApplication.java) demonstrates how to project the incoming events from the Apache Kafka topic to a custom JSON data type.
-In the above example, user defines the input stream, the output stream, and a RocksDB table for the application, and then provide the processing logic defined in BadPageViewTaskFactory. All descriptors (i.e. input/output streams and tables) and the [TaskFactory](javadocs/org/apache/samza/task/TaskFactory.html) are registered to the [TaskApplicationDescriptor](javadocs/org/apache/samza/application/descriptors/TaskApplicationDescriptor.html).
+- The [WikipediaStatsTaskApplication](https://github.com/apache/samza-hello-samza/blob/master/src/main/java/samza/examples/wikipedia/task/application/WikipediaStatsTaskApplication.java) demonstrates how to calculate and emit periodic statistics about the incoming events while using a local KV store for durability.
-## TaskFactory
-You will need to implement a [TaskFactory](javadocs/org/apache/samza/task/TaskFactory.html) to create task instances to execute user defined processing logic. Correspondingly, StreamTaskFactory and AsyncStreamTaskFactory are used to create StreamTask and AsyncStreamTask respectively. The [StreamTaskFactory](javadocs/org/apache/samza/task/StreamTaskFactory.html) for the above example is shown below:
+### Key Concepts
-{% highlight java %}
+#### TaskApplication
+
+A [TaskApplication](javadocs/org/apache/samza/application/TaskApplication) describes the inputs, outputs, state, configuration and the processing logic for an application written using Samza's Low Level Task API.
+
+A typical TaskApplication implementation consists of the following stages:
- package com.example.samza;
+ 1. Configuring the inputs, outputs and state (tables) using the appropriate [SystemDescriptor](javadocs/org/apache/samza/system/descriptors/SystemDescriptor)s, [InputDescriptor](javadocs/org/apache/samza/descriptors/InputDescriptor)s, [OutputDescriptor](javadocs/org/apache/samza/system/descriptors/OutputDescriptor)s and [TableDescriptor](javadocs/org/apache/samza/table/descriptors/TableDescriptor)s.
+ 2. Adding the descriptors above to the provided [TaskApplicationDescriptor](javadocs/org/apache/samza/application/descriptors/TaskApplicationDescriptor)
+ 3. Defining the processing logic in a [StreamTask](javadocs/org/apache/samza/task/StreamTask) or an [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask) implementation, and adding its corresponding [StreamTaskFactory](javadocs/org/apache/samza/task/StreamTaskFactory) or [AsyncStreamTaskFactory](javadocs/org/apache/samza/task/AsyncStreamTaskFactory) to the TaskApplicationDescriptor.
- public class BadPageViewTaskFactory implements StreamTaskFactory {
+The following example TaskApplication removes page views with "bad URLs" from the input stream:
+
+{% highlight java %}
+
+ public class PageViewFilter implements TaskApplication {
@Override
- public StreamTask createInstance() {
- // Add input, output streams and tables
- return new BadPageViewFilterTask();
+ public void describe(TaskApplicationDescriptor appDescriptor) {
+ // Step 1: configure the inputs and outputs using descriptors
+ KafkaSystemDescriptor ksd = new KafkaSystemDescriptor("kafka")
+ .withConsumerZkConnect(ImmutableList.of("..."))
+ .withProducerBootstrapServers(ImmutableList.of("...", "..."));
+ KafkaInputDescriptor<PageViewEvent> kid =
+ ksd.getInputDescriptor("pageViewEvent", new JsonSerdeV2<>(PageViewEvent.class));
+ KafkaOutputDescriptor<PageViewEvent>> kod =
+ ksd.getOutputDescriptor("goodPageViewEvent", new JsonSerdeV2<>(PageViewEvent.class)));
+ RocksDbTableDescriptor badUrls =
+ new RocksDbTableDescriptor(“badUrls”, KVSerde.of(new StringSerde(), new IntegerSerde());
+
+ // Step 2: Add input, output streams and tables
+ appDescriptor
+ .withInputStream(kid)
+ .withOutputStream(kod)
+ .withTable(badUrls)
+
+ // Step 3: define the processing logic
+ appDescriptor.withTaskFactory(new PageViewFilterTaskFactory());
}
}
-
+
{% endhighlight %}
-Similarly, here is an example of [AsyncStreamTaskFactory](javadocs/org/apache/samza/task/AsyncStreamTaskFactory.html):
+#### TaskFactory
+Your [TaskFactory](javadocs/org/apache/samza/task/TaskFactory) will be used to create instances of your Task in each of Samza's processors. If you're implementing a StreamTask, you can provide a [StreamTaskFactory](javadocs/org/apache/samza/task/StreamTaskFactory). Similarly, if you're implementing an AsyncStreamTask, you can provide an [AsyncStreamTaskFactory](javadocs/org/apache/samza/task/AsyncStreamTaskFactory). For example:
{% highlight java %}
-
- package com.example.samza;
- public class BadPageViewAsyncTaskFactory implements AsyncStreamTaskFactory {
+ public class PageViewFilterTaskFactory implements StreamTaskFactory {
@Override
- public AsyncStreamTask createInstance() {
- // Add input, output streams and tables
- return new BadPageViewAsyncFilterTask();
+ public StreamTask createInstance() {
+ return new PageViewFilterTask();
}
}
+
{% endhighlight %}
-## Task classes
+#### Task Interfaces
-The actual processing logic is implemented in [StreamTask](javadocs/org/apache/samza/task/StreamTask.html) or [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask.html) classes.
+Your processing logic can be implemented in a [StreamTask](javadocs/org/apache/samza/task/StreamTask) or an [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask).
-### StreamTask
-You should implement [StreamTask](javadocs/org/apache/samza/task/StreamTask.html) for synchronous process, where the message processing is complete after the process method returns. An example of StreamTask is a computation that does not involve remote calls:
+##### StreamTask
+You can implement a [StreamTask](javadocs/org/apache/samza/task/StreamTask) for synchronous message processing. Samza delivers messages to the task one at a time, and considers each message to be processed when the process method call returns. For example:
{% highlight java %}
-
- package com.example.samza;
- public class BadPageViewFilterTask implements StreamTask {
+ public class PageViewFilterTask implements StreamTask {
@Override
- public void process(IncomingMessageEnvelope envelope,
- MessageCollector collector,
- TaskCoordinator coordinator) {
- // process message synchronously
+ public void process(
+ IncomingMessageEnvelope envelope,
+ MessageCollector collector,
+ TaskCoordinator coordinator) {
+
+ // process the message in the envelope synchronously
}
}
+
{% endhighlight %}
-### AsyncStreamTask
-The [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask.html) interface, on the other hand, supports asynchronous process, where the message processing may not be complete after the processAsync method returns. Various concurrent libraries like Java NIO, ParSeq and Akka can be used here to make asynchronous calls, and the completion is marked by invoking the [TaskCallback](javadocs/org/apache/samza/task/TaskCallback.html). Samza will continue to process next message or shut down the container based on the callback status. An example of AsyncStreamTask is a computation that make remote calls but don’t block on the call completion:
+Note that synchronous message processing does not imply sequential execution. Multiple instances of your Task class implementation may still run concurrently within a container.
+
+##### AsyncStreamTask
+You can implement a [AsyncStreamTask](javadocs/org/apache/samza/task/AsyncStreamTask) for asynchronous message processing. This can be useful when you need to perform long running I/O operations to process a message, e.g., making an http request. For example:
{% highlight java %}
-
- package com.example.samza;
- public class BadPageViewAsyncFilterTask implements AsyncStreamTask {
+ public class AsyncPageViewFilterTask implements AsyncStreamTask {
@Override
public void processAsync(IncomingMessageEnvelope envelope,
- MessageCollector collector,
- TaskCoordinator coordinator,
- TaskCallback callback) {
- // process message with asynchronous calls
- // fire callback upon completion, e.g. invoking callback from asynchronous call completion thread
+ MessageCollector collector,
+ TaskCoordinator coordinator,
+ TaskCallback callback) {
+
+ // process message asynchronously
+ // invoke callback.complete or callback.failure upon completion
}
}
+
{% endhighlight %}
-# Runtime Objects
+Samza delivers the incoming message and a [TaskCallback](javadocs/org/apache/samza/task/TaskCallback) with the processAsync() method call, and considers each message to be processed when its corresponding callback.complete() or callback.failure() has been invoked. If callback.failure() is invoked, or neither callback.complete() or callback.failure() is invoked within <code>task.callback.ms</code> milliseconds, Samza will shut down the running Container.
+
+If configured, Samza will keep up to <code>task.max.concurrency</code> number of messages processing asynchronously at a time within each Task Instance. Note that while message delivery (i.e., processAsync invocation) is guaranteed to be in-order within a stream partition, message processing may complete out of order when setting <code>task.max.concurrency</code> > 1.
+
+For more details on asynchronous and concurrent processing, see the [Samza Async API and Multithreading User Guide](../../../tutorials/{{site.version}}/samza-async-user-guide).
-## Task Instances in Runtime
-When you run your job, Samza will create many instances of your task class (potentially on multiple machines). These task instances process the messages from the input streams.
+##### Additional Task Interfaces
-## Messages from Input Streams
+There are a few other interfaces you can implement in your StreamTask or AsyncStreamTask that provide additional functionality.
-For each message that Samza receives from the task’s input streams, the [process](javadocs/org/apache/samza/task/StreamTask.html#process-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-) or [processAsync](javadocs/org/apache/samza/task/AsyncStreamTask.html#processAsync-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-org.apache.samza.task.TaskCallback-) method is called. The [envelope](javadocs/org/apache/samza/system/IncomingMessageEnvelope.html) contains three things of importance: the message, the key, and the stream that the message came from.
+###### InitableTask
+You can implement the [InitableTask](javadocs/org/apache/samza/task/InitableTask) interface to access the [Context](javadocs/org/apache/samza/context/Context). Context provides access to any runtime objects you need in your task,
+whether they're provided by the framework, or your own.
{% highlight java %}
- /** Every message that is delivered to a StreamTask is wrapped
- * in an IncomingMessageEnvelope, which contains metadata about
- * the origin of the message. */
- public class IncomingMessageEnvelope {
- /** A deserialized message. */
- Object getMessage() { ... }
-
- /** A deserialized key. */
- Object getKey() { ... }
-
- /** The stream and partition that this message came from. */
- SystemStreamPartition getSystemStreamPartition() { ... }
+ public interface InitableTask {
+ void init(Context context) throws Exception;
}
+
{% endhighlight %}
-The key and value are declared as Object, and need to be cast to the correct type. The serializer/deserializer are defined via InputDescriptor, as described [here](high-level-api.md#data-serialization). A deserializer can convert these bytes into any other type, for example the JSON deserializer mentioned above parses the byte array into java.util.Map, java.util.List and String objects.
-
-The [getSystemStreamPartition()](javadocs/org/apache/samza/system/IncomingMessageEnvelope.html#getSystemStreamPartition--) method returns a [SystemStreamPartition](javadocs/org/apache/samza/system/SystemStreamPartition.html) object, which tells you where the message came from. It consists of three parts:
-1. The *system*: the name of the system from which the message came, as defined as SystemDescriptor in your TaskApplication. You can have multiple systems for input and/or output, each with a different name.
-2. The *stream name*: the name of the stream (topic, queue) within the source system. This is also defined as InputDescriptor in the TaskApplication.
-3. The [*partition*](javadocs/org/apache/samza/Partition.html): a stream is normally split into several partitions, and each partition is assigned to one task instance by Samza.
-
-The API looks like this:
+###### ClosableTask
+You can implement the [ClosableTask](javadocs/org/apache/samza/task/ClosableTask) to clean up any runtime state during shutdown. This interface is deprecated. It's recommended to use the [ApplicationContainerContext](javadocs/org/apache/samza/context/ApplicationContainerContext) and [ApplicationTaskContext](javadocs/org/apache/samza/context/ApplicationContainerContext) APIs to manage the lifecycle of any runtime objects.
{% highlight java %}
-
- /** A triple of system name, stream name and partition. */
- public class SystemStreamPartition extends SystemStream {
- /** The name of the system which provides this stream. It is
- defined in the Samza job's configuration. */
- public String getSystem() { ... }
-
- /** The name of the stream/topic/queue within the system. */
- public String getStream() { ... }
-
- /** The partition within the stream. */
- public Partition getPartition() { ... }
+ public interface ClosableTask {
+ void close() throws Exception;
}
+
{% endhighlight %}
-In the example user-implemented TaskApplication above, the system name is “kafka”, the stream name is “pageViewEvent”. (The name “kafka” isn’t special — you can give your system any name you want.) If you have several input streams feeding into your StreamTask or AsyncStreamTask, you can use the SystemStreamPartition to determine what kind of message you’ve received.
-
-## Messages to Output Streams
-What about sending messages? If you take a look at the [process()](javadocs/org/apache/samza/task/StreamTask.html#process-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-) method in StreamTask, you’ll see that you get a [MessageCollector](javadocs/org/apache/samza/task/MessageCollector.html). Similarly, you will get it in [processAsync()](javadocs/org/apache/samza/task/AsyncStreamTask.html#processAsync-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-org.apache.samza.task.TaskCallback-) method in AsyncStreamTask as well.
+###### WindowableTask
+You can implement the [WindowableTask](javadocs/org/apache/samza/task/WindowableTask) interface to implement processing logic that is invoked periodically by the framework.
{% highlight java %}
- /** When a task wishes to send a message, it uses this interface. */
- public interface MessageCollector {
- void send(OutgoingMessageEnvelope envelope);
+ public interface WindowableTask {
+ void window(MessageCollector collector, TaskCoordinator coordinator) throws Exception;
}
-
-{% endhighlight %}
-
-To send a message, you create an [OutgoingMessageEnvelope](javadocs/org/apache/samza/system/OutgoingMessageEnvelope.html) object and pass it to the MessageCollector. At a minimum, the envelope specifies the message you want to send, and the system and stream name to send it to. Optionally you can specify the partitioning key and other parameters. See the [javadoc](javadocs/org/apache/samza/system/OutgoingMessageEnvelope.html) for details.
-**NOTE**: Please only use the MessageCollector object within the process() or processAsync() method. If you hold on to a MessageCollector instance and use it again later, your messages may not be sent correctly.
+{% endhighlight %}
-For example, here’s a simple example to send out “Good PageViewEvents” in the BadPageViewFilterTask:
+###### EndOfStreamListenerTask
+You can implement the [EndOfStreamListenerTask](javadocs/org/apache/samza/task/EndOfStreamListenerTask) interface to implement processing logic that is invoked when a Task Instance has reached the end of all input SystemStreamPartitions it's consuming. This is typically relevant when running Samza as a batch job.
{% highlight java %}
-
- public class BadPageViewFilterTask implements StreamTask {
- // Send outgoing messages to a stream called "words"
- // in the "kafka" system.
- private final SystemStream OUTPUT_STREAM =
- new SystemStream("kafka", "goodPageViewEvent");
- @Override
- public void process(IncomingMessageEnvelope envelope,
- MessageCollector collector,
- TaskCoordinator coordinator) {
- if (isBadPageView(envelope)) {
- // skip the message, increment the counter, do not send it
- return;
- }
- collector.send(new OutgoingMessageEnvelope(OUTPUT_STREAM, envelope.getKey(), envelope.getValue()));
- }
+ public interface EndOfStreamListenerTask {
+ void onEndOfStream(MessageCollector collector, TaskCoordinator coordinator) throws Exception;
}
{% endhighlight %}
-## Accessing Tables
-There are many cases that you will need to lookup a table when processing an incoming message. Samza allows access to tables by a unique name through [TaskContext.getTable()](javadocs/org/apache/samza/task/TaskContext.html#getTable-java.lang.String-) method. [TaskContext](javadocs/org/apache/samza/task/TaskContext.html) is accessed via [Context.getTaskContext()](javadocs/org/apache/samza/context/Context.html#getTaskContext--) in the [InitiableTask’s init()]((javadocs/org/apache/samza/task/InitableTask.html#init-org.apache.samza.context.Context-)) method. A user code example to access a table in the above TaskApplication example is here:
+### Common Operations
-{% highlight java %}
-
- public class BadPageViewFilter implements StreamTask, InitableTask {
- private final SystemStream OUTPUT_STREAM = new SystemStream(“kafka”, “goodPageViewEvent”);
- private ReadWriteTable<String, Integer> badPageUrlTable;
- @Override
- public void init(Context context) {
- badPageUrlTable = (ReadWriteTable<String, Integer>) context.getTaskContext().getTable("badPageUrlTable");
- }
+#### Receiving Messages from Input Streams
- @Override
- public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
- String key = (String)message.getKey();
- if (badPageUrlTable.containsKey(key)) {
- // skip the message, increment the counter, do not send it
- badPageUrlTable.put(key, badPageUrlTable.get(key) + 1);
- return;
- }
- collector.send(new OutgoingMessageEnvelope(OUTPUT_STREAM, key, message.getValue()));
- }
- }
+Samza calls your Task instance's [process](javadocs/org/apache/samza/task/StreamTask#process-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-) or [processAsync](javadocs/org/apache/samza/task/AsyncStreamTask#processAsync-org.apache.samza.system.IncomingMessageEnvelope-org.apache.samza.task.MessageCollector-org.apache.samza.task.TaskCoordinator-org.apache.samza.task.TaskCallback-) method with each incoming message on your input streams. The [IncomingMessageEnvelope](javadocs/org/apache/samza/system/IncomingMessageEnvelope) can be used to obtain the following information: the de-serialized key, the de-serialized message, and the [SystemStreamPartition](javadocs/org/apache/samza/system/SystemStreamPartition) that the message came from.
-{% endhighlight %}
+The key and message objects need to be cast to the correct type in your Task implementation based on the [Serde](javadocs/org/apache/samza/serializers/Serde.html) provided for the InputDescriptor for the input stream.
-For more detailed AsyncStreamTask example, follow the tutorial in [Samza Async API and Multithreading User Guide](../../../tutorials/{{site.version}}/samza-async-user-guide.html). For more details on APIs, please refer to [Configuration](../jobs/configuration.md) and [Javadocs](javadocs).
+The [SystemStreamPartition](javadocs/org/apache/samza/system/SystemStreamPartition) object tells you where the message came from. It consists of three parts:
+1. The *system*: the name of the system the message came from, as specified for the SystemDescriptor in your TaskApplication. You can have multiple systems for input and/or output, each with a different name.
+2. The *stream name*: the name of the stream (e.g., topic, queue) within the input system. This is the physical name of the stream, as specified for the InputDescriptor in your TaskApplication.
+3. The [*partition*](javadocs/org/apache/samza/Partition): A stream is normally split into several partitions, and each partition is assigned to one task instance by Samza.
-# Other Task Interfaces
+If you have several input streams for your TaskApplication, you can use the SystemStreamPartition to determine what kind of message you’ve received.
-There are other task interfaces to allow additional processing logic to be applied, besides the main per-message processing logic defined in StreamTask and AsyncStreamTask. You will need to implement those task interfaces in addition to StreamTask or AsyncStreamTask.
+#### Sending Messages to Output Streams
+To send a message to a stream, you first create an [OutgoingMessageEnvelope](javadocs/org/apache/samza/system/OutgoingMessageEnvelope). At a minimum, you need to provide the message you want to send, and the system and stream to send it to. Optionally you can specify the partitioning key and other parameters. See the [javadoc](javadocs/org/apache/samza/system/OutgoingMessageEnvelope) for details.
-## InitiableTask
-This task interface allows users to initialize objects that are accessed within a task instance.
+You can then send the OutgoingMessageEnvelope using the [MessageCollector](javadocs/org/apache/samza/task/MessageCollector) provided with the process() or processAsync() call. You **must** use the MessageCollector delivered for the message you're currently processing. Holding on to a MessageCollector and reusing it later will cause your messages to not be sent correctly.
{% highlight java %}
- public interface InitableTask {
- void init(Context context) throws Exception;
+ /** When a task wishes to send a message, it uses this interface. */
+ public interface MessageCollector {
+ void send(OutgoingMessageEnvelope envelope);
}
-{% endhighlight %}
-
-## WindowableTask
-This task interface allows users to define a processing logic that is invoked periodically within a task instance.
-
-{% highlight java %}
- public interface WindowableTask {
- void window(MessageCollector collector, TaskCoordinator coordinator) throws Exception;
- }
-
{% endhighlight %}
-## ClosableTask
-This task interface defines the additional logic when closing a task. Usually, it is in pair with InitableTask to release system resources allocated for this task instance.
+#### Accessing Tables
-{% highlight java %}
+A [Table](javadocs/org/apache/samza/table/Table) is an abstraction for data sources that support random access by key. It is an evolution of the older [KeyValueStore](javadocs/org/apache/samza/storage/kv/KeyValueStore) API. It offers support for both local and remote data sources and composition through hybrid tables. For remote data sources, a [RemoteTable] provides optimized access with caching, rate-limiting, and retry support. Depending on the implementation, a Table can be a [ReadableTable](javadocs/org/apache/samza/table/ReadableTable) or a [ReadWriteTable](javadocs/org/apache/samza/table/ReadWriteTable).
+
+In the Low Level API, you can obtain and use a Table as follows:
- public interface ClosableTask {
- void close() throws Exception;
- }
-
-{% endhighlight %}
+1. Use the appropriate TableDescriptor to specify the table properties.
+2. Register the TableDescriptor with the TaskApplicationDescriptor.
+3. Obtain a Table reference within the task implementation using [TaskContext.getTable()](javadocs/org/apache/samza/task/TaskContext#getTable-java.lang.String-). [TaskContext](javadocs/org/apache/samza/task/TaskContext) is available via [Context.getTaskContext()](javadocs/org/apache/samza/context/Context#getTaskContext--), which in turn is available by implementing [InitiableTask. init()]((javadocs/org/apache/samza/task/InitableTask#init-org.apache.samza.context.Context-)).
-## EndOfStreamListenerTask
-This task interface defines the additional logic when a task instance has reached the end of all input SystemStreamPartitions (see Samza as a batch job).
+For example:
{% highlight java %}
- public interface EndOfStreamListenerTask {
- void onEndOfStream(MessageCollector collector, TaskCoordinator coordinator) throws Exception;
+ public class PageViewFilterTask implements StreamTask, InitableTask {
+ private final SystemStream outputStream = new SystemStream(“kafka”, “goodPageViewEvent”);
+
+ private ReadWriteTable<String, Integer> badUrlsTable;
+
+ @Override
+ public void init(Context context) {
+ badUrlsTable = (ReadWriteTable<String, Integer>) context.getTaskContext().getTable("badUrls");
+ }
+
+ @Override
+ public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) {
+ String key = (String) message.getKey();
+ if (badUrlsTable.containsKey(key)) {
+ // skip the message, increment the counter, do not send it
+ badPageUrlTable.put(key, badPageUrlTable.get(key) + 1);
+ } else {
+ collector.send(new OutgoingMessageEnvelope(outputStream, key, message.getValue())); }
+ }
}
-
+
{% endhighlight %}
-# Legacy Task Application
+### Legacy Applications
-For legacy task application which do not implement TaskApplication interface, you may specify the system, stream, and local stores in your job’s configuration, in addition to task.class. An incomplete example of configuration for legacy task application could look like this (see the [configuration](../jobs/configuration.md) documentation for more detail):
+For legacy Low Level API applications, you can continue specifying your system, stream and store properties along with your task.class in configuration. An incomplete example of configuration for legacy task application looks like this (see the [configuration](../jobs/configuration.md) documentation for more detail):
{% highlight jproperties %}
- # This is the class above, which Samza will instantiate when the job is run
+ # This is the Task class that Samza will instantiate when the job is run
task.class=com.example.samza.PageViewFilterTask
# Define a system called "kafka" (you can give it any name, and you can define