Modified: samza/site/learn/documentation/latest/api/table-api.html URL: http://svn.apache.org/viewvc/samza/site/learn/documentation/latest/api/table-api.html?rev=1906774&r1=1906773&r2=1906774&view=diff ============================================================================== --- samza/site/learn/documentation/latest/api/table-api.html (original) +++ samza/site/learn/documentation/latest/api/table-api.html Wed Jan 18 19:33:25 2023 @@ -227,6 +227,12 @@ + <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> @@ -538,6 +544,14 @@ + <li class="hide"><a href="/learn/documentation/1.8.0/api/table-api">1.8.0</a></li> + + + + <li class="hide"><a href="/learn/documentation/1.7.0/api/table-api">1.7.0</a></li> + + + <li class="hide"><a href="/learn/documentation/1.6.0/api/table-api">1.6.0</a></li> @@ -643,7 +657,7 @@ <p>Samza Table API is an abstraction for data sources that support random access by key, which simplifies stream-table-join. It is the natural -evolution of the existing <a href="https://github.com/apache/samza/blob/master/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngine.scala";>storage API</a>, +evolution of the existing [storage API] (https://github.com/apache/samza/blob/master/samza-kv/src/main/scala/org/apache/samza/storage/kv/KeyValueStorageEngine.scala), and it offers support for both local and remote data sources and composition through hybrid tables.</p> @@ -673,109 +687,109 @@ which are all encapsulated under the Sam <h1 id="sample-applications">Sample Applications</h1> <p>Sample applications demonstrating how to use Samza Table API can be found -<a href="https://github.com/apache/samza-hello-samza/tree/latest/src/main/java/samza/examples/cookbook";>here</a>.</p> +[here] (https://github.com/apache/samza-hello-samza/tree/latest/src/main/java/samza/examples/cookbook).</p> <h1 id="architecture-and-concepts">Architecture and Concepts</h1> -<p>The diagram below illustrates the overall architecture of Samza Table API. </p> +<p>The diagram below illustrates the overall architecture of Samza Table API.</p> -<p><img src="/img/latest/learn/documentation/api/table-api-arch.svg" alt="diagram-medium"></p> +<p><img src="/img/latest/learn/documentation/api/table-api-arch.svg" alt="diagram-medium" /></p> <p>Letâs look at a few concepts before diving into the API.</p> -<p><img src="/img/latest/learn/documentation/api/table-class-diagram.svg" alt="diagram-medium"></p> +<p><img src="/img/latest/learn/documentation/api/table-api-class-diagram.png" alt="diagram-medium" /></p> -<p><a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/Table.java";><code>Table</code></a> - +<p>[<code class="language-plaintext highlighter-rouge">Table</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/Table.java) - This interface represents a dataset that can be accessed by key. We support -two types of tables: read-only and read-write. A table can be accessed either +gets, put, updates and deletes. A table can be accessed either synchronously or asynchronously and a request may contain one or more keys. There are three broad categories of tables: local, remote and hybrid.</p> -<p><a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadableTable.java";><code>ReadableTable</code></a> - -Interface that represents a read-only table. It implements Table.</p> +<p>[<code class="language-plaintext highlighter-rouge">AsyncReadWriteUpdateTable</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/AsyncReadWriteUpdateTable.java) - +Interface that represents a read-write-update table with asynchronous operations. Core interface for all table implementations.</p> -<p><a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadWriteTable.java";><code>ReadWriteTable</code></a> - -Interface that represents a read-write table. It implements Table.</p> +<p>[<code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadWriteUpdateTable.java) - +Interface that represents a read-write-update table. It implements AsyncReadWriteUpdateTable. Supports synchronous operations as well.</p> -<p><a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/operators/TableDescriptor.java";><code>TableDescriptor</code></a> - +<p>[<code class="language-plaintext highlighter-rouge">TableDescriptor</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/operators/TableDescriptor.java) - User-facing object that contains metadata that completely describes a table. It may include identifier, serialization, provider, configuration, etc. Example implementations of this interface are</p> <ul> -<li><a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTableDescriptor</code></a> -facilitates access to remotely stored data, </li> -<li><a href="https://github.com/apache/samza/blob/master/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java";><code>InMemoryTableDescriptor</code></a> -describes a table stored in-memory, and </li> -<li><a href="https://github.com/apache/samza/blob/master/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableDescriptor.java";><code>RocksDbTableDescriptor</code></a> -describes a table stored in RocksDB.</li> + <li>[<code class="language-plaintext highlighter-rouge">RemoteTableDescriptor</code>] (https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java) + facilitates access to remotely stored data,</li> + <li>[<code class="language-plaintext highlighter-rouge">InMemoryTableDescriptor</code>] (https://github.com/apache/samza/blob/master/samza-kv-inmemory/src/main/java/org/apache/samza/storage/kv/inmemory/InMemoryTableDescriptor.java) + describes a table stored in-memory, and</li> + <li>[<code class="language-plaintext highlighter-rouge">RocksDbTableDescriptor</code>] (https://github.com/apache/samza/blob/master/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbTableDescriptor.java) + describes a table stored in RocksDB.</li> </ul> <h1 id="table-sync-and-async-api">Table Sync and Async API</h1> <p>Samza Table supports both synchronous and asynchronous API. Below is an example for -the <strong><code>get</code></strong> operation.</p> +the <strong><code class="language-plaintext highlighter-rouge">get</code></strong> operation.</p> -<figure class="highlight"><pre><code class="language-java" data-lang="java"><span></span> <span class="cm">/**</span> -<span class="cm"> * Gets the value associated with the specified {@code key}.</span> -<span class="cm"> *</span> -<span class="cm"> * @param key the key with which the associated value is to be fetched.</span> -<span class="cm"> * @param args additional arguments</span> -<span class="cm"> * @return if found, the value associated with the specified {@code key}; otherwise, {@code null}.</span> -<span class="cm"> * @throws NullPointerException if the specified {@code key} is {@code null}.</span> -<span class="cm"> */</span> - <span class="n">V</span> <span class="nf">get</span><span class="o">(</span><span class="n">K</span> <span class="n">key</span><span class="o">,</span> <span class="n">Object</span> <span class="o">...</span> <span class="n">args</span><span class="o">);</span> - - <span class="cm">/**</span> -<span class="cm"> * Asynchronously gets the value associated with the specified {@code key}.</span> -<span class="cm"> *</span> -<span class="cm"> * @param key the key with which the associated value is to be fetched.</span> -<span class="cm"> * @param args additional arguments</span> -<span class="cm"> * @return completableFuture for the requested value</span> -<span class="cm"> * @throws NullPointerException if the specified {@code key} is {@code null}.</span> -<span class="cm"> */</span> - <span class="n">CompletableFuture</span><span class="o"><</span><span class="n">V</span><span class="o">></span> <span class="nf">getAsync</span><span class="o">(</span><span class="n">K</span> <span class="n">key</span><span class="o">,</span> <span class="n">Object</span> <span class="o">...</span> <span class="n">args</span><span class="o">);</span></code></pre></figure> +<figure class="highlight"><pre><code class="language-java" data-lang="java"> <span class="cm">/** + * Gets the value associated with the specified {@code key}. + * + * @param key the key with which the associated value is to be fetched. + * @param args additional arguments + * @return if found, the value associated with the specified {@code key}; otherwise, {@code null}. + * @throws NullPointerException if the specified {@code key} is {@code null}. + */</span> + <span class="no">V</span> <span class="nf">get</span><span class="o">(</span><span class="no">K</span> <span class="n">key</span><span class="o">,</span> <span class="nc">Object</span> <span class="o">...</span> <span class="n">args</span><span class="o">);</span> + + <span class="cm">/** + * Asynchronously gets the value associated with the specified {@code key}. + * + * @param key the key with which the associated value is to be fetched. + * @param args additional arguments + * @return completableFuture for the requested value + * @throws NullPointerException if the specified {@code key} is {@code null}. + */</span> + <span class="nc">CompletableFuture</span><span class="o"><</span><span class="no">V</span><span class="o">></span> <span class="nf">getAsync</span><span class="o">(</span><span class="no">K</span> <span class="n">key</span><span class="o">,</span> <span class="nc">Object</span> <span class="o">...</span> <span class="n">args</span><span class="o">);</span></code></pre></figure> <h1 id="using-table-with-samza-high-level-api">Using Table with Samza High Level API</h1> <p>The code snippet below illustrates the usage of table in Samza high level API.</p> -<figure class="highlight"><pre><code class="language-java" data-lang="java"><span></span> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaStreamApplication</span> <span class="kd">implements</span> <span class="n">StreamApplication</span> <span class="o">{</span> +<figure class="highlight"><pre><code class="language-java" data-lang="java"> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaStreamApplication</span> <span class="kd">implements</span> <span class="nc">StreamApplication</span> <span class="o">{</span> <span class="mi">2</span> <span class="nd">@Override</span> - <span class="mi">3</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="n">StreamApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> - <span class="mi">4</span> <span class="n">TableDescriptor</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">desc</span> <span class="o">=</span> <span class="k">new</span> <span class="n">InMemoryTableDescriptor</span><span class="o">(</span> - <span class="mi">5</span> <span class="s">"t1"</span><span class="o">,</span> <span class="n">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="n">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="n">ProfileJsonSerde</span><span class="o">()));</span> + <span class="mi">3</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="nc">StreamApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> + <span class="mi">4</span> <span class="nc">TableDescriptor</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">desc</span> <span class="o">=</span> <span class="k">new</span> <span class="nc">InMemoryTableDescriptor</span><span class="o">(</span> + <span class="mi">5</span> <span class="s">"t1"</span><span class="o">,</span> <span class="nc">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="nc">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="nc">ProfileJsonSerde</span><span class="o">()));</span> <span class="mi">6</span> - <span class="mi">7</span> <span class="n">Table</span><span class="o"><</span><span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>></span> <span class="n">table</span> <span class="o">=</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="n">desc</span><span class="o">);</span> + <span class="mi">7</span> <span class="nc">Table</span><span class="o"><</span><span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>></span> <span class="n">table</span> <span class="o">=</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="n">desc</span><span class="o">);</span> <span class="mi">8</span> - <span class="mi">9</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getInputStream</span><span class="o">(</span><span class="s">"PageView"</span><span class="o">,</span> <span class="k">new</span> <span class="n">NoOpSerde</span><span class="o"><</span><span class="n">PageView</span><span class="o">>())</span> -<span class="mi">10</span> <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="n">MyMapFunc</span><span class="o">())</span> -<span class="mi">11</span> <span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">table</span><span class="o">,</span> <span class="k">new</span> <span class="n">MyJoinFunc</span><span class="o">())</span> + <span class="mi">9</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getInputStream</span><span class="o">(</span><span class="s">"PageView"</span><span class="o">,</span> <span class="k">new</span> <span class="nc">NoOpSerde</span><span class="o"><</span><span class="nc">PageView</span><span class="o">>())</span> +<span class="mi">10</span> <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="nc">MyMapFunc</span><span class="o">())</span> +<span class="mi">11</span> <span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">table</span><span class="o">,</span> <span class="k">new</span> <span class="nc">MyJoinFunc</span><span class="o">())</span> <span class="mi">12</span> <span class="o">.</span><span class="na">sendTo</span><span class="o">(</span><span class="n">anotherTable</span><span class="o">);</span> <span class="mi">13</span> <span class="o">}</span> <span class="mi">14</span> <span class="o">}</span> <span class="mi">15</span> -<span class="mi">16</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyMapFunc</span> <span class="kd">implements</span> <span class="n">MapFunction</span><span class="o"><</span><span class="n">PageView</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">>></span> <span class="o">{</span> -<span class="mi">17</span> <span class="kd">private</span> <span class="n">ReadableTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> +<span class="mi">16</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyMapFunc</span> <span class="kd">implements</span> <span class="nc">MapFunction</span><span class="o"><</span><span class="nc">PageView</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">>></span> <span class="o">{</span> +<span class="mi">17</span> <span class="kd">private</span> <span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> <span class="mi">18</span> <span class="mi">19</span> <span class="nd">@Override</span> -<span class="mi">20</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="n">Config</span> <span class="n">config</span><span class="o">,</span> <span class="n">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">21</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="n">ReadableTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> +<span class="mi">20</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="nc">Config</span> <span class="n">config</span><span class="o">,</span> <span class="nc">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">21</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> <span class="mi">22</span> <span class="o">}</span> <span class="mi">23</span> <span class="mi">24</span> <span class="nd">@Override</span> -<span class="mi">25</span> <span class="kd">public</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">></span> <span class="nf">apply</span><span class="o">(</span><span class="n">PageView</span> <span class="n">message</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">26</span> <span class="k">return</span> <span class="k">new</span> <span class="n">KV</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">message</span><span class="o">.</span><span class="na">getId</span><span class="o">(),</span> <span class="n">message</span><span class="o">);</span> +<span class="mi">25</span> <span class="kd">public</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">></span> <span class="nf">apply</span><span class="o">(</span><span class="nc">PageView</span> <span class="n">message</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">26</span> <span class="k">return</span> <span class="k">new</span> <span class="no">KV</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">message</span><span class="o">.</span><span class="na">getId</span><span class="o">(),</span> <span class="n">message</span><span class="o">);</span> <span class="mi">27</span> <span class="o">}</span> <span class="mi">28</span> <span class="o">}</span> <span class="mi">29</span> -<span class="mi">30</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyJoinFunc</span> <span class="kd">implements</span> <span class="n">StreamTableJoinFunction</span> -<span class="mi">31</span> <span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">>,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>,</span> <span class="n">EnrichedPageView</span><span class="o">></span> <span class="o">{</span> +<span class="mi">30</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyJoinFunc</span> <span class="kd">implements</span> <span class="nc">StreamTableJoinFunction</span> +<span class="mi">31</span> <span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">>,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>,</span> <span class="nc">EnrichedPageView</span><span class="o">></span> <span class="o">{</span> <span class="mi">32</span> <span class="mi">33</span> <span class="nd">@Override</span> -<span class="mi">34</span> <span class="kd">public</span> <span class="n">EnrichedPageView</span> <span class="nf">apply</span><span class="o">(</span><span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">></span> <span class="n">m</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">r</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">34</span> <span class="kd">public</span> <span class="nc">EnrichedPageView</span> <span class="nf">apply</span><span class="o">(</span><span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">></span> <span class="n">m</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">r</span><span class="o">)</span> <span class="o">{</span> <span class="mi">35</span> <span class="n">counterPerJoinFn</span><span class="o">.</span><span class="na">get</span><span class="o">(</span><span class="k">this</span><span class="o">.</span><span class="na">currentSeqNo</span><span class="o">).</span><span class="na">incrementAndGet</span><span class="o">();</span> -<span class="mi">36</span> <span class="k">return</span> <span class="n">r</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="kc">null</span> <span class="o">:</span> <span class="k">new</span> <span class="n">EnrichedPageView</span><span class="o">(</span> +<span class="mi">36</span> <span class="k">return</span> <span class="n">r</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="kc">null</span> <span class="o">:</span> <span class="k">new</span> <span class="nc">EnrichedPageView</span><span class="o">(</span> <span class="mi">37</span> <span class="n">m</span><span class="o">.</span><span class="na">getValue</span><span class="o">().</span><span class="na">getPageKey</span><span class="o">(),</span> <span class="n">m</span><span class="o">.</span><span class="na">getKey</span><span class="o">(),</span> <span class="n">r</span><span class="o">.</span><span class="na">getValue</span><span class="o">().</span><span class="na">getCompany</span><span class="o">());</span> <span class="mi">38</span> <span class="o">}</span> <span class="mi">39</span> <span class="o">}</span></code></pre></figure> @@ -784,107 +798,109 @@ the <strong><code>get</code></strong> op join with a table and finally write the output to another table.</p> <ol> -<li>Line 4-5: A <code>TableDescriptor</code> for an in-memory table is created, and then -the serde is set. </li> -<li>Line 7: A table object is created from the <code>TableDescriptor</code>; internally, -the <code>TableDescriptor</code> is converted to a <code>TableSpec</code>, and registered with the -<code>TaskApplicationDescriptor</code>. The table object has a reference to the <code>TableSpec</code>.</li> -<li>Line 9: creates an <code>InputStream</code></li> -<li>Line 10: applies a map operator defined in line 16-28, all table instances -can be accessed from the task context in <code>Task.init()</code>. In this example, + <li>Line 4-5: A <code class="language-plaintext highlighter-rouge">TableDescriptor</code> for an in-memory table is created, and then +the serde is set.</li> + <li>Line 7: A table object is created from the <code class="language-plaintext highlighter-rouge">TableDescriptor</code>; internally, +the <code class="language-plaintext highlighter-rouge">TableDescriptor</code> is converted to a <code class="language-plaintext highlighter-rouge">TableSpec</code>, and registered with the +<code class="language-plaintext highlighter-rouge">TaskApplicationDescriptor</code>. The table object has a reference to the <code class="language-plaintext highlighter-rouge">TableSpec</code>.</li> + <li>Line 9: creates an <code class="language-plaintext highlighter-rouge">InputStream</code></li> + <li>Line 10: applies a map operator defined in line 16-28, all table instances +can be accessed from the task context in <code class="language-plaintext highlighter-rouge">Task.init()</code>. In this example, it is stored in a local variable.</li> -<li>Line 11: joins the mapped stream with the table using the supplied join + <li>Line 11: joins the mapped stream with the table using the supplied join function defined in lines 30-39.</li> -<li>Line 12: writes the join result stream to another table</li> + <li>Line 12: writes the join result stream to another table</li> </ol> <h1 id="using-table-with-samza-high-level-api-using-side-inputs">Using Table with Samza High Level API using Side Inputs</h1> <p>The code snippet below illustrates the usage of table in Samza high level API using side inputs.</p> -<figure class="highlight"><pre><code class="language-java" data-lang="java"><span></span> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaStreamApplication</span> <span class="kd">implements</span> <span class="n">StreamApplication</span> <span class="o">{</span> +<figure class="highlight"><pre><code class="language-java" data-lang="java"> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaStreamApplication</span> <span class="kd">implements</span> <span class="nc">StreamApplication</span> <span class="o">{</span> <span class="mi">2</span> <span class="nd">@Override</span> - <span class="mi">3</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="n">StreamApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> - <span class="mi">4</span> <span class="n">TableDescriptor</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">desc</span> <span class="o">=</span> <span class="k">new</span> <span class="n">InMemoryTableDescriptor</span><span class="o">(</span> - <span class="mi">5</span> <span class="s">"t1"</span><span class="o">,</span> <span class="n">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="n">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="n">ProfileJsonSerde</span><span class="o">()))</span> - <span class="mi">6</span> <span class="o">.</span><span class="na">withSideInputs</span><span class="o">(</span><span class="n">ImmutableList</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">PROFILE_STREAM</span><span class="o">))</span> + <span class="mi">3</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="nc">StreamApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> + <span class="mi">4</span> <span class="nc">TableDescriptor</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">desc</span> <span class="o">=</span> <span class="k">new</span> <span class="nc">InMemoryTableDescriptor</span><span class="o">(</span> + <span class="mi">5</span> <span class="s">"t1"</span><span class="o">,</span> <span class="nc">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="nc">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="nc">ProfileJsonSerde</span><span class="o">()))</span> + <span class="mi">6</span> <span class="o">.</span><span class="na">withSideInputs</span><span class="o">(</span><span class="nc">ImmutableList</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="no">PROFILE_STREAM</span><span class="o">))</span> <span class="mi">7</span> <span class="o">.</span><span class="na">withSideInputsProcessor</span><span class="o">((</span><span class="n">msg</span><span class="o">,</span> <span class="n">store</span><span class="o">)</span> <span class="o">-></span> <span class="o">{</span> - <span class="mi">8</span> <span class="n">Profile</span> <span class="n">profile</span> <span class="o">=</span> <span class="o">(</span><span class="n">Profile</span><span class="o">)</span> <span class="n">msg</span><span class="o">.</span><span class="na">getMessage</span><span class="o">();</span> + <span class="mi">8</span> <span class="nc">Profile</span> <span class="n">profile</span> <span class="o">=</span> <span class="o">(</span><span class="nc">Profile</span><span class="o">)</span> <span class="n">msg</span><span class="o">.</span><span class="na">getMessage</span><span class="o">();</span> <span class="mi">9</span> <span class="kt">int</span> <span class="n">key</span> <span class="o">=</span> <span class="n">profile</span><span class="o">.</span><span class="na">getMemberId</span><span class="o">();</span> -<span class="mi">10</span> <span class="k">return</span> <span class="n">ImmutableList</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="n">Entry</span><span class="o"><>(</span><span class="n">key</span><span class="o">,</span> <span class="n">profile</span><span class="o">));</span> +<span class="mi">10</span> <span class="k">return</span> <span class="nc">ImmutableList</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="nc">Entry</span><span class="o"><>(</span><span class="n">key</span><span class="o">,</span> <span class="n">profile</span><span class="o">));</span> <span class="mi">11</span> <span class="o">});</span> <span class="mi">12</span> -<span class="mi">13</span> <span class="n">Table</span><span class="o"><</span><span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>></span> <span class="n">table</span> <span class="o">=</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="n">desc</span><span class="o">);</span> +<span class="mi">13</span> <span class="nc">Table</span><span class="o"><</span><span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>></span> <span class="n">table</span> <span class="o">=</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="n">desc</span><span class="o">);</span> <span class="mi">14</span> -<span class="mi">15</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getInputStream</span><span class="o">(</span><span class="s">"PageView"</span><span class="o">,</span> <span class="k">new</span> <span class="n">NoOpSerde</span><span class="o"><</span><span class="n">PageView</span><span class="o">>())</span> -<span class="mi">16</span> <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="n">MyMapFunc</span><span class="o">())</span> -<span class="mi">17</span> <span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">table</span><span class="o">,</span> <span class="k">new</span> <span class="n">MyJoinFunc</span><span class="o">())</span> +<span class="mi">15</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">getInputStream</span><span class="o">(</span><span class="s">"PageView"</span><span class="o">,</span> <span class="k">new</span> <span class="nc">NoOpSerde</span><span class="o"><</span><span class="nc">PageView</span><span class="o">>())</span> +<span class="mi">16</span> <span class="o">.</span><span class="na">map</span><span class="o">(</span><span class="k">new</span> <span class="nc">MyMapFunc</span><span class="o">())</span> +<span class="mi">17</span> <span class="o">.</span><span class="na">join</span><span class="o">(</span><span class="n">table</span><span class="o">,</span> <span class="k">new</span> <span class="nc">MyJoinFunc</span><span class="o">())</span> <span class="mi">18</span> <span class="o">.</span><span class="na">sendTo</span><span class="o">(</span><span class="n">anotherTable</span><span class="o">);</span> <span class="mi">19</span> <span class="o">}</span> <span class="mi">21</span> <span class="o">}</span> <span class="mi">22</span> -<span class="mi">23</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyMapFunc</span> <span class="kd">implements</span> <span class="n">MapFunction</span><span class="o"><</span><span class="n">PageView</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">>></span> <span class="o">{</span> -<span class="mi">24</span> <span class="kd">private</span> <span class="n">ReadableTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> +<span class="mi">23</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyMapFunc</span> <span class="kd">implements</span> <span class="nc">MapFunction</span><span class="o"><</span><span class="nc">PageView</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">>></span> <span class="o">{</span> +<span class="mi">24</span> <span class="kd">private</span> <span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> <span class="mi">25</span> <span class="mi">26</span> <span class="nd">@Override</span> -<span class="mi">27</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="n">Config</span> <span class="n">config</span><span class="o">,</span> <span class="n">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">28</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="n">ReadableTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> +<span class="mi">27</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="nc">Config</span> <span class="n">config</span><span class="o">,</span> <span class="nc">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">28</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> <span class="mi">29</span> <span class="o">}</span> <span class="mi">30</span> <span class="mi">31</span> <span class="nd">@Override</span> -<span class="mi">32</span> <span class="kd">public</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">></span> <span class="nf">apply</span><span class="o">(</span><span class="n">PageView</span> <span class="n">message</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">33</span> <span class="k">return</span> <span class="k">new</span> <span class="n">KV</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">message</span><span class="o">.</span><span class="na">getId</span><span class="o">(),</span> <span class="n">message</span><span class="o">);</span> +<span class="mi">32</span> <span class="kd">public</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">></span> <span class="nf">apply</span><span class="o">(</span><span class="nc">PageView</span> <span class="n">message</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">33</span> <span class="k">return</span> <span class="k">new</span> <span class="no">KV</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="n">message</span><span class="o">.</span><span class="na">getId</span><span class="o">(),</span> <span class="n">message</span><span class="o">);</span> <span class="mi">34</span> <span class="o">}</span> <span class="mi">35</span> <span class="o">}</span> <span class="mi">36</span> -<span class="mi">37</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyJoinFunc</span> <span class="kd">implements</span> <span class="n">StreamTableJoinFunction</span> -<span class="mi">38</span> <span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">>,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>,</span> <span class="n">EnrichedPageView</span><span class="o">></span> <span class="o">{</span> +<span class="mi">37</span> <span class="kd">static</span> <span class="kd">class</span> <span class="nc">MyJoinFunc</span> <span class="kd">implements</span> <span class="nc">StreamTableJoinFunction</span> +<span class="mi">38</span> <span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">>,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>,</span> <span class="nc">EnrichedPageView</span><span class="o">></span> <span class="o">{</span> <span class="mi">39</span> <span class="mi">40</span> <span class="nd">@Override</span> -<span class="mi">41</span> <span class="kd">public</span> <span class="n">EnrichedPageView</span> <span class="nf">apply</span><span class="o">(</span><span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">PageView</span><span class="o">></span> <span class="n">m</span><span class="o">,</span> <span class="n">KV</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">r</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">41</span> <span class="kd">public</span> <span class="nc">EnrichedPageView</span> <span class="nf">apply</span><span class="o">(</span><span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">PageView</span><span class="o">></span> <span class="n">m</span><span class="o">,</span> <span class="no">KV</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">r</span><span class="o">)</span> <span class="o">{</span> <span class="mi">42</span> <span class="n">counterPerJoinFn</span><span class="o">.</span><span class="na">get</span><span class="o">(</span><span class="k">this</span><span class="o">.</span><span class="na">currentSeqNo</span><span class="o">).</span><span class="na">incrementAndGet</span><span class="o">();</span> -<span class="mi">43</span> <span class="k">return</span> <span class="n">r</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="kc">null</span> <span class="o">:</span> <span class="k">new</span> <span class="n">EnrichedPageView</span><span class="o">(</span> +<span class="mi">43</span> <span class="k">return</span> <span class="n">r</span> <span class="o">==</span> <span class="kc">null</span> <span class="o">?</span> <span class="kc">null</span> <span class="o">:</span> <span class="k">new</span> <span class="nc">EnrichedPageView</span><span class="o">(</span> <span class="mi">44</span> <span class="n">m</span><span class="o">.</span><span class="na">getValue</span><span class="o">().</span><span class="na">getPageKey</span><span class="o">(),</span> <span class="n">m</span><span class="o">.</span><span class="na">getKey</span><span class="o">(),</span> <span class="n">r</span><span class="o">.</span><span class="na">getValue</span><span class="o">().</span><span class="na">getCompany</span><span class="o">());</span> <span class="mi">45</span> <span class="o">}</span> <span class="mi">46</span> <span class="o">}</span></code></pre></figure> -<p>The code above uses side inputs to populate the profile table. -1. Line 6: Denotes the source stream for the profile table -2. Line 7-11: Provides an implementation of <code>SideInputsProcessor</code> that reads from profile stream - and populates the table. -3. Line 17: Incoming page views are joined against the profile table.</p> +<p>The code above uses side inputs to populate the profile table.</p> +<ol> + <li>Line 6: Denotes the source stream for the profile table</li> + <li>Line 7-11: Provides an implementation of <code class="language-plaintext highlighter-rouge">SideInputsProcessor</code> that reads from profile stream + and populates the table.</li> + <li>Line 17: Incoming page views are joined against the profile table.</li> +</ol> <h1 id="using-table-with-samza-low-level-api">Using Table with Samza Low Level API</h1> <p>The code snippet below illustrates the usage of table in Samza Low Level Task API.</p> -<figure class="highlight"><pre><code class="language-java" data-lang="java"><span></span> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaTaskApplication</span> <span class="kd">implements</span> <span class="n">TaskApplication</span> <span class="o">{</span> +<figure class="highlight"><pre><code class="language-java" data-lang="java"> <span class="mi">1</span> <span class="kd">class</span> <span class="nc">SamzaTaskApplication</span> <span class="kd">implements</span> <span class="nc">TaskApplication</span> <span class="o">{</span> <span class="mi">2</span> <span class="mi">3</span> <span class="nd">@Override</span> - <span class="mi">4</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="n">TaskApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> - <span class="mi">5</span> <span class="n">DelegatingSystemDescriptor</span> <span class="n">ksd</span> <span class="o">=</span> <span class="k">new</span> <span class="n">DelegatingSystemDescriptor</span><span class="o">(</span><span class="s">"mySystem"</span><span class="o">);</span> + <span class="mi">4</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">describe</span><span class="o">(</span><span class="nc">TaskApplicationDescriptor</span> <span class="n">appDesc</span><span class="o">)</span> <span class="o">{</span> + <span class="mi">5</span> <span class="nc">DelegatingSystemDescriptor</span> <span class="n">ksd</span> <span class="o">=</span> <span class="k">new</span> <span class="nc">DelegatingSystemDescriptor</span><span class="o">(</span><span class="s">"mySystem"</span><span class="o">);</span> <span class="mi">6</span> - <span class="mi">7</span> <span class="n">TableDescriptor</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">tableDesc</span> <span class="o">=</span> <span class="k">new</span> <span class="n">InMemoryTableDescriptor</span><span class="o">(</span> - <span class="mi">8</span> <span class="s">"t1"</span><span class="o">,</span> <span class="n">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="n">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="n">ProfileJsonSerde</span><span class="o">()));</span> - <span class="mi">9</span> <span class="n">GenericInputDescriptor</span><span class="o"><</span><span class="n">Profile</span><span class="o">></span> <span class="n">profileISD</span> <span class="o">=</span> <span class="n">ksd</span><span class="o">.</span><span class="na">getInputDescriptor</span><span class="o">(</span><span class="s">"Profile"</span><span class="o">,</span> <span class="k">new</span> <span class="n">NoOpSerde</span><span class="o"><>());</span> + <span class="mi">7</span> <span class="nc">TableDescriptor</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">tableDesc</span> <span class="o">=</span> <span class="k">new</span> <span class="nc">InMemoryTableDescriptor</span><span class="o">(</span> + <span class="mi">8</span> <span class="s">"t1"</span><span class="o">,</span> <span class="nc">KVSerde</span><span class="o">.</span><span class="na">of</span><span class="o">(</span><span class="k">new</span> <span class="nc">IntegerSerde</span><span class="o">(),</span> <span class="k">new</span> <span class="nc">ProfileJsonSerde</span><span class="o">()));</span> + <span class="mi">9</span> <span class="nc">GenericInputDescriptor</span><span class="o"><</span><span class="nc">Profile</span><span class="o">></span> <span class="n">profileISD</span> <span class="o">=</span> <span class="n">ksd</span><span class="o">.</span><span class="na">getInputDescriptor</span><span class="o">(</span><span class="s">"Profile"</span><span class="o">,</span> <span class="k">new</span> <span class="nc">NoOpSerde</span><span class="o"><>());</span> <span class="mi">10</span> <span class="mi">11</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">addTable</span><span class="o">(</span><span class="n">tableDesc</span><span class="o">);</span> <span class="mi">12</span> <span class="n">appDesc</span><span class="o">.</span><span class="na">addInputStream</span><span class="o">(</span><span class="n">profileISD</span><span class="o">);</span> <span class="mi">13</span> <span class="o">}</span> <span class="mi">14</span> <span class="mi">15</span> -<span class="mi">16</span> <span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyStreamTask</span> <span class="kd">implements</span> <span class="n">StreamTask</span><span class="o">,</span> <span class="n">InitableTask</span> <span class="o">{</span> -<span class="mi">17</span> <span class="kd">private</span> <span class="n">ReadWriteTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> +<span class="mi">16</span> <span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyStreamTask</span> <span class="kd">implements</span> <span class="nc">StreamTask</span><span class="o">,</span> <span class="nc">InitableTask</span> <span class="o">{</span> +<span class="mi">17</span> <span class="kd">private</span> <span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">></span> <span class="n">profileTable</span><span class="o">;</span> <span class="mi">18</span> <span class="mi">19</span> <span class="nd">@Override</span> -<span class="mi">20</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="n">Config</span> <span class="n">config</span><span class="o">,</span> <span class="n">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">21</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="n">ReadWriteTable</span><span class="o"><</span><span class="n">Integer</span><span class="o">,</span> <span class="n">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> +<span class="mi">20</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">init</span><span class="o">(</span><span class="nc">Config</span> <span class="n">config</span><span class="o">,</span> <span class="nc">TaskContext</span> <span class="n">context</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">21</span> <span class="n">profileTable</span> <span class="o">=</span> <span class="o">(</span><span class="nc">ReadWriteUpdateTable</span><span class="o"><</span><span class="nc">Integer</span><span class="o">,</span> <span class="nc">Profile</span><span class="o">>)</span> <span class="n">context</span><span class="o">.</span><span class="na">getTable</span><span class="o">(</span><span class="s">"t1"</span><span class="o">);</span> <span class="mi">22</span> <span class="o">}</span> <span class="mi">23</span> <span class="mi">24</span> <span class="nd">@Override</span> -<span class="mi">25</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="o">(</span><span class="n">IncomingMessageEnvelope</span> <span class="n">envelope</span><span class="o">,</span> <span class="n">MessageCollector</span> <span class="n">collector</span><span class="o">,</span> <span class="n">TaskCoordinator</span> <span class="n">coordinator</span><span class="o">)</span> <span class="o">{</span> -<span class="mi">26</span> <span class="n">String</span> <span class="n">key</span> <span class="o">=</span> <span class="o">(</span><span class="n">String</span><span class="o">)</span><span class="n">message</span><span class="o">.</span><span class="na">getKey</span><span class="o">();</span> -<span class="mi">27</span> <span class="n">Profile</span> <span class="n">profile</span> <span class="o">=</span> <span class="o">(</span><span class="n">Profile</span><span class="o">)</span><span class="n">message</span><span class="o">.</span><span class="na">getMessage</span><span class="o">();</span> +<span class="mi">25</span> <span class="kd">public</span> <span class="kt">void</span> <span class="nf">process</span><span class="o">(</span><span class="nc">IncomingMessageEnvelope</span> <span class="n">envelope</span><span class="o">,</span> <span class="nc">MessageCollector</span> <span class="n">collector</span><span class="o">,</span> <span class="nc">TaskCoordinator</span> <span class="n">coordinator</span><span class="o">)</span> <span class="o">{</span> +<span class="mi">26</span> <span class="nc">String</span> <span class="n">key</span> <span class="o">=</span> <span class="o">(</span><span class="nc">String</span><span class="o">)</span><span class="n">message</span><span class="o">.</span><span class="na">getKey</span><span class="o">();</span> +<span class="mi">27</span> <span class="nc">Profile</span> <span class="n">profile</span> <span class="o">=</span> <span class="o">(</span><span class="nc">Profile</span><span class="o">)</span><span class="n">message</span><span class="o">.</span><span class="na">getMessage</span><span class="o">();</span> <span class="mi">28</span> <span class="n">profileTable</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">key</span><span class="o">,</span> <span class="n">profile</span><span class="o">);</span> <span class="mi">29</span> <span class="o">}</span> <span class="mi">30</span> <span class="o">}</span></code></pre></figure> @@ -893,247 +909,272 @@ function defined in lines 30-39.</li> join with a table and finally write the output to another table.</p> <ol> -<li>Line 7-8: A <code>TableDescriptor</code> for an in-memory table is created with tableId ât1â, - and then the serde is set. </li> -<li>Line 9: creates an <code>InputStreamDescriptor</code>.</li> -<li>Line 11-12: adds <code>TableDescriptor</code> and <code>InputStreamDescriptor</code> to the <code>TaskApplicationDescriptor</code>.</li> -<li>Line 16: <code>InitiableStreamTask</code> is implemented.</li> -<li>Line 21: A reference to table ât1â is obtained in the <code>Task.init()</code> method.</li> -<li>Line 26-28: gets the profile key and record from the incoming stream and writes to the table. </li> + <li>Line 7-8: A <code class="language-plaintext highlighter-rouge">TableDescriptor</code> for an in-memory table is created with tableId ât1â, + and then the serde is set.</li> + <li>Line 9: creates an <code class="language-plaintext highlighter-rouge">InputStreamDescriptor</code>.</li> + <li>Line 11-12: adds <code class="language-plaintext highlighter-rouge">TableDescriptor</code> and <code class="language-plaintext highlighter-rouge">InputStreamDescriptor</code> to the <code class="language-plaintext highlighter-rouge">TaskApplicationDescriptor</code>.</li> + <li>Line 16: <code class="language-plaintext highlighter-rouge">InitiableStreamTask</code> is implemented.</li> + <li>Line 21: A reference to table ât1â is obtained in the <code class="language-plaintext highlighter-rouge">Task.init()</code> method.</li> + <li>Line 26-28: gets the profile key and record from the incoming stream and writes to the table.</li> </ol> -<p><a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadableTable.java";><code>ReadableTable</code></a> -or <a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadWriteTable.java";><code>ReadWriteTable</code></a> +<p>[<code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadWriteUpdateTable.java) +or [<code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/table/ReadWriteUpdateTable.java) can be used in the -<a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/task/StreamTask.java#L49";><code>StreamTask.process()</code></a> +[<code class="language-plaintext highlighter-rouge">StreamTask.process()</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/task/StreamTask.java#L49) method on the table reference obtained in the -<a href="https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/task/InitableTask.java#L35";><code>InitableTask.init()</code></a> +[<code class="language-plaintext highlighter-rouge">InitableTask.init()</code>] (https://github.com/apache/samza/blob/master/samza-api/src/main/java/org/apache/samza/task/InitableTask.java#L35) method.</p> <h1 id="table-metrics">Table Metrics</h1> <p>The table below summarizes table metrics:</p> -<table><thead> -<tr> -<th>Metrics</th> -<th>Class</th> -<th>Description</th> -</tr> -</thead><tbody> -<tr> -<td><code>num-batches</code></td> -<td><code>AsyncBatchingTable</code></td> -<td>Number of batch operations</td> -</tr> -<tr> -<td><code>batch-ns</code></td> -<td><code>AsyncBatchingTable</code></td> -<td>Time interval between opening and closing a batch</td> -</tr> -<tr> -<td><code>get-ns</code></td> -<td><code>ReadableTable</code></td> -<td>Average latency of <code>get/getAsync()</code> operations</td> -</tr> -<tr> -<td><code>getAll-ns</code></td> -<td><code>ReadableTable</code></td> -<td>Average latency of <code>getAll/getAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-gets</code></td> -<td><code>ReadableTable</code></td> -<td>Count of <code>get/getAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-getAlls</code></td> -<td><code>ReadableTable</code></td> -<td>Count of <code>getAll/getAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-missed-lookups</code></td> -<td><code>ReadableTable</code></td> -<td>Count of missed get/getAll() operations</td> -</tr> -<tr> -<td><code>read-ns</code></td> -<td><code>ReadableTable</code></td> -<td>Average latency of <code>readAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-reads</code></td> -<td><code>ReadableTable</code></td> -<td>Count of <code>readAsync()</code> operations</td> -</tr> -<tr> -<td><code>put-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of <code>put/putAsync()</code> operations</td> -</tr> -<tr> -<td><code>putAll-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of <code>putAll/putAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-puts</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of <code>put/putAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-putAlls</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of <code>putAll/putAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>delete-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of <code>delete/deleteAsync()</code> operations</td> -</tr> -<tr> -<td><code>deleteAll-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of <code>deleteAll/deleteAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>delete-num</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of <code>delete/deleteAsync()</code> operations</td> -</tr> -<tr> -<td><code>deleteAll-num</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of <code>deleteAll/deleteAllAsync()</code> operations</td> -</tr> -<tr> -<td><code>num-writes</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of <code>writeAsync()</code> operations</td> -</tr> -<tr> -<td><code>write-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of <code>writeAsync()</code> operations</td> -</tr> -<tr> -<td><code>flush-ns</code></td> -<td><code>ReadWriteTable</code></td> -<td>Average latency of flush operations</td> -</tr> -<tr> -<td><code>flush-num</code></td> -<td><code>ReadWriteTable</code></td> -<td>Count of flush operations</td> -</tr> -<tr> -<td><code>hit-rate</code></td> -<td><code>CachingTable</code></td> -<td>Cache hit rate (%)</td> -</tr> -<tr> -<td><code>miss-rate</code></td> -<td><code>CachingTable</code></td> -<td>Cache miss rate (%)</td> -</tr> -<tr> -<td><code>req-count</code></td> -<td><code>CachingTable</code></td> -<td>Count of requests</td> -</tr> -<tr> -<td><code>retry-count</code></td> -<td><code>TableRetryPolicy</code></td> -<td>Count of retries executed (excluding the first attempt)</td> -</tr> -<tr> -<td><code>success-count</code></td> -<td><code>TableRetryPolicy</code></td> -<td>Count of successes at first attempt</td> -</tr> -<tr> -<td><code>perm-failure-count</code></td> -<td><code>TableRetryPolicy</code></td> -<td>Count of operations that failed permanently and exhausted all retries</td> -</tr> -<tr> -<td><code>retry-timer</code></td> -<td><code>TableRetryPolicy</code></td> -<td>Total time spent in each IO; this is updated only when at least one retry has been attempted.</td> -</tr> -</tbody></table> +<table> + <thead> + <tr> + <th>Metrics</th> + <th>Class</th> + <th>Description</th> + </tr> + </thead> + <tbody> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-batches</code></td> + <td><code class="language-plaintext highlighter-rouge">AsyncBatchingTable</code></td> + <td>Number of batch operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">batch-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">AsyncBatchingTable</code></td> + <td>Time interval between opening and closing a batch</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">get-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">get/getAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">getAll-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">getAll/getAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-gets</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">get/getAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-getAlls</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">getAll/getAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-missed-lookups</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of missed get/getAll() operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">read-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">readAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-reads</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">readAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">put-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">put/putAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">putAll-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">putAll/putAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-puts</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">put/putAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-putAlls</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">putAll/putAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">update-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">update/updateAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">updateAll-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">updateAll/updateAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-updates</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">update/updateAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-updateAlls</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">updatesAll/updateAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">delete-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">delete/deleteAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">deleteAll-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">deleteAll/deleteAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">delete-num</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">delete/deleteAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">deleteAll-num</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">deleteAll/deleteAllAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">num-writes</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of <code class="language-plaintext highlighter-rouge">writeAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">write-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of <code class="language-plaintext highlighter-rouge">writeAsync()</code> operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">flush-ns</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Average latency of flush operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">flush-num</code></td> + <td><code class="language-plaintext highlighter-rouge">ReadWriteUpdateTable</code></td> + <td>Count of flush operations</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">hit-rate</code></td> + <td><code class="language-plaintext highlighter-rouge">CachingTable</code></td> + <td>Cache hit rate (%)</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">miss-rate</code></td> + <td><code class="language-plaintext highlighter-rouge">CachingTable</code></td> + <td>Cache miss rate (%)</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">req-count</code></td> + <td><code class="language-plaintext highlighter-rouge">CachingTable</code></td> + <td>Count of requests</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">retry-count</code></td> + <td><code class="language-plaintext highlighter-rouge">TableRetryPolicy</code></td> + <td>Count of retries executed (excluding the first attempt)</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">success-count</code></td> + <td><code class="language-plaintext highlighter-rouge">TableRetryPolicy</code></td> + <td>Count of successes at first attempt</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">perm-failure-count</code></td> + <td><code class="language-plaintext highlighter-rouge">TableRetryPolicy</code></td> + <td>Count of operations that failed permanently and exhausted all retries</td> + </tr> + <tr> + <td><code class="language-plaintext highlighter-rouge">retry-timer</code></td> + <td><code class="language-plaintext highlighter-rouge">TableRetryPolicy</code></td> + <td>Total time spent in each IO; this is updated only when at least one retry has been attempted.</td> + </tr> + </tbody> +</table> <h1 id="table-types">Table Types</h1> <h2 id="remote-table">Remote Table</h2> -<p><a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTable</code></a> +<p><a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code class="language-plaintext highlighter-rouge">RemoteTable</code></a> provides a unified abstraction for Samza applications to access any remote data 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, +access new types of stores by writing pluggable I/O âRead/Writeâ functions, implementations of -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/TableReadFunction.java";><code>TableReadFunction</code></a> and -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/TableWriteFunction.java";><code>TableWriteFunction</code></a> +<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/TableReadFunction.java";><code class="language-plaintext highlighter-rouge">TableReadFunction</code></a> and +<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/TableWriteFunction.java";><code class="language-plaintext highlighter-rouge">TableWriteFunction</code></a> interfaces. Remote Table also provides common functionality, eg. rate limiting (built-in) and caching (hybrid).</p> <p>The async APIs in Remote Table are recommended over the sync versions for higher throughput. They can be used with Samza with Low Level Task API to achieve the maximum -throughput. </p> +throughput.</p> <p>Remote Tables are represented by class -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadableTable.java";><code>RemoteReadableTable</code></a> and -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadWriteTable.java";><code>RemoteReadWriteTable</code></a>. +<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadWriteUpdateTable.java";><code class="language-plaintext highlighter-rouge">RemoteReadWriteUpdateTable</code></a> and +<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteReadWriteUpdateTable.java";><code class="language-plaintext highlighter-rouge">RemoteReadWriteUpdateTable</code></a>. All configuration options of a Remote Table can be found in the -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTableDescriptor</code></a> class. </p> +<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code class="language-plaintext highlighter-rouge">RemoteTableDescriptor</code></a> class.</p> <p>Couchbase is supported as remote table. See -<a href="https://github.com/apache/samza/blob/master/samza-kv-couchbase/src/main/java/org/apache/samza/table/remote/couchbase/CouchbaseTableReadFunction.java";><code>CouchbaseTableReadFunction</code></a> and -<a href="https://github.com/apache/samza/blob/master/samza-kv-couchbase/src/main/java/org/apache/samza/table/remote/couchbase/CouchbaseTableWriteFunction.java";><code>CouchbaseTableWriteFunction</code></a>.</p> +<a href="https://github.com/apache/samza/blob/master/samza-kv-couchbase/src/main/java/org/apache/samza/table/remote/couchbase/CouchbaseTableReadFunction.java";><code class="language-plaintext highlighter-rouge">CouchbaseTableReadFunction</code></a> and +<a href="https://github.com/apache/samza/blob/master/samza-kv-couchbase/src/main/java/org/apache/samza/table/remote/couchbase/CouchbaseTableWriteFunction.java";><code class="language-plaintext highlighter-rouge">CouchbaseTableWriteFunction</code></a>.</p> <h3 id="batching">Batching</h3> <p>Remote Table has built-in client-side batching support for its async executions. This is useful when a remote data store supports batch processing and is not sophisticated enough -to handle heavy inbound requests. </p> +to handle heavy inbound requests.</p> <h4 id="configuration">Configuration</h4> -<p>Batching can be enabled with <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTableDescriptor</code></a> -by providing a <a href="https://github.com/apache/samza/samza-api/src/main/java/org/apache/samza/table/batching/BatchProvider.java";><code>BatchProvider</code></a> +<p>Batching can be enabled with <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code class="language-plaintext highlighter-rouge">RemoteTableDescriptor</code></a> +by providing a <a href="https://github.com/apache/samza/samza-api/src/main/java/org/apache/samza/table/batching/BatchProvider.java";><code class="language-plaintext highlighter-rouge">BatchProvider</code></a> The user can choose:</p> <ol> -<li>A <a href="https://github.com/apache/samza/samza-core/src/main/java/org/apache/samza/table/batching/CompactBatchProvider.java";><code>CompactBatchProvider</code></a> which provides a batch such that + <li>A <a href="https://github.com/apache/samza/samza-core/src/main/java/org/apache/samza/table/batching/CompactBatchProvider.java";><code class="language-plaintext highlighter-rouge">CompactBatchProvider</code></a> which provides a batch such that the operations are compacted by the key. For update operations, the latter update will override the value of the previous one when they have the same key. For query operations, -the operations will be combined as a single operation when they have the same key. </li> -<li>A <a href="https://github.com/apache/samza/samza-core/src/main/java/org/apache/samza/table/batching/CompleteBatchProvider.java";><code>CompleteBatchProvider</code></a> which provides a batch such that +the operations will be combined as a single operation when they have the same key.</li> + <li>A <a href="https://github.com/apache/samza/samza-core/src/main/java/org/apache/samza/table/batching/CompleteBatchProvider.java";><code class="language-plaintext highlighter-rouge">CompleteBatchProvider</code></a> which provides a batch such that all the operations will be visible to the remote store regardless of the keys.</li> -<li>A user-defined instance of [<code>BatchProvider</code>].</li> + <li>A user-defined instance of [<code class="language-plaintext highlighter-rouge">BatchProvider</code>].</li> </ol> -<p>For each [<code>BatchProvider</code>], the user can config the following: -1. Specify the max size the batch can grow before being closed by <code>withmaxBatchSize(int)</code> -2. Specify the max time the batch can last before being closed by <code>withmaxBatchDelay(Duration)</code></p> +<p>For each [<code class="language-plaintext highlighter-rouge">BatchProvider</code>], the user can config the following:</p> +<ol> + <li>Specify the max size the batch can grow before being closed by <code class="language-plaintext highlighter-rouge">withmaxBatchSize(int)</code></li> + <li>Specify the max time the batch can last before being closed by <code class="language-plaintext highlighter-rouge">withmaxBatchDelay(Duration)</code></li> +</ol> <h3 id="rate-limiting">Rate Limiting</h3> <p>Remote Table has built-in client-side rate limiting support in both of its sync and async executions. This is useful when a remote data store does not have server-side rate limiting or is not sophisticated enough to handle heavy -inbound requests. </p> +inbound requests.</p> -<h4 id="configuration">Configuration</h4> +<h4 id="configuration-1">Configuration</h4> -<p>Rate limiting can be enabled with <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTableDescriptor</code></a> in two ways:</p> +<p>Rate limiting can be enabled with <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code class="language-plaintext highlighter-rouge">RemoteTableDescriptor</code></a> in two ways:</p> <ol> -<li>Default: <code>withReadRateLimit()</code> and <code>withWriteRateLimit()</code>. The + <li>Default: <code class="language-plaintext highlighter-rouge">withReadRateLimit()</code> and <code class="language-plaintext highlighter-rouge">withWriteRateLimit()</code>. The <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/util/EmbeddedTaggedRateLimiter.java";>default implementation</a> uses <a href="https://google.github.io/guava/releases/snapshot-jre/api/docs/com/google/common/util/concurrent/RateLimiter.html";>Guava</a> for rate limiting and provides basic throttling on read/write rates.</li> -<li>User-defined instance: <code>withRateLimiter()</code>. Tailored for more advanced + <li>User-defined instance: <code class="language-plaintext highlighter-rouge">withRateLimiter()</code>. Tailored for more advanced usages, eg. custom policies and/or rate limiter libraries</li> </ol> @@ -1150,23 +1191,23 @@ behaviors of distributed data stores. To built-in support for retrying failed I/O operations originated from the data store clients.</p> -<p>Retry capability can be added to a <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code>RemoteTableDescriptor</code></a> by providing a -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/retry/TableRetryPolicy.java";><code>TableRetryPolicy</code></a>, - which consists of three aspects: </p> +<p>Retry capability can be added to a <a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/remote/RemoteTableDescriptor.java";><code class="language-plaintext highlighter-rouge">RemoteTableDescriptor</code></a> by providing a +[<code class="language-plaintext highlighter-rouge">TableRetryPolicy</code>] (https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/retry/TableRetryPolicy.java), + which consists of three aspects:</p> <ul> -<li>Backoff/Sleep policy - fixed, random, exponential with jitters</li> -<li>Termination policy - by total attempts and/or delay</li> -<li>Retriable exception classification - predicate on <code>Throwable</code></li> + <li>Backoff/Sleep policy - fixed, random, exponential with jitters</li> + <li>Termination policy - by total attempts and/or delay</li> + <li>Retriable exception classification - predicate on <code class="language-plaintext highlighter-rouge">Throwable</code></li> </ul> <p>By default, retry is disabled as such failed I/O operations will propagate up and the caller is expected to handle the exception. When enabled, retry is on a -per-request basis such that each individual request is retried independently. </p> +per-request basis such that each individual request is retried independently.</p> <p>Lastly, Remote Table retry provides a set of standard metrics for monitoring. They can be found in -<a href="https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/retry/RetryMetrics.java";><code>RetryMetrics</code></a>.</p> +[<code class="language-plaintext highlighter-rouge">RetryMetrics</code>] (https://github.com/apache/samza/blob/master/samza-core/src/main/java/org/apache/samza/table/retry/RetryMetrics.java).</p> <h3 id="supporting-additional-operations">Supporting Additional Operations</h3> @@ -1174,59 +1215,59 @@ They can be found in supported through the Get/Put/Delete methods. Two categories of operations are supported</p> <ul> -<li>Get/Put/Delete operations with additional arguments</li> -<li>Arbitrary operations through readAsync() and writeAsync()<br></li> + <li>Get/Put/Delete operations with additional arguments</li> + <li>Arbitrary operations through readAsync() and writeAsync()</li> </ul> <p>We only mandate implementers of table functions to provide implementation for Get/Put/Delete without additional arguments. End users can subclass a table function, and invoke operations on remote store directly, if they are not supported by a table function.</p> -<figure class="highlight"><pre><code class="language-java" data-lang="java"><span></span> <span class="mi">1</span> <span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyCouchbaseTableWriteFunction</span><span class="o"><</span><span class="n">V</span><span class="o">></span> <span class="kd">extends</span> <span class="n">CouchbaseTableWriteFunction</span><span class="o"><</span><span class="n">V</span><span class="o">></span> <span class="o">{</span> +<figure class="highlight"><pre><code class="language-java" data-lang="java"> <span class="mi">1</span> <span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyCouchbaseTableWriteFunction</span><span class="o"><</span><span class="no">V</span><span class="o">></span> <span class="kd">extends</span> <span class="nc">CouchbaseTableWriteFunction</span><span class="o"><</span><span class="no">V</span><span class="o">></span> <span class="o">{</span> <span class="mi">2</span> - <span class="mi">3</span> <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">int</span> <span class="n">OP_COUNTER</span> <span class="o">=</span> <span class="mi">1</span><span class="o">;</span> + <span class="mi">3</span> <span class="kd">public</span> <span class="kd">static</span> <span class="kd">final</span> <span class="kt">int</span> <span class="no">OP_COUNTER</span> <span class="o">=</span> <span class="mi">1</span><span class="o">;</span> <span class="mi">4</span> <span class="mi">5</span> <span class="nd">@Override</span> - <span class="mi">6</span> <span class="kd">public</span> <span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="n">CompletableFuture</span><span class="o"><</span><span class="n">T</span><span class="o">></span> <span class="nf">writeAsync</span><span class="o">(</span><span class="kt">int</span> <span class="n">opId</span><span class="o">,</span> <span class="n">Object</span><span class="o">...</span> <span class="n">args</span><span class="o">)</span> <span class="o">{</span> - <span class="mi">7</span> <span class="k">if</span> <span class="o">(</span><span class="n">OP_COUNTER</span> <span class="o">==</span> <span class="n">opId</span><span class="o">)</span> <span class="o">{</span>
[... 397 lines stripped ...]
