CINXE.COM

<!DOCTYPE html>  <html lang="en"> <head> <meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1, shrink-to-fit=no"> <meta charset="utf-8"> <title>Samza - Feature Preview</title> <link rel="apple-touch-icon-precomposed" sizes="57x57" href="/img/favicon/apple-touch-icon-57x57.png" /> <link rel="apple-touch-icon-precomposed" sizes="114x114" href="/img/favicon/apple-touch-icon-114x114.png" /> <link rel="apple-touch-icon-precomposed" sizes="72x72" href="/img/favicon/apple-touch-icon-72x72.png" /> <link rel="apple-touch-icon-precomposed" sizes="144x144" href="/img/favicon/apple-touch-icon-144x144.png" /> <link rel="apple-touch-icon-precomposed" sizes="60x60" href="/img/favicon/apple-touch-icon-60x60.png" /> <link rel="apple-touch-icon-precomposed" sizes="120x120" href="/img/favicon/apple-touch-icon-120x120.png" /> <link rel="apple-touch-icon-precomposed" sizes="76x76" href="/img/favicon/apple-touch-icon-76x76.png" /> <link rel="apple-touch-icon-precomposed" sizes="152x152" href="/img/favicon/apple-touch-icon-152x152.png" /> <link rel="icon" type="image/png" href="/img/favicon/favicon-196x196.png" sizes="196x196" /> <link rel="icon" type="image/png" href="/img/favicon/favicon-96x96.png" sizes="96x96" /> <link rel="icon" type="image/png" href="/img/favicon/favicon-32x32.png" sizes="32x32" /> <link rel="icon" type="image/png" href="/img/favicon/favicon-16x16.png" sizes="16x16" /> <link rel="icon" type="image/png" href="/img/favicon/favicon-128.png" sizes="128x128" /> <meta name="application-name" content="https://samza.apache.org" /> <meta name="msapplication-TileColor" content="#FFFFFF" /> <meta name="msapplication-TileImage" content="/img/favicon/mstile-144x144.png" /> <meta name="msapplication-square70x70logo" content="/img/favicon/mstile-70x70.png" /> <meta name="msapplication-square150x150logo" content="/img/favicon/mstile-150x150.png" /> <meta name="msapplication-wide310x150logo" content="/img/favicon/mstile-310x150.png" /> <meta name="msapplication-square310x310logo" content="/img/favicon/mstile-310x310.png" /> <link href="/css/ionicons.min.css" rel="stylesheet"> <link href="/css/google-fonts.css" rel="stylesheet"> <link href="/css/syntax.css" rel="stylesheet"/> <link rel="stylesheet" href="/css/main.new.css" /> </head> <body class="page">  <div class="main-navigation" data-plugin="menu"> <div class="main-navigation__toggle" data-menu-closed> </div> <div class="main-navigation__toggle main-navigation__toggle--opened" data-menu-opened> </div> <div class="main-navigation__inner"> <div class="main-navigation__logo"> <a href="/"> <img class="main-navigation__logo-img" src="/img/samza-logo.png" srcset="/img/samza-logo.png 1x, /img/samza-logo@2x.png 2x" alt="Samza Logo" /> </a> </div> <div class="main-navigation__items" data-menu-opened> <a class="main-navigation__item" href="/">Home</a> <a class="main-navigation__item" href="/learn/documentation/latest/core-concepts/core-concepts.html">Docs</a> <a class="main-navigation__item" href="/powered-by/">Powered By</a> <a class="main-navigation__item" href="/startup/download/">Downloads</a> <a class="main-navigation__item" href="/blog/">Blog</a> <div class="main-navigation__item main-navigation__item--group"> <div class="main-navigation__item-group-title"> Community </div> <div class="main-navigation__item-group-list"> <a class="main-navigation__item" href="/community/contact-us.html">Contact Us</a> <a class="main-navigation__item" href="/contribute/contributors-corner.html">Contributor's Corner</a> <a class="main-navigation__item" href="/community/committers.html">PMC Members and committers</a> <a class="main-navigation__item" href="/meetups/">Talks and Meetups</a> </div> </div> </div> </div> </div> <div class="container"> <div class="container__toggle"> </div>   <div class="side-navigation">  <div class="side-navigation__group side-navigation__group--has-nested" data-plugin="sub-menu" data-sub-menu-show-class="side-navigation__group--has-nested-visible">  <div class="side-navigation__group-title"> Getting Started </div> <div class="side-navigation__group-items " data-sub-menu >  <a class="side-navigation__group-item" data-match-active="" href="/startup/quick-start/latest/">QuickStart</a> <a class="side-navigation__group-item" data-match-active="" href="/startup/code-examples/latest/">Code Examples</a>   </div>  </div>  <div class="side-navigation__group side-navigation__group--has-nested" data-plugin="sub-menu" data-sub-menu-show-class="side-navigation__group--has-nested-visible">  <div class="side-navigation__group-title"> Documentation </div> <div class="side-navigation__group-items side-navigation__group-has-submenus" data-sub-menu data-documentation="/learn/documentation/latest/">    </div>  </div>  <div class="side-navigation__group side-navigation__group--has-nested" data-plugin="sub-menu" data-sub-menu-show-class="side-navigation__group--has-nested-visible">  <div class="side-navigation__group-title"> Releases </div> <div class="side-navigation__group-items " data-sub-menu >  <a class="side-navigation__group-item" data-match-active="" href="/releases/1.8.0">1.8.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.7.0">1.7.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.6.0">1.6.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.5.1">1.5.1</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.5.0">1.5.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.4.0">1.4.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.3.1">1.3.1</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.3.0">1.3.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.2.0">1.2.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.1.0">1.1.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/1.0.0">1.0.0</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/0.14">0.14</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/0.13">0.13</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/0.12">0.12</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/0.11">0.11</a> <a class="side-navigation__group-item" data-match-active="" href="/releases/0.10">0.10</a>   </div>  </div>  <div class="side-navigation__group">  <a class="side-navigation__group-title" data-plugin="top-menu" data-match-active="" href="/blog/"> Blog </a>     </div>  <div class="side-navigation__group side-navigation__group--has-nested" data-plugin="sub-menu" data-sub-menu-show-class="side-navigation__group--has-nested-visible">  <div class="side-navigation__group-title"> Community </div> <div class="side-navigation__group-items " data-sub-menu >  <a class="side-navigation__group-item" data-match-active="" href="/community/contact-us.html">Contact Us</a> <a class="side-navigation__group-item" data-match-active="" href="/contribute/contributors-corner.html">Contributor's Corner</a> <a class="side-navigation__group-item" data-match-active="" href="/contribute/enhancement-proposal.html">Enhancement Proposal</a> <a class="side-navigation__group-item" data-match-active="" href="/community/committers.html">PMC members & Committers</a> <a class="side-navigation__group-item" data-match-active="" href="/meetups/">Talks and Meetups</a>   </div>  </div>  <div class="side-navigation__group side-navigation__group--has-nested" data-plugin="sub-menu" data-sub-menu-show-class="side-navigation__group--has-nested-visible">  <div class="side-navigation__group-title"> Case Studies </div> <div class="side-navigation__group-items " data-sub-menu >  <a class="side-navigation__group-item" data-match-active="exact" href="/case-studies/">View All</a> <hr>  <a class="side-navigation__group-item" href="/case-studies/ebay" data-match-active="">eBay</a> <a class="side-navigation__group-item" href="/case-studies/tripadvisor" data-match-active="">TripAdvisor</a> <a class="side-navigation__group-item" href="/case-studies/slack" data-match-active="">Slack</a> <a class="side-navigation__group-item" href="/case-studies/optimizely" data-match-active="">Optimizely</a> <a class="side-navigation__group-item" href="/case-studies/redfin" data-match-active="">Redfin</a> <a class="side-navigation__group-item" href="/case-studies/linkedin" data-match-active="">LinkedIn</a>  </div>  </div> </div> <div class="section"> <div class="content"> <h2>Feature Preview</h2>  <ol> <li><a href="#overview">Overview</a></li> <li><a href="#try-it-out">Try It Out</a></li> <li><a href="#architecture">Architecture</a></li> <li><a href="#high-level-api">High Level API</a></li> <li><a href="#flexible-deployment-model">Flexible Deployment Model</a></li> </ol> <hr /> <h2 id="overview">Overview</h2> Samza 0.13.0 introduces a new programming model and a new deployment model. They’re being released as a preview because they represent major enhancements to how developers work with Samza, so it is beneficial for both early adopters and the Samza development community to experiment with the release and provide feedback. The following sections introduce the new features and link to tutorials which demonstrate how to use them. Please try them and send feedback to the <a href="mailto:dev@samza.apache.org">dev mailing list</a> <hr /> <h3 id="try-it-out">Try it Out</h3> Want to skip all the details and get some hands on experience? There are three tutorials to help you get acquainted with running Samza applications in both YARN and embedded modes and programming with the high level API: <ul> <li><a href="/learn/tutorials/latest/hello-samza-high-level-yarn.html">Yarn Deployment</a> - run a pre-existing Wikipedia application on YARN and observe the output.</li> <li><a href="/learn/tutorials/latest/hello-samza-high-level-code.html">High Level API Code Walkthrough</a> - walk through building the Wikipedia application, step by step.</li> <li><a href="/learn/tutorials/latest/hello-samza-high-level-zk.html">ZooKeeper Deployment</a> - run a pre-existing Wikipedia application with ZooKeeper coordination and observe the output.</li> </ul> <hr /> <h2 id="architecture">Architecture</h2> <h3 id="introduction">Introduction</h3> The Samza high level API provides a unified way to handle both streaming and batch data. You can describe the end-to-end application logic in a single program with operators like map, filter, window, and join to accomplish what previously required multiple jobs. The API is designed to be portable. The same application code can be deployed in batch or streaming modes, embedded or with a cluster manager environments, and can switch between Kafka, Kinesis, HDFS or other systems with a simple configuration change. This portability is enabled by a new architecture which is described in the sections below. <h3 id="concepts">Concepts</h3> The Samza architecture has been overhauled with distinct layers to handle each stage of application development. The following diagram shows an overview of Apache Samza architecture with the high level API. <img src="/img/latest/learn/documentation/introduction/layered-arch.png" alt="Architecture diagram" style="max-width: 100%; height: auto;" onclick="window.open(this.src)" /> There are four layers in the architecture. The following sections describe each of the layers. <h4 id="i-high-level-api">I. High Level API</h4> The high level API provides the libraries to define your application logic. The <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/application/StreamApplication.html">StreamApplication</a> is the central abstraction which your application must implement. You start by declaring your inputs as instances of <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/MessageStream.html">MessageStream</a>. 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. For a deeper dive into the high level API, see <a href="#high-level-api">high level API section</a> below. <h4 id="ii-applicationrunner">II. ApplicationRunner</h4> Samza uses an <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/runtime/ApplicationRunner.html">ApplicationRunner</a> to run a stream application. The ApplicationRunner generates the configs (such as input/output streams), creates intermediate streams, and starts the execution. There are two types of ApplicationRunner: RemoteApplicationRunner - submits the application to a remote cluster. This runner is invoked via the run-app.sh script. To use RemoteApplicationRunner, set the following configurations <figure class="highlight"><pre><code class="language-jproperties" data-lang="jproperties"># The StreamApplication class to run app.class=com.company.job.YourStreamApplication job.factory.class=org.apache.samza.job.yarn.YarnJobFactory</code></pre></figure> Then use run-app.sh to run the application in the remote cluster. The script will invoke the RemoteApplicationRunner, which will launch one or more jobs using the factory specified with job.factory.class. Follow the <a href="/learn/tutorials/latest/hello-samza-high-level-yarn.html">yarn deployment tutorial</a> to try it out. LocalApplicationRunner - runs the application in the JVM process of the runner. For example, to launch your application on multiple machines using ZooKeeper for coordination, you can run multiple instances of LocalApplicationRunner on various machines. After the applications load they will start cordinatinating their actions through ZooKeeper. Here is an example to run the StreamApplication in your program using the LocalApplicationRunner: <figure class="highlight"><pre><code class="language-java" data-lang="java">public static void main(String[] args) throws Exception { CommandLine cmdLine = new CommandLine(); Config config = cmdLine.loadConfig(cmdLine.parser().parse(args)); LocalApplicationRunner localRunner = new LocalApplicationRunner(config); StreamApplication app = new YourStreamApplication(); localRunner.run(app); // Wait for the application to finish localRunner.waitForFinish(); System.out.println("Application completed with status " + localRunner.status(app)); }</code></pre></figure> Follow the <a href="/learn/tutorials/latest/hello-samza-high-level-zk.html">ZooKeeper deployment tutorial</a> to try it out. <h5 id="execution-plan">Execution Plan</h5> The ApplicationRunner generates a physical execution plan for your processing logic before it starts executing it. The plan represents the runtime structure of the application. Particularly, it provides visibility into the generated intermediate streams. Once the job is deployed, the plan can be viewed as follows: <ul> <li>For applications launched using run-app.sh, Samza will create a plan directory under your application deployment directory and write the plan.json file there.</li> <li>For the applications launched using your own script (e.g. for LocalApplicationRunner), please create a plan directory at the same level as bin, and point the <code class="language-plaintext highlighter-rouge">EXECUTION_PLAN_DIR</code> environment variable to its location.</li> </ul> To view the plan, open the bin/plan.html file in a browser. Here’s a sample plan visualization: <img src="/img/latest/learn/documentation/introduction/execution-plan.png" alt="Execution plan" style="max-width: 100%; height: auto;" onclick="window.open(this.src)" /> <h4 id="iii-execution-models">III. Execution Models</h4> Samza supports two types of execution models: cluster based execution and embedded execution. In cluster based execution, Samza will run and manage your application on a multi-tenant cluster. Samza ships with support for <a href="http://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/YARN.html">YARN</a>. You can implement your own StreamJob and a corresponding ResourceManagerFactory to add support for another cluster manager. In the embedded execution model, you can use Samza as a lightweight library within your application. You can spin up multiple instances of your application which will distribute and coordinate processing among themselves. This mode provides flexibility for running your applications in arbitrary hosting environments:It also supports pluggable coordination logic with out-of-the-box support for two types of coordination: <ul> <li>ZooKeeper based coordination - Samza can be configured to use ZooKeeper to manage group membership and partition assignment among instances of your application. This allows the you to dynamically scale your application by spinning up more instances or scaling down by shutting some down.</li> <li>External coordination - Samza can run your application in a single JVM locally without coordination, or multiple JVMs with a static partition assignment. This is helpful when running in containerized environments like Kubernetes or Amazon ECS.</li> </ul> For more details on running Samza in embedded mode, take a look at the <a href="#flexible-deployment-model">flexible deployment model</a> section below. <h4 id="iv-processor">IV. Processor</h4> The lowest execution unit of a Samza application is the processor. It reads the configs generated from the <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/runtime/ApplicationRunner.html">ApplicationRunner</a> and processes the input stream partitions assigned by the JobCoordinator. It can access local state using a <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/storage/kv/KeyValueStore.html">KeyValueStore</a> implementation (e.g. RocksDB or in-memory) and remote state (e.g. REST service) using multithreading. <hr /> <h2 id="high-level-api">High Level API</h2> 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. <h2 id="code-examples">Code Examples</h2> Check out some examples to see the high-level API in action. <ol> <li><a href="https://github.com/apache/samza-hello-samza/blob/e5943a000eef87e077c422e09dc20f09d4e876ca/src/main/java/samza/examples/cookbook/PageViewAdClickJoiner.java">Pageview AdClick Joiner</a> demonstrates joining a stream of PageViews with a stream of AdClicks, e.g. to analyze which pages get the most ad clicks.</li> <li><a href="https://github.com/apache/samza-hello-samza/blob/e5943a000eef87e077c422e09dc20f09d4e876ca/src/main/java/samza/examples/cookbook/PageViewFilterApp.java">Pageview Repartitioner</a> illustrates re-partitioning the incoming stream of PageViews.</li> <li><a href="https://github.com/apache/samza-hello-samza/blob/e5943a000eef87e077c422e09dc20f09d4e876ca/src/main/java/samza/examples/cookbook/PageViewSessionizerApp.java">Pageview Sessionizer</a> groups the incoming stream of events into sessions based on user activity.</li> <li><a href="https://github.com/apache/samza-hello-samza/blob/e5943a000eef87e077c422e09dc20f09d4e876ca/src/main/java/samza/examples/cookbook/TumblingPageViewCounterApp.java">Pageview by Region</a> counts the number of views per-region over tumbling time intervals.</li> </ol> <h2 id="key-concepts">Key Concepts</h2> <h3 id="streamapplication">StreamApplication</h3> When writing your stream processing application using the Samza high-level API, you should implement a <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/application/StreamApplication.html">StreamApplication</a> and define your processing logic in the init method. <figure class="highlight"><pre><code class="language-java" data-lang="java">public void init(StreamGraph graph, Config config) { … }</code></pre></figure> For example, here is a StreamApplication that validates and decorates page views with viewer’s profile information. <figure class="highlight"><pre><code class="language-java" data-lang="java">public class BadPageViewFilter implements StreamApplication { @Override public void init(StreamGraph graph, Config config) { MessageStream<PageView> pageViews = graph.getInputStream(“page-views”, new JsonSerdeV2<>(PageView.class)); pageViews.filter(this::isValidPageView) .map(this::addProfileInformation) .sendTo(graph.getOutputStream(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class))) } }</code></pre></figure> <h3 id="messagestream">MessageStream</h3> 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: <ol> <li> Using StreamGraph.getInputStream to get the MessageStream for a given input stream (e.g., a Kafka topic). </li> <li> By transforming an existing MessageStream using operations like map, filter, window, join etc. </li> </ol> <h2 id="anatomy-of-a-typical-samza-streamapplication">Anatomy of a typical Samza StreamApplication</h2> There are 3 simple steps to write your stream processing applications using the Samza high-level API. <h3 id="step-1-obtain-the-input-streams">Step 1: Obtain the input streams:</h3> You can obtain the MessageStream for your input stream ID (“page-views”) using StreamGraph.getInputStream. <figure class="highlight"><pre><code class="language-java" data-lang="java"> MessageStream<PageView> pageViewInput = graph.getInputStream(“page-views”, new JsonSerdeV2<>(PageView.class)); </code></pre></figure> The first parameter <code class="language-plaintext highlighter-rouge">page-views</code> is the logical stream ID. Each stream ID is associated with a physical name and a 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. <figure class="highlight"><pre><code class="language-jproperties" data-lang="jproperties">streams.page-views.samza.system=kafka systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory systems.kafka.consumer.zookeeper.connect=localhost:2181 systems.kafka.producer.bootstrap.servers=localhost:9092 streams.page-views.samza.physical.name=PageViewEvent</code></pre></figure> The second parameter is a serde to de-serialize the incoming message. <h3 id="step-2-define-your-transformation-logic">Step 2: Define your transformation logic:</h3> You are now ready to define your StreamApplication logic as a series of transformations on MessageStreams. <figure class="highlight"><pre><code class="language-java" data-lang="java">MessageStream<DecoratedPageViews> decoratedPageViews = pageViewInput.filter(this::isValidPageView) .map(this::addProfileInformation);</code></pre></figure> <h3 id="step-3-write-the-output-to-an-output-stream">Step 3: Write the output to an output stream:</h3> Finally, you can create an OutputStream using StreamGraph.getOutputStream and send the transformed messages through it. <figure class="highlight"><pre><code class="language-java" data-lang="java">// Send messages with userId as the key to “decorated-page-views”. decoratedPageViews.sendTo( graph.getOutputStream(“decorated-page-views”, new JsonSerdeV2<>(DecoratedPageView.class)));</code></pre></figure> The first parameter <code class="language-plaintext highlighter-rouge">decorated-page-views</code> is a logical stream ID. The properties for this stream ID can be overridden just like the stream IDs for input streams. For example: <figure class="highlight"><pre><code class="language-jproperties" data-lang="jproperties">streams.decorated-page-views.samza.system=kafka streams.decorated-page-views.samza.physical.name=DecoratedPageViewEvent</code></pre></figure> The second parameter is a serde to de-serialize the outgoing message. <h2 id="operators">Operators</h2> The high level API supports common operators like map, flatmap, filter, merge, joins, and windowing on streams. Most of these operators accept corresponding Functions and these functions are <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/InitableFunction.html">Initable</a>. <h3 id="map">Map</h3> Applies the provided 1:1 <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/MapFunction.html">MapFunction</a> 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). <figure class="highlight"><pre><code class="language-java" data-lang="java">MessageStream<Integer> numbers = ... MessageStream<Integer> tripled= numbers.map(m -> m * 3) MessageStream<String> stringified = numbers.map(m -> String.valueOf(m))</code></pre></figure> <h3 id="flatmap">Flatmap</h3> Applies the provided 1:n <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/FlatMapFunction.html">FlatMapFunction</a> to each element in the MessageStream and returns the transformed MessageStream. The FlatMapFunction takes in a single message and returns zero or more messages. <figure class="highlight"><pre><code class="language-java" data-lang="java">MessageStream<String> sentence = ... // Parse the sentence into its individual words splitting by space MessageStream<String> words = sentence.flatMap(sentence -> Arrays.asList(sentence.split(“ ”))</code></pre></figure> <h3 id="filter">Filter</h3> Applies the provided <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/FilterFunction.html">FilterFunction</a> 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. <figure class="highlight"><pre><code class="language-java" data-lang="java">MessageStream<String> words = ... // Extract only the long words MessageStream<String> longWords = words.filter(word -> word.size() > 15); // Extract only the short words MessageStream<String> shortWords = words.filter(word -> word.size() < 3);</code></pre></figure> <h3 id="partitionby">PartitionBy</h3> 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. <figure class="highlight"><pre><code class="language-java" data-lang="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 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.</code></pre></figure> <h3 id="merge">Merge</h3> Merges the MessageStream with all the provided MessageStreams and returns the merged stream. <figure class="highlight"><pre><code class="language-java" data-lang="java">MessageStream<ServiceCall> serviceCall1 = ... MessageStream<ServiceCall> serviceCall2 = ... // Merge individual “ServiceCall” streams and create a new merged MessageStream MessageStream<ServiceCall> serviceCallMerged = serviceCall1.merge(serviceCall2)</code></pre></figure> The merge transform preserves the order of each MessageStream, so if message <code class="language-plaintext highlighter-rouge">m1</code> appears before <code class="language-plaintext highlighter-rouge">m2</code> in any provided stream, then, <code class="language-plaintext highlighter-rouge">m1</code> also appears before <code class="language-plaintext highlighter-rouge">m2</code> in the merged stream. As an alternative to the <code class="language-plaintext highlighter-rouge">merge</code> instance method, you also can use the <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/MessageStream.html#mergeAll-java.util.Collection-">MessageStream#mergeAll</a> static method to merge MessageStreams without operating on an initial stream. <h3 id="sendto-stream">SendTo (stream)</h3> 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. <figure class="highlight"><pre><code class="language-java" data-lang="java">// Output a new message with userId as the key and region as the value to the “user-region” stream. MessageStream<PageView> pageViews = ... MessageStream<KV<String, PageView>> keyedPageViews = pageViews.map(KV.of(pageView.getUserId(), pageView.getRegion())); OutputStream<KV<String, String>> userRegions = graph.getOutputStream(“user-region”, KVSerde.of(new StringSerde(), new StringSerde())); keyedPageViews.sendTo(userRegions);</code></pre></figure> <h3 id="sendto-table">SendTo (table)</h3> Sends all messages from this MessageStream to the provided table, the expected message type is KV. <figure class="highlight"><pre><code class="language-java" data-lang="java"> // Write a new message with memberId as the key and profile as the value to a table. streamGraph.getInputStream("Profile", new NoOpSerde<Profile>()) .map(m -> KV.of(m.getMemberId(), m)) .sendTo(table);</code></pre></figure> <h3 id="sink">Sink</h3> Allows sending messages from this MessageStream to an output system using the provided <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/SinkFunction.html">SinkFunction</a>. This offers more control than sendTo since the SinkFunction has access to the <code class="language-plaintext highlighter-rouge">MessageCollector</code> and the <code class="language-plaintext highlighter-rouge">TaskCoordinator</code>. 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.) <figure class="highlight"><pre><code class="language-java" data-lang="java">// Repartition pageView by userId. MessageStream<PageView> pageViews = ... 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)); } )</code></pre></figure> <h3 id="join-stream-stream">Join (stream-stream)</h3> The stream-stream Join operator joins messages from two MessageStreams using the provided pairwise <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/JoinFunction.html">JoinFunction</a>. 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. <figure class="highlight"><pre><code class="language-java" data-lang="java">// Joins a stream of OrderRecord with a stream of ShipmentRecord by orderId with a TTL of 20 minutes. // Results are produced to a new stream of FulfilledOrderRecord. MessageStream<OrderRecord> orders = … MessageStream<ShipmentRecord> shipments = … MessageStream<FulfilledOrderRecord> shippedOrders = orders.join(shipments, new OrderShipmentJoiner(), new StringSerde(), // serde for the join key new JsonSerdeV2<>(OrderRecord.class), new JsonSerdeV2<>(ShipmentRecord.class), // serde for both streams Duration.ofMinutes(20), // join TTL "shipped-order-stream") // operator ID // Constructs a new FulfilledOrderRecord by extracting the order timestamp from the OrderRecord and the shipment timestamp from the ShipmentRecord. class OrderShipmentJoiner implements JoinFunction<String, OrderRecord, ShipmentRecord, FulfilledOrderRecord> { @Override public FulfilledOrderRecord apply(OrderRecord message, ShipmentRecord otherMessage) { return new FulfilledOrderRecord(message.orderId, message.orderTimestamp, otherMessage.shipTimestamp); } @Override public String getFirstKey(OrderRecord message) { return message.orderId; } @Override public String getSecondKey(ShipmentRecord message) { return message.orderId; } }</code></pre></figure> <h3 id="join-stream-table">Join (stream-table)</h3> The stream-table Join operator joins messages from a MessageStream using the provided <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/StreamTableJoinFunction.html">StreamTableJoinFunction</a>. 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. <figure class="highlight"><pre><code class="language-java" data-lang="java"> streamGraph.getInputStream("PageView", new NoOpSerde<PageView>()) .partitionBy(PageView::getMemberId, v -> v, "p1") .join(table, new PageViewToProfileJoinFunction()) ...</code></pre></figure> <figure class="highlight"><pre><code class="language-java" data-lang="java"><table class="rouge-table"><tbody><tr><td class="gutter gl"><pre class="lineno">1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 </pre></td><td class="code"><pre>public class PageViewToProfileJoinFunction implements StreamTableJoinFunction <Integer, KV<Integer, PageView>, KV<Integer, Profile>, EnrichedPageView> { @Override public EnrichedPageView apply(KV<Integer, PageView> m, KV<Integer, Profile> r) { return r != null ? new EnrichedPageView(...) : null; } @Override public Integer getMessageKey(KV<Integer, PageView> message) { return message.getKey(); } @Override public Integer getRecordKey(KV<Integer, Profile> record) { return record.getKey(); } } </pre></td></tr></tbody></table></code></pre></figure> <h3 id="window">Window</h3> <h4 id="windowing-concepts">Windowing Concepts</h4> 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 <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/windows/Window.html#setEarlyTrigger-org.apache.samza.operators.triggers.Trigger-">early triggers</a> that allow emitting results speculatively before all data for the window has arrived, or late triggers that allow handling late messages for the window. Aggregator Function: By default, the emitted WindowPane will contain all the messages for the window. Instead of retaining all messages, you typically define a more compact data structure for the WindowPane and update it incrementally as new messages arrive, e.g. for keeping a count of messages in the window. To do this, you can provide an aggregating <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/operators/functions/FoldLeftFunction.html">FoldLeftFunction</a> which is invoked for each incoming message added to the window and defines how to update the WindowPane for that message. Accumulation Mode: A window’s accumulation mode determines how results emitted from a window relate to previously emitted results for the same window. This is particularly useful when the window is configured with early or late triggers. The accumulation mode can either be discarding or accumulating. A discarding window clears all state for the window at every emission. Each emission will only correspond to new messages that arrived since the previous emission for the window. An accumulating window retains window results from previous emissions. Each emission will contain all messages that arrived since the beginning of the window. <h4 id="window-types">Window Types:</h4> The Samza high-level API currently supports tumbling and session windows. Tumbling Window: A tumbling window defines a series of contiguous, fixed size time intervals in the stream. Examples: <figure class="highlight"><pre><code class="language-java" data-lang="java">// Group the pageView stream into 3 second tumbling windows keyed by the userId. MessageStream<PageView> pageViews = ... 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. 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()));</code></pre></figure> Session Window: A session window groups a MessageStream into sessions. A session captures a period of activity over a MessageStream and is defined by a gap. A session is closed and results are emitted if no new messages arrive for the window for the gap duration. Examples: <figure class="highlight"><pre><code class="language-java" data-lang="java">// Sessionize a stream of page views, and count the number of page-views in a session for every user. MessageStream<PageView> pageViews = … 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, new StringSerde(), new IntegerSerde())); // Compute the maximum value over tumbling windows of 3 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()))</code></pre></figure> <h3 id="table">Table</h3> 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: <ol> <li>Create a table</li> <li>Populate the table using the sendTo() operator</li> <li>Join a stream with the table using the join() operator</li> </ol> <figure class="highlight"><pre><code class="language-java" data-lang="java"><table class="rouge-table"><tbody><tr><td class="gutter gl"><pre class="lineno">1 2 3 4 5 </pre></td><td class="code"><pre>final StreamApplication app = (streamGraph, cfg) -> { Table<KV<Integer, Profile>> table = streamGraph.getTable(new InMemoryTableDescriptor("t1") .withSerde(KVSerde.of(new IntegerSerde(), new ProfileJsonSerde()))); ... }; </pre></td></tr></tbody></table></code></pre></figure> Example above creates a TableDescriptor object, which contains all information about a table. The currently supported table types are <a href="https://github.com/apache/samza/blob/master/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java">InMemoryTableDescriptor</a> and <a href="https://github.com/apache/samza/blob/master/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/rocksdb/RocksDbTableDescriptor.java">RocksDbTableDescriptor</a>. Notice the type of records in a table is KV, and <a href="https://samza.apache.org/learn/documentation/latest/container/serialization.html">Serdes</a> for both key and value of records needs to be defined (line 4). Additional parameters can be added based on individual table types. More details about step 2 and 3 can be found at operator section. <hr /> <h2 id="flexible-deployment-model">Flexible Deployment Model</h2> <h3 id="introduction-1">Introduction</h3> Prior to Samza 0.13.0, Samza only supported cluster-managed deployment with <a href="http://hadoop.apache.org/docs/current/hadoop-yarn/hadoop-yarn-site/YARN.html">YARN</a>. With Samza 0.13.0, the deployment model has been simplified and decoupled from YARN. If you prefer cluster management, you can still use YARN or you can implement your own extension to deploy Samza on other cluster management systems. But what if you want to avoid cluster management systems altogether? Samza now ships with the ability to deploy applications as a simple embedded library with pluggable coordination. With embedded mode, you can leverage Samza processors directly in your application and deploy it in whatever way you prefer. Samza has a pluggable job coordinator layer to perform leader election and assign work to the processors. This section will focus on the new embedded deployment capability. <h3 id="concepts-1">Concepts</h3> Let’s take a closer look at how embedded deployment works. The <a href="#architecture">architecture</a> section above provided an overview of the layers that enable the flexible deployment model. The new embedded mode comes into the picture at the deployment layer. The deployment layer includes assignment of input partitions to the available processors. There are two types of partition assignment models which are controlled with the job.coordinator.factory in configuration: <h4 id="external-partition-management">External Partition Management</h4> With external partition management, Samza doesn’t manage the partitioning by itself. Instead it uses a <code class="language-plaintext highlighter-rouge">PassthroughJobCoordinator</code> which honors whatever partition mapping is provided by the <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/container/grouper/stream/SystemStreamPartitionGrouper.html">SystemStreamPartitionGrouper</a>. There are two common patterns for external partition management: <ul> <li>Using high level Kafka consumer - partition assignment is done by the high level Kafka consumer itself. To use this model, you need to implement and configure a SystemFactory which provides the Kafka high level consumer. Then you need to configure job.systemstreampartition.grouper.factory to org.apache.samza.container.grouper.stream.AllSspToSingleTaskGrouper so Kafka’s partition assignments all go to one task.</li> <li>Customized partitioning - partition assignment is done with a custom grouper. The grouper logic is completely up to you. A practical example of this model is to implement a custom grouper which reads static partition assignments from the configuration.</li> </ul> Samza ships with a <code class="language-plaintext highlighter-rouge">PassthroughJobCoordinatorFactory</code> which facilitates this type of partition management. <h4 id="dynamic-partition-management">Dynamic Partition Management</h4> With dynamic partitioning, partitions are distributed between the available processors at runtime. If the number of available processors changes (for example, if some are shut down or added) the mapping will be regenerated and re-distributed to all the processors. Information about current mapping is contained in a special structure called the JobModel. There is one leader processor which generates the JobModel and distributes it to the other processors. The leader is determined by a “leader election” process. Let’s take a closer look at how dynamic coordination works. <h4 id="coordination-service">Coordination service</h4> Dynamic coordination of the processors assumes presence of a coordination service. The main responsibilities of the service are: <ul> <li>Leader Election - electing a single processor, which will be responsible for JobModel calculation and distribution or for intermediate streams creation.</li> <li>Central barrier and latch - coordination primitives used by the processors.</li> <li>JobModel notifications - notifying the processors about availability of a new JobModel.</li> <li>JobModel storage the coordination service dictates where the JobModel is persisted.</li> </ul> The coordination service is currently derived from the job coordinator factory. Samza ships with a <code class="language-plaintext highlighter-rouge">ZkJobCoordinatorFactory</code> implementation which has a corresponding <code class="language-plaintext highlighter-rouge">ZkCoordinationServiceFactory</code>. Let’s walk through the coordination sequence for a ZooKeeper based embedded application: <ul> <li>Each processor (participant) will register with the pluggable coordination service. During the registration it will provide its own participant ID.</li> <li>One of the participants will be elected as the leader.</li> <li>The leader monitors the list of all the active participants.</li> <li>Whenever the list of the participants changes, the leader will generate a new JobModel for the current participants.</li> <li>The new JobModel will be pushed to a common storage. The default implementation uses ZooKeeper for this purpose.</li> <li>Participants are notified that the new JobModel is available. Notification is done through the coordination services, e.g. ZooKeeper.</li> <li>The participants will stop processing, apply the new JobModel, and then resume processing.</li> </ul> The following diagram shows the relationships of the coordinators in the ZooKeeper coordination service implementation. <img src="/img/latest/learn/documentation/introduction/coordination-service.png" alt="Coordination service diagram" style="max-width: 100%; height: auto;" onclick="window.open(this.src)" /> Here are a few important details about the coordination service: <ul> <li>In order to ensure that no two partitions are processed twice by different processors, processing is paused and the processors synchronize on a barrier. Once all the processors are paused, the new JobModel is applied and the processing resumes. The barrier is implemented using the coordination service.</li> <li>During startup and shutdown the processors will be joining/leaving one after another. To avoid redundant JobModel re-calculation, there is a debounce timer which waits for some short period of time (2 seconds by default, configurable in a future release) for more processors to join or leave. Each time a processor joins or leaves, the timer is reset. When the timer expires the JobModel is finally recalculated.</li> <li>If the processors require local store for adjacent or temporary data, we would want to keep its mapping across restarts. For this we uses some extra information about each processor, which uniquely identifies it and its location. If the same processor is restarted on the same location we will try to assign it the same partitions. This locality information should survive the restarts, so it is stored on a common storage (currently using ZooKeeper).</li> </ul> <h3 id="user-guide">User guide</h3> Embedded deployment is designed to help users who want more control over the deployment of their application. So it is the user’s responsibility to configure and deploy the processors. In case of ZooKeeper coordination, you also need to configure the URL for an instance of ZooKeeper. Additionally, each processor requires a unique ID to be used with the coordination service. If location affinity is important, this ID should be unique for each processor on a specific hostname (assuming local Storage services). To address this requirement, Samza uses a <a href="/learn/documentation/latest/api/javadocs/org/apache/samza/runtime/ProcessorIdGenerator.html">ProcessorIdGenerator</a> to provide the ID for each processor. If no generator is explicitly configured, the default one will create a UUID for each processor. <h4 id="configuration">Configuration</h4> To run an embedded Samza processor, you need to configure the coordinator service using the job.coordinator.factory property. Also, there is currently one taskname grouper that supports embedded mode, so you must configure that explicitly. Let’s take a look at how to configure the two coordination service implementations that ship with Samza. To use ZooKeeper-based coordination, the following configs are required: <figure class="highlight"><pre><code class="language-jproperties" data-lang="jproperties">job.coordinator.factory=org.apache.samza.zk.ZkJobCoordinatorFactory job.coordinator.zk.connect=yourzkconnection task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerIdsFactory</code></pre></figure> To use external coordination, the following configs are needed: <figure class="highlight"><pre><code class="language-properties" data-lang="properties">job.coordinator.factory=org.apache.samza.standalone.PassthroughJobCoordinatorFactory task.name.grouper.factory=org.apache.samza.container.grouper.task.GroupByContainerIdsFactory</code></pre></figure> <h4 id="api">API</h4> As mentioned in the <a href="#architecture">architecture</a> section above, you use the LocalApplicationRunner to launch your processors from your application code, like this: <figure class="highlight"><pre><code class="language-java" data-lang="java">public class WikipediaZkLocalApplication { public static void main(String[] args) { CommandLine cmdLine = new CommandLine(); OptionSet options = cmdLine.parser().parse(args); Config config = cmdLine.loadConfig(options); LocalApplicationRunner runner = new LocalApplicationRunner(config); WikipediaApplication app = new WikipediaApplication(); runner.run(app); runner.waitForFinish(); } }</code></pre></figure> In the code above, <code class="language-plaintext highlighter-rouge">WikipediaApplication</code> is an application written with the <a href="#high-level-api">high level API</a>. Check out the <a href="/learn/tutorials/latest/hello-samza-high-level-zk.html">tutorial</a> to run this application with ZooKeeper coordination on your machine now. <h4 id="deployment-and-scaling">Deployment and Scaling</h4> You can deploy the application instances in any way you prefer. If using the coordination service, you can add or remove instances at any time and the leader’s job coordinator (elected via the CoordinationService) will automatically recalculate the JobModel after the debounce time and apply it to the available processors. So, to scale up your application, you simply start more processors. <h3 id="known-issues">Known issues</h3> Take note of the following issues with the embedded deployment feature for the 0.13.0 release. They will be fixed in a subsequent release. <ul> <li>The GroupByContainerCount default taskname grouper isn’t supported.</li> <li>Host affinity is not enabled.</li> <li>ZkJobCoordinator metrics are not provided yet. Metrics will soon be added for <ul> <li>Number of JobModel recalculations</li> <li>Number of Read/Writes</li> <li>Leader reelections</li> <li>more..</li> </ul> </li> <li>The LocalApplicationRunner does not yet support the low level API. This means you cannot use StreamTask with LocalApplicationRunner.</li> <li>Currently, ‘app.id’ config must be unique for all the application using this zk cluster.</li> </ul> </div> </div> </div>   <footer> <div class="footer-inner"> <div class="side-by-side"> <div> <div class="footer__heading">Learn More</div> <div class="footer__items"> <a class="footer__item" href="/meetups/">Meetups</a> <a class="footer__item" href="/blog/">Blog</a> <a class="footer__item" href="/learn/documentation/latest/introduction/background.html">About</a> </div> </div> <div> <div class="footer__heading">Community</div> <div class="footer__items"> <a class="footer__item" href="/community/contact-us.html">Contact Us</a> <a class="footer__item" href="/contribute/contributors-corner.html">Contributors' Corner</a> <a class="footer__item" href="/community/committers.html">PMC members and committers</a> <a class="footer__item" href="/powered-by/">Powered By</a> </div> </div> <div> <div class="quick-links"> <a class="quick-link" href="/startup/download" target="_blank"> </a> <a class="quick-link" href="https://git-wip-us.apache.org/repos/asf?p=samza.git;a=tree" target="_blank"> </a> <a class="quick-link" href="https://twitter.com/samzastream" target="_blank"> </a> </div> <script>document.write(new Date().getFullYear());</script> © samza.apache.org </div> </div> </div> </footer>  <script> (function (i, s, o, g, r, a, m) { i['GoogleAnalyticsObject'] = r; i[r] = i[r] || function () { (i[r].q = i[r].q || []).push(arguments) }, i[r].l = 1 * new Date(); a = s.createElement(o), m = s.getElementsByTagName(o)[0]; a.async = 1; a.src = g; m.parentNode.insertBefore(a, m) })(window, document, 'script', '//www.google-analytics.com/analytics.js', 'ga'); ga('create', 'UA-43122768-1', 'apache.org'); ga('send', 'pageview'); </script> <script src="/js/main.new.js"></script> </body> </html>