You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by gi...@apache.org on 2021/10/04 18:04:04 UTC
[beam] branch asf-site updated: Publishing website 2021/10/04
18:03:25 at commit 2a3d887
This is an automated email from the ASF dual-hosted git repository.
git-site-role pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/asf-site by this push:
new fead510 Publishing website 2021/10/04 18:03:25 at commit 2a3d887
fead510 is described below
commit fead510df4a26d24cdbd06c10cb2af8c252bead6
Author: jenkins <bu...@apache.org>
AuthorDate: Mon Oct 4 18:03:26 2021 +0000
Publishing website 2021/10/04 18:03:25 at commit 2a3d887
---
website/generated-content/documentation/index.xml | 150 ++++++++++++++++++---
.../io/developing-io-python/index.html | 4 +-
.../patterns/cross-language/index.html | 4 +-
.../documentation/programming-guide/index.html | 96 +++++++++----
website/generated-content/get-started/index.xml | 8 +-
.../get-started/mobile-gaming-example/index.html | 8 +-
website/generated-content/sitemap.xml | 2 +-
7 files changed, 215 insertions(+), 57 deletions(-)
diff --git a/website/generated-content/documentation/index.xml b/website/generated-content/documentation/index.xml
index 56112c0..c4ca3be 100644
--- a/website/generated-content/documentation/index.xml
+++ b/website/generated-content/documentation/index.xml
@@ -978,7 +978,7 @@ to <code>_CountingSource</code>. Then, create the wrapper <code>PTransf
</a>
<pre><code>class ReadFromCountingSource(PTransform):
def __init__(self, count):
-super(ReadFromCountingSource, self).__init__()
+super().__init__()
self._count = count
def expand(self, pcoll):
return pcoll | iobase.Read(_CountingSource(self._count))</code></pre>
@@ -1003,7 +1003,7 @@ numbers = pipeline | &#39;ProduceNumbers&#39; &gt;&gt; ReadFromC
<pre><code>class WriteToKVSink(PTransform):
def __init__(self, simplekv, url, final_table_name):
self._simplekv = simplekv
-super(WriteToKVSink, self).__init__()
+super().__init__()
self._url = url
self._final_table_name = final_table_name
def expand(self, pcoll):
@@ -5509,7 +5509,8 @@ the pipeline.</li>
</ul>
</span>
<p class="language-go">If your <code>PCollection</code> uses any non-global windowing function, the Beam Go SDK
-behaves the same way as with global windowing.</p>
+behaves the same way as with global windowing. Windows that are empty in the input
+<code>PCollection</code> will likewise be empty in the output collection.</p>
<h5 id="combining-values-in-a-keyed-pcollection">4.2.4.6. Combining values in a keyed PCollection</h5>
<p>After creating a keyed PCollection (for example, by using a <code>GroupByKey</code>
transform), a common pattern is to combine the collection of values associated
@@ -5724,7 +5725,7 @@ Non-idempotent functions are supported by Beam, but require additional
thought to ensure correctness when there are external side effects.</p>
<span class="language-java language-py">
<blockquote>
-<p><strong>Note:</strong> These requirements apply to subclasses of <code>DoFn</code></span> (a function object
+<p><strong>Note:</strong> These requirements apply to subclasses of <code>DoFn</code>(a function object
used with the <a href="#pardo">ParDo</a> transform), <code>CombineFn</code> (a function object used
with the <a href="#combine">Combine</a> transform), and <code>WindowFn</code> (a function object
used with the <a href="#windowing">Window</a> transform).</p>
@@ -5732,7 +5733,7 @@ used with the <a href="#windowing">Window</a> transform).</p>
</span>
<span class="language-go">
<blockquote>
-<p><strong>Note:</strong> These requirements apply to <code>DoFn</code>s</span> (a function object
+<p><strong>Note:</strong> These requirements apply to <code>DoFn</code>s (a function object
used with the <a href="#pardo">ParDo</a> transform), <code>CombineFn</code>s (a function object used
with the <a href="#combine">Combine</a> transform), and <code>WindowFn</code>s (a function object
used with the <a href="#windowing">Window</a> transform).</p>
@@ -8558,7 +8559,7 @@ window.</p>
<li>Sliding Time Windows</li>
<li>Per-Session Windows</li>
<li>Single Global Window</li>
-<li>Calendar-based Windows (not supported by the Beam SDK for Python)</li>
+<li>Calendar-based Windows (not supported by the Beam SDK for Python or Go)</li>
</ul>
<p>You can also define your own <code>WindowFn</code> if you have a more complex need.</p>
<p>Note that each element can logically belong to more than one window, depending
@@ -8648,6 +8649,16 @@ into fixed windows, each 60 seconds in length:</p>
<span class="n">items</span> <span class="o">|</span> <span class="s1">&#39;window&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">WindowInto</span><span class="p">(</span><span class="n">window</span><span class="o">.</span><span class="n">FixedWindows</span><span class="p">(</span><span class="mi">60</span><span class="p">)))< [...]
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">fixedWindowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="mi">60</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Second</span><span class="p">),</span>
+<span class="nx">items</span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<h4 id="using-sliding-time-windows">8.3.2. Sliding time windows</h4>
<p>The following example code shows how to apply <code>Window</code> to divide a <code>PCollection</code>
into sliding time windows. Each window is 30 seconds in length, and a new window
@@ -8672,6 +8683,16 @@ begins every five seconds:</p>
<span class="n">items</span> <span class="o">|</span> <span class="s1">&#39;window&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">WindowInto</span><span class="p">(</span><span class="n">window</span><span class="o">.</span><span class="n">SlidingWindows</span><span class="p">(</span><span class="mi">30</span><span class="p">,< [...]
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">slidingWindowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewSlidingWindows</span><span class="p">(</span><span class="mi">5</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Second</span><span class="p">,</span> <span class="mi">30</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Second</ [...]
+<span class="nx">items</span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<h4 id="using-session-windows">8.3.3. Session windows</h4>
<p>The following example code shows how to apply <code>Window</code> to divide a <code>PCollection</code>
into session windows, where each session must be separated by a time gap of at
@@ -8696,6 +8717,16 @@ least 10 minutes (600 seconds):</p>
<span class="n">items</span> <span class="o">|</span> <span class="s1">&#39;window&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">WindowInto</span><span class="p">(</span><span class="n">window</span><span class="o">.</span><span class="n">Sessions</span><span class="p">(</span><span class="mi">10</span> <span class="o">*</spa [...]
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">sessionWindowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewSessions</span><span class="p">(</span><span class="mi">600</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Second</span><span class="p">),</span>
+<span class="nx">items</span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<p>Note that the sessions are per-key — each key in the collection will have its
own session groupings depending on the data distribution.</p>
<h4 id="using-single-global-window">8.3.4. Single global window</h4>
@@ -8722,6 +8753,16 @@ a single global window for a <code>PCollection</code>:</p>
<span class="n">items</span> <span class="o">|</span> <span class="s1">&#39;window&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">WindowInto</span><span class="p">(</span><span class="n">window</span><span class="o">.</span><span class="n">GlobalWindows</span><span class="p">()))</span></code></pre></div>
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">globalWindowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewGlobalWindows</span><span class="p">(),</span>
+<span class="nx">items</span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<h3 id="watermarks-and-late-data">8.4. Watermarks and late data</h3>
<p>In any data processing system, there is a certain amount of lag between the time
a data event occurs (the &ldquo;event time&rdquo;, determined by the timestamp on the data
@@ -8786,6 +8827,17 @@ the end of a window.</p>
<span class="n">allowed_lateness</span><span class="o">=</span><span class="n">Duration</span><span class="p">(</span><span class="n">seconds</span><span class="o">=</span><span class="mi">2</span><span class="o">*</span><span class="mi">24</span><span class="o">*</span><span class="mi">60</span><span class="o">*</span><span class="mi">60</span><span class="p">))</span> <span class="c1"># 2 days</sp [...]
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">windowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span> <span class="nx">items</span><span class="p">,</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">AllowedLateness</span><span class="p">(</span><span class="mi">2</span><span class="o">*</span><span class="mi">24</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Hour</span><span class="p">),</span> <span class="c1">// 2 days
+</span><span class="c1"></span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<p>When you set <code>.withAllowedLateness</code> on a <code>PCollection</code>, that allowed lateness
propagates forward to any subsequent <code>PCollection</code> derived from the first
<code>PCollection</code> you applied allowed lateness to. If you want to change the allowed
@@ -8840,7 +8892,29 @@ with a <code>DoFn</code> to attach the timestamps to each element in your
<span class="n">timestamped_items</span> <span class="o">=</span> <span class="n">items</span> <span class="o">|</span> <span class="s1">&#39;timestamp&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">ParDo</span><span class="p">(</span><span class="n">AddTimestampDoFn</span><span class="p">())</span></code></pre></div>
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="c1">// AddTimestampDoFn extracts an event time from a LogEntry.
+</span><span class="c1"></span><span class="kd">func</span> <span class="nf">AddTimestampDoFn</span><span class="p">(</span><span class="nx">element</span> <span class="nx">LogEntry</span><span class="p">,</span> <span class="nx">emit</span> <span class="kd">func</span><span class="p">(</span><span class="nx">beam</span><span class="p">.</span><span class="nx">EventTime</span><span class="p">,</span> & [...]
+<span class="nx">et</span> <span class="o">:=</span> <span class="nf">extractEventTime</span><span class="p">(</span><span class="nx">element</span><span class="p">)</span>
+<span class="c1">// Defining an emitter with beam.EventTime as the first parameter
+</span><span class="c1"></span> <span class="c1">// allows the DoFn to set the event time for the emitted element.
+</span><span class="c1"></span> <span class="nf">emit</span><span class="p">(</span><span class="nx">mtime</span><span class="p">.</span><span class="nf">FromTime</span><span class="p">(</span><span class="nx">et</span><span class="p">),</span> <span class="nx">element</span><span class="p">)</span>
+<span class="p">}</span>
+<span class="c1">// Use the DoFn with ParDo as normal.
+</span><span class="c1"></span>
+<span class="nx">stampedLogs</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">ParDo</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span> <span class="nx">AddTimestampDoFn</span><span class="p">,</span> <span class="nx">unstampedLogs</span><span class="p">)</span></code></pre></div>
+</div>
+</div>
<h2 id="triggers">9. Triggers</h2>
+<span class="language-go">
+<blockquote>
+<p><strong>Note:</strong> The Trigger API in the Beam SDK for Go is currently experimental and subject to change.</p>
+</blockquote>
+</span>
<p>When collecting and grouping data into windows, Beam uses <strong>triggers</strong> to
determine when to emit the aggregated results of each window (referred to as a
<em>pane</em>). If you use Beam&rsquo;s default windowing configuration and <a href="#default-trigger">default
@@ -8936,8 +9010,10 @@ firings:</p>
<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
<img src="/images/copy-icon.svg"/>
</a>
-<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go">
-<span class="nx">trigger</span> <span class="o">:=</span> <span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterEndOfWindow</span><span class="p">().</span><span class="nf">EarlyFiring</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterProcessingTime</span><span class="p">(</span><span class="mi">60000</span><span clas [...]
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">trigger</span> <span class="o">:=</span> <span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterEndOfWindow</span><span class="p">().</span>
+<span class="nf">EarlyFiring</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterProcessingTime</span><span class="p">().</span>
+<span class="nf">PlusDelay</span><span class="p">(</span><span class="mi">60</span> <span class="o">*</span> <span class="nx">time</span><span class="p">.</span><span class="nx">Second</span><span class="p">)).</span>
+<span class="nf">LateFiring</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerRepeat</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterCount</span><span class="p">(</span><span class="mi">1</span><span class="p">)))</span></code></pre></div>
</div>
</div>
<h4 id="default-trigger">9.1.1. Default trigger</h4>
@@ -8992,9 +9068,10 @@ trigger for a <code>PCollection</code>, which emits results one minute aft
element in that window has been processed. The <code>accumulation_mode</code> parameter
sets the window&rsquo;s <strong>accumulation mode</strong>.</p>
<p class="language-go">You set the trigger(s) for a <code>PCollection</code> by passing in the <code>beam.Trigger</code> parameter
-when you use the <code>beam.WindowInto</code> transform. This code sample sets an Always
-trigger for a <code>PCollection</code>, which emits results every time an element in that window has been processed. The <code>beam.AccumulationMode</code> parameter
-sets the window&rsquo;s <strong>accumulation mode</strong>.</p>
+when you use the <code>beam.WindowInto</code> transform. This code sample sets a time-based
+trigger for a <code>PCollection</code>, which emits results one minute after the first
+element in that window has been processed.
+The <code>beam.AccumulationMode</code> parameter sets the window&rsquo;s <strong>accumulation mode</strong>.</p>
<div class='language-java snippet'>
<div class="notebook-skip code-snippet">
<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
@@ -9023,10 +9100,13 @@ sets the window&rsquo;s <strong>accumulation mode</strong>.</p>
<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
<img src="/images/copy-icon.svg"/>
</a>
-<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go">
-<span class="nx">windowSize</span> <span class="o">:=</span> <span class="mi">10</span> <span class="o">*</span> <span class="nx">time</span><span class="p">.</span><span class="nx">Second</span>
-<span class="nx">trigger</span> <span class="o">:=</span> <span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAlways</span><span class="p">()</span>
-<span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span> <span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="nx">windowSize</span><span class="p">),</span> <span class="nx">pCollection</span><span class="p">,</span> <span cl [...]
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">windowedItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span> <span class="nx">pcollection</span><span class="p">,</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">Trigger</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterProcessingTime</span><span class="p">().</span>
+<span class="nf">PlusDelay</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">)),</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">AllowedLateness</span><span class="p">(</span><span class="mi">30</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">PanesDiscard</span><span class="p">(),</span>
+<span class="p">)</span></code></pre></div>
</div>
</div>
<h4 id="window-accumulation-modes">9.4.1. Window accumulation modes</h4>
@@ -9079,8 +9159,10 @@ passes the end of the window, you can apply an <em>allowed lateness</em> w
your windowing configuration. This gives your trigger the opportunity to react
to the late data. If allowed lateness is set, the default trigger will emit new
results immediately whenever late data arrives.</p>
-<p>You set the allowed lateness by using <code>.withAllowedLateness()</code> when you set your
-windowing function:</p>
+<p>You set the allowed lateness by using <span class="language-java"><code>.withAllowedLateness()</code></span>
+<span class="language-py"><code>allowed_lateness</code></span>
+<span class="language-go"><code>beam.AllowedLateness()</code></span>
+when you set your windowing function:</p>
<div class='language-java snippet'>
<div class="notebook-skip code-snippet">
<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
@@ -9106,10 +9188,26 @@ windowing function:</p>
<span class="o">|</span> <span class="o">...</span></code></pre></div>
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">allowedToBeLateItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span> <span class="nx">pcollection</span><span class="p">,</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">Trigger</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterProcessingTime</span><span class="p">().</span>
+<span class="nf">PlusDelay</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">)),</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">AllowedLateness</span><span class="p">(</span><span class="mi">30</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span>
+<span class="p">)</span></code></pre></div>
+</div>
+</div>
<p>This allowed lateness propagates to all <code>PCollection</code>s derived as a result of
applying transforms to the original <code>PCollection</code>. If you want to change the
allowed lateness later in your pipeline, you can apply
-<code>Window.configure().withAllowedLateness()</code> again, explicitly.</p>
+<span class="language-java"><code>Window.configure().withAllowedLateness()</code></span>
+<span class="language-py"><code>allowed_lateness</code></span>
+<span class="language-go"><code>beam.AllowedLateness()</code></span>
+again, explicitly.</p>
<h3 id="composite-triggers">9.5. Composite triggers</h3>
<p>You can combine multiple triggers to form <strong>composite triggers</strong>, and can
specify a trigger to emit results repeatedly, at most once, or under other
@@ -9193,6 +9291,20 @@ trigger stops executing</li>
<span class="n">accumulation_mode</span><span class="o">=</span><span class="n">AccumulationMode</span><span class="o">.</span><span class="n">DISCARDING</span><span class="p">)</span></code></pre></div>
</div>
</div>
+<div class='language-go snippet'>
+<div class="notebook-skip code-snippet">
+<a class="copy" type="button" data-bs-toggle="tooltip" data-bs-placement="bottom" title="Copy to clipboard">
+<img src="/images/copy-icon.svg"/>
+</a>
+<div class="highlight"><pre class="chroma"><code class="language-go" data-lang="go"><span class="nx">compositeTriggerItems</span> <span class="o">:=</span> <span class="nx">beam</span><span class="p">.</span><span class="nf">WindowInto</span><span class="p">(</span><span class="nx">s</span><span class="p">,</span>
+<span class="nx">window</span><span class="p">.</span><span class="nf">NewFixedWindows</span><span class="p">(</span><span class="mi">1</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">),</span> <span class="nx">pcollection</span><span class="p">,</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">Trigger</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterEndOfWindow</span><span class="p">().</span>
+<span class="nf">LateFiring</span><span class="p">(</span><span class="nx">window</span><span class="p">.</span><span class="nf">TriggerAfterProcessingTime</span><span class="p">().</span>
+<span class="nf">PlusDelay</span><span class="p">(</span><span class="mi">10</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Minute</span><span class="p">))),</span>
+<span class="nx">beam</span><span class="p">.</span><span class="nf">AllowedLateness</span><span class="p">(</span><span class="mi">2</span><span class="o">*</span><span class="mi">24</span><span class="o">*</span><span class="nx">time</span><span class="p">.</span><span class="nx">Hour</span><span class="p">),</span>
+<span class="p">)</span></code></pre></div>
+</div>
+</div>
<h4 id="other-composite-triggers">9.5.3. Other composite triggers</h4>
<p>You can also build other sorts of composite triggers. The following example code
shows a simple composite trigger that fires whenever the pane has at least 100
@@ -14000,7 +14112,7 @@ limitations under the License.
<span class="nd">@ptransform.PTransform.register_urn</span><span class="p">(</span><span class="n">URN</span><span class="p">,</span> <span class="bp">None</span><span class="p">)</span>
<span class="k">class</span> <span class="nc">PythonTransform</span><span class="p">(</span><span class="n">ptransform</span><span class="o">.</span><span class="n">PTransform</span><span class="p">):</span>
<span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
-<span class="nb">super</span><span class="p">(</span><span class="n">PythonTransform</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="fm">__init__</span><span class="p">()</span>
+<span class="nb">super</span><span class="p">()</span><span class="o">.</span><span class="fm">__init__</span><span class="p">()</span>
<span class="k">def</span> <span class="nf">expand</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">pcoll</span><span class="p">):</span>
<span class="k">return</span> <span class="p">(</span><span class="n">pcoll</span>
<span class="o">|</span> <span class="s2">&#34;Input preparation&#34;</span>
diff --git a/website/generated-content/documentation/io/developing-io-python/index.html b/website/generated-content/documentation/io/developing-io-python/index.html
index bbd41ee..f95205c 100644
--- a/website/generated-content/documentation/io/developing-io-python/index.html
+++ b/website/generated-content/documentation/io/developing-io-python/index.html
@@ -126,7 +126,7 @@ that they are not exposed to end-users. For the source, rename <code>CountingSou
to <code>_CountingSource</code>. Then, create the wrapper <code>PTransform</code>, called
<code>ReadFromCountingSource</code>:</p><div class=snippet><div class="notebook-skip code-snippet without_switcher"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><pre><code>class ReadFromCountingSource(PTransform):
def __init__(self, count):
- super(ReadFromCountingSource, self).__init__()
+ super().__init__()
self._count = count
def expand(self, pcoll):
@@ -134,7 +134,7 @@ to <code>_CountingSource</code>. Then, create the wrapper <code>PTransform</code
numbers = pipeline | 'ProduceNumbers' >> ReadFromCountingSource(count)</code></pre></div></div><p>For the sink, rename <code>SimpleKVSink</code> to <code>_SimpleKVSink</code>. Then, create the wrapper <code>PTransform</code>, called <code>WriteToKVSink</code>:</p><div class=snippet><div class="notebook-skip code-snippet without_switcher"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a>< [...]
def __init__(self, simplekv, url, final_table_name):
self._simplekv = simplekv
- super(WriteToKVSink, self).__init__()
+ super().__init__()
self._url = url
self._final_table_name = final_table_name
diff --git a/website/generated-content/documentation/patterns/cross-language/index.html b/website/generated-content/documentation/patterns/cross-language/index.html
index ae8afd8..b171fdb 100644
--- a/website/generated-content/documentation/patterns/cross-language/index.html
+++ b/website/generated-content/documentation/patterns/cross-language/index.html
@@ -53,7 +53,7 @@ function openMenu(){addPlaceholder();blockScroll();}</script><div class="clearfi
<span class=nd>@ptransform.PTransform.register_urn</span><span class=p>(</span><span class=n>URN</span><span class=p>,</span> <span class=bp>None</span><span class=p>)</span>
<span class=k>class</span> <span class=nc>PythonTransform</span><span class=p>(</span><span class=n>ptransform</span><span class=o>.</span><span class=n>PTransform</span><span class=p>):</span>
<span class=k>def</span> <span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>):</span>
- <span class=nb>super</span><span class=p>(</span><span class=n>PythonTransform</span><span class=p>,</span> <span class=bp>self</span><span class=p>)</span><span class=o>.</span><span class=fm>__init__</span><span class=p>()</span>
+ <span class=nb>super</span><span class=p>()</span><span class=o>.</span><span class=fm>__init__</span><span class=p>()</span>
<span class=k>def</span> <span class=nf>expand</span><span class=p>(</span><span class=bp>self</span><span class=p>,</span> <span class=n>pcoll</span><span class=p>):</span>
<span class=k>return</span> <span class=p>(</span><span class=n>pcoll</span>
@@ -96,7 +96,7 @@ function openMenu(){addPlaceholder();blockScroll();}</script><div class="clearfi
<span class=n>logging</span><span class=o>.</span><span class=n>getLogger</span><span class=p>()</span><span class=o>.</span><span class=n>setLevel</span><span class=p>(</span><span class=n>logging</span><span class=o>.</span><span class=n>INFO</span><span class=p>)</span>
<span class=n>main</span><span class=p>(</span><span class=n>sys</span><span class=o>.</span><span class=n>argv</span><span class=p>)</span></code></pre></div></div></div><h2 id=how-to-run-the-cross-language-pipeline>How to run the cross-language pipeline?</h2><p>In this section, the steps to run a cross-language pipeline are set out:</p><ol><li><p>Start the <strong>expansion service</strong> with your Python transforms: <code>python expansion_service.py -p 9097</code></p></li><li><p>S [...]
<code>./gradlew :runners:spark:job-server:runShadow</code></li><li>Using the pre-build Docker container:
-<code>docker run -net=host apache/beam_spark_job_server</code></li></ul></li><li><p><strong>Run pipeline</strong>: <code>mvn exec:java -Dexec.mainClass=CrossLanguagePipeline \ -Pportable-runner \ -Dexec.args=" \ --runner=PortableRunner \ --jobEndpoint=localhost:8099 \ --useExternal=true \ --expansionServiceURL=localhost:9097 \ --experiments=beam_fn_api"</code></p></li></ol><div class=feedback><p class=update>Last updated on 2020/12/25</p><h3>Have you found everything you were looking for [...]
+<code>docker run -net=host apache/beam_spark_job_server</code></li></ul></li><li><p><strong>Run pipeline</strong>: <code>mvn exec:java -Dexec.mainClass=CrossLanguagePipeline \ -Pportable-runner \ -Dexec.args=" \ --runner=PortableRunner \ --jobEndpoint=localhost:8099 \ --useExternal=true \ --expansionServiceURL=localhost:9097 \ --experiments=beam_fn_api"</code></p></li></ol><div class=feedback><p class=update>Last updated on 2021/10/04</p><h3>Have you found everything you were looking for [...]
<a href=http://www.apache.org>The Apache Software Foundation</a>
| <a href=/privacy_policy>Privacy Policy</a>
| <a href=/feed.xml>RSS Feed</a><br><br>Apache Beam, Apache, Beam, the Beam logo, and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation. All other products or name brands are trademarks of their respective holders, including The Apache Software Foundation.</div></div></div></div></footer></body></html>
\ No newline at end of file
diff --git a/website/generated-content/documentation/programming-guide/index.html b/website/generated-content/documentation/programming-guide/index.html
index 9cc9792..c092a4c 100644
--- a/website/generated-content/documentation/programming-guide/index.html
+++ b/website/generated-content/documentation/programming-guide/index.html
@@ -1098,7 +1098,8 @@ applying <code>Combine</code>:</p><ul><li>Specify <code>.withoutDefaults</code>,
when used as a side input. You’ll generally only need to use this option if
the result of your pipeline’s <code>Combine</code> is to be used as a side input later in
the pipeline.</li></ul></span><p class=language-go>If your <code>PCollection</code> uses any non-global windowing function, the Beam Go SDK
-behaves the same way as with global windowing.</p><h5 id=combining-values-in-a-keyed-pcollection>4.2.4.6. Combining values in a keyed PCollection</h5><p>After creating a keyed PCollection (for example, by using a <code>GroupByKey</code>
+behaves the same way as with global windowing. Windows that are empty in the input
+<code>PCollection</code> will likewise be empty in the output collection.</p><h5 id=combining-values-in-a-keyed-pcollection>4.2.4.6. Combining values in a keyed PCollection</h5><p>After creating a keyed PCollection (for example, by using a <code>GroupByKey</code>
transform), a common pattern is to combine the collection of values associated
with each key into a single, merged value. Drawing on the previous example from
<code>GroupByKey</code>, a key-grouped <code>PCollection</code> called <code>groupedWords</code> looks like this:</p><pre><code> cat, [1,5,9]
@@ -1230,10 +1231,10 @@ run multiple times. As such, you should be cautious about including things like
state dependency in your user code.</p><p>In general, your user code must fulfill at least these requirements:</p><ul><li>Your function object must be <strong>serializable</strong>.</li><li>Your function object must be <strong>thread-compatible</strong>, and be aware that <em>the
Beam SDKs are not thread-safe</em>.</li></ul><p>In addition, it’s recommended that you make your function object <strong>idempotent</strong>.
Non-idempotent functions are supported by Beam, but require additional
-thought to ensure correctness when there are external side effects.</p><span class="language-java language-py"><blockquote><p><strong>Note:</strong> These requirements apply to subclasses of <code>DoFn</code></span> (a function object
+thought to ensure correctness when there are external side effects.</p><span class="language-java language-py"><blockquote><p><strong>Note:</strong> These requirements apply to subclasses of <code>DoFn</code>(a function object
used with the <a href=#pardo>ParDo</a> transform), <code>CombineFn</code> (a function object used
with the <a href=#combine>Combine</a> transform), and <code>WindowFn</code> (a function object
-used with the <a href=#windowing>Window</a> transform).</p></blockquote></span><span class=language-go><blockquote><p><strong>Note:</strong> These requirements apply to <code>DoFn</code>s</span> (a function object
+used with the <a href=#windowing>Window</a> transform).</p></blockquote></span><span class=language-go><blockquote><p><strong>Note:</strong> These requirements apply to <code>DoFn</code>s (a function object
used with the <a href=#pardo>ParDo</a> transform), <code>CombineFn</code>s (a function object used
with the <a href=#combine>Combine</a> transform), and <code>WindowFn</code>s (a function object
used with the <a href=#windowing>Window</a> transform).</p></blockquote></span><h4 id=user-code-serializability>4.3.1. Serializability</h4><p>Any function object you provide to a transform must be <strong>fully serializable</strong>.
@@ -2546,7 +2547,7 @@ for that <code>PCollection</code>. The <code>GroupByKey</code> transform groups
<code>PCollection</code> by both key and window, based on the windowing function. The
subsequent <code>ParDo</code> transform gets applied multiple times per key, once for each
window.</p><h3 id=provided-windowing-functions>8.2. Provided windowing functions</h3><p>You can define different kinds of windows to divide the elements of your
-<code>PCollection</code>. Beam provides several windowing functions, including:</p><ul><li>Fixed Time Windows</li><li>Sliding Time Windows</li><li>Per-Session Windows</li><li>Single Global Window</li><li>Calendar-based Windows (not supported by the Beam SDK for Python)</li></ul><p>You can also define your own <code>WindowFn</code> if you have a more complex need.</p><p>Note that each element can logically belong to more than one window, depending
+<code>PCollection</code>. Beam provides several windowing functions, including:</p><ul><li>Fixed Time Windows</li><li>Sliding Time Windows</li><li>Per-Session Windows</li><li>Single Global Window</li><li>Calendar-based Windows (not supported by the Beam SDK for Python or Go)</li></ul><p>You can also define your own <code>WindowFn</code> if you have a more complex need.</p><p>Note that each element can logically belong to more than one window, depending
on the windowing function you use. Sliding time windowing, for example, creates
overlapping windows wherein a single element can be assigned to multiple
windows.</p><h4 id=fixed-time-windows>8.2.1. Fixed time windows</h4><p>The simplest form of windowing is using <strong>fixed time windows</strong>: given a
@@ -2594,26 +2595,34 @@ into fixed windows, each 60 seconds in length:</p><div class="language-java snip
<span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>fixedWindowedItems</span> <span class=o>=</span> <span class=n>items</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span>
<span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=n>FixedWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardSeconds</span><span class=o>(</span><span class=n>60</span><span class=o>))));</span></code></pre></div></div></div><div class="language-py snippet"><div cla [...]
<span class=n>fixed_windowed_items</span> <span class=o>=</span> <span class=p>(</span>
- <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>FixedWindows</span><span class=p>(</span><span class=mi>60</span><span class=p>)))</span></code></pre></div></div></div><h4 id=using-sliding-time-windows>8.3.2. Sliding time windows</h4><p>The following exa [...]
+ <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>FixedWindows</span><span class=p>(</span><span class=mi>60</span><span class=p>)))</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=mi>60</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Second</span><span class=p>),</span>
+ <span class=nx>items</span><span class=p>)</span></code></pre></div></div></div><h4 id=using-sliding-time-windows>8.3.2. Sliding time windows</h4><p>The following example code shows how to apply <code>Window</code> to divide a <code>PCollection</code>
into sliding time windows. Each window is 30 seconds in length, and a new window
begins every five seconds:</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>items</span> <span class=o>=</span> <span cl [...]
<span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>slidingWindowedItems</span> <span class=o>=</span> <span class=n>items</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span>
<span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=n>SlidingWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardSeconds</span><span class=o>(</span><span class=n>30</span><span class=o>)).</span><span class=na>every</span><span class=o>(</span><span class=n>Duration< [...]
<span class=n>sliding_windowed_items</span> <span class=o>=</span> <span class=p>(</span>
- <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>SlidingWindows</span><span class=p>(</span><span class=mi>30</span><span class=p>,</span> <span class=mi>5</span><span class=p>)))</span></code></pre></div></div></div><h4 id=using-session-windows>8.3.3. Se [...]
+ <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>SlidingWindows</span><span class=p>(</span><span class=mi>30</span><span class=p>,</span> <span class=mi>5</span><span class=p>)))</span></code></pre></div></div></div><div class="language-go snippet"><div [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewSlidingWindows</span><span class=p>(</span><span class=mi>5</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Second</span><span class=p>,</span> <span class=mi>30</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Second</span><span class=p>),</span>
+ <span class=nx>items</span><span class=p>)</span></code></pre></div></div></div><h4 id=using-session-windows>8.3.3. Session windows</h4><p>The following example code shows how to apply <code>Window</code> to divide a <code>PCollection</code>
into session windows, where each session must be separated by a time gap of at
least 10 minutes (600 seconds):</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>items</span> <span class=o>=</span> <sp [...]
<span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>sessionWindowedItems</span> <span class=o>=</span> <span class=n>items</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span>
<span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=n>Sessions</span><span class=o>.</span><span class=na>withGapDuration</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardSeconds</span><span class=o>(</span><span class=n>600</span><span class=o>))));</span></code></pre></div></div></div><div class="language-py snippet [...]
<span class=n>session_windowed_items</span> <span class=o>=</span> <span class=p>(</span>
- <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>Sessions</span><span class=p>(</span><span class=mi>10</span> <span class=o>*</span> <span class=mi>60</span><span class=p>)))</span></code></pre></div></div></div><p>Note that the sessions are per-key — ea [...]
+ <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>Sessions</span><span class=p>(</span><span class=mi>10</span> <span class=o>*</span> <span class=mi>60</span><span class=p>)))</span></code></pre></div></div></div><div class="language-go snippet"><div clas [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewSessions</span><span class=p>(</span><span class=mi>600</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Second</span><span class=p>),</span>
+ <span class=nx>items</span><span class=p>)</span></code></pre></div></div></div><p>Note that the sessions are per-key — each key in the collection will have its
own session groupings depending on the data distribution.</p><h4 id=using-single-global-window>8.3.4. Single global window</h4><p>If your <code>PCollection</code> is bounded (the size is fixed), you can assign all the
elements to a single global window. The following example code shows how to set
a single global window for a <code>PCollection</code>:</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>items</span> <sp [...]
<span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>batchItems</span> <span class=o>=</span> <span class=n>items</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span>
<span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=k>new</span> <span class=n>GlobalWindows</span><span class=o>()));</span></code></pre></div></div></div><div class="language-py snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div cla [...]
<span class=n>global_windowed_items</span> <span class=o>=</span> <span class=p>(</span>
- <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>GlobalWindows</span><span class=p>()))</span></code></pre></div></div></div><h3 id=watermarks-and-late-data>8.4. Watermarks and late data</h3><p>In any data processing system, there is a certain amount of l [...]
+ <span class=n>items</span> <span class=o>|</span> <span class=s1>'window'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>WindowInto</span><span class=p>(</span><span class=n>window</span><span class=o>.</span><span class=n>GlobalWindows</span><span class=p>()))</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs- [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewGlobalWindows</span><span class=p>(),</span>
+ <span class=nx>items</span><span class=p>)</span></code></pre></div></div></div><h3 id=watermarks-and-late-data>8.4. Watermarks and late data</h3><p>In any data processing system, there is a certain amount of lag between the time
a data event occurs (the “event time”, determined by the timestamp on the data
element itself) and the time the actual data element gets processed at any stage
in your pipeline (the “processing time”, determined by the clock on the system
@@ -2652,7 +2661,10 @@ the end of a window.</p><div class="language-java snippet"><div class="notebook-
<span class=n>trigger</span><span class=o>=</span><span class=n>trigger_fn</span><span class=p>,</span>
<span class=n>accumulation_mode</span><span class=o>=</span><span class=n>accumulation_mode</span><span class=p>,</span>
<span class=n>timestamp_combiner</span><span class=o>=</span><span class=n>timestamp_combiner</span><span class=p>,</span>
- <span class=n>allowed_lateness</span><span class=o>=</span><span class=n>Duration</span><span class=p>(</span><span class=n>seconds</span><span class=o>=</span><span class=mi>2</span><span class=o>*</span><span class=mi>24</span><span class=o>*</span><span class=mi>60</span><span class=o>*</span><span class=mi>60</span><span class=p>))</span> <span class=c1># 2 days</span></code></pre></div></div></div><p>When you set <code>.withAllowedLateness</code> on a <code>PCollection [...]
+ <span class=n>allowed_lateness</span><span class=o>=</span><span class=n>Duration</span><span class=p>(</span><span class=n>seconds</span><span class=o>=</span><span class=mi>2</span><span class=o>*</span><span class=mi>24</span><span class=o>*</span><span class=mi>60</span><span class=o>*</span><span class=mi>60</span><span class=p>))</span> <span class=c1># 2 days</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet" [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span> <span class=nx>items</span><span class=p>,</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>AllowedLateness</span><span class=p>(</span><span class=mi>2</span><span class=o>*</span><span class=mi>24</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Hour</span><span class=p>),</span> <span class=c1>// 2 days
+</span><span class=c1></span><span class=p>)</span></code></pre></div></div></div><p>When you set <code>.withAllowedLateness</code> on a <code>PCollection</code>, that allowed lateness
propagates forward to any subsequent <code>PCollection</code> derived from the first
<code>PCollection</code> you applied allowed lateness to. If you want to change the allowed
lateness later in your pipeline, you must do so explicitly by applying
@@ -2685,7 +2697,19 @@ with a <code>DoFn</code> to attach the timestamps to each element in your <code>
<span class=c1># TimestampedValue.</span>
<span class=k>yield</span> <span class=n>beam</span><span class=o>.</span><span class=n>window</span><span class=o>.</span><span class=n>TimestampedValue</span><span class=p>(</span><span class=n>element</span><span class=p>,</span> <span class=n>unix_timestamp</span><span class=p>)</span>
-<span class=n>timestamped_items</span> <span class=o>=</span> <span class=n>items</span> <span class=o>|</span> <span class=s1>'timestamp'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>ParDo</span><span class=p>(</span><span class=n>AddTimestampDoFn</span><span class=p>())</span></code></pre></div></div></div><h2 id=triggers>9. Triggers</h2><p>When collecting and grouping data into windows, Beam uses <strong>triggers</strong> to
+<span class=n>timestamped_items</span> <span class=o>=</span> <span class=n>items</span> <span class=o>|</span> <span class=s1>'timestamp'</span> <span class=o>>></span> <span class=n>beam</span><span class=o>.</span><span class=n>ParDo</span><span class=p>(</span><span class=n>AddTimestampDoFn</span><span class=p>())</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip [...]
+</span><span class=c1></span><span class=kd>func</span> <span class=nf>AddTimestampDoFn</span><span class=p>(</span><span class=nx>element</span> <span class=nx>LogEntry</span><span class=p>,</span> <span class=nx>emit</span> <span class=kd>func</span><span class=p>(</span><span class=nx>beam</span><span class=p>.</span><span class=nx>EventTime</span><span class=p>,</span> <span class=nx>LogEntry</span><span class=p>))</span> <span class=p>{</span>
+ <span class=nx>et</span> <span class=o>:=</span> <span class=nf>extractEventTime</span><span class=p>(</span><span class=nx>element</span><span class=p>)</span>
+ <span class=c1>// Defining an emitter with beam.EventTime as the first parameter
+</span><span class=c1></span> <span class=c1>// allows the DoFn to set the event time for the emitted element.
+</span><span class=c1></span> <span class=nf>emit</span><span class=p>(</span><span class=nx>mtime</span><span class=p>.</span><span class=nf>FromTime</span><span class=p>(</span><span class=nx>et</span><span class=p>),</span> <span class=nx>element</span><span class=p>)</span>
+<span class=p>}</span>
+
+
+
+<span class=c1>// Use the DoFn with ParDo as normal.
+</span><span class=c1></span>
+<span class=nx>stampedLogs</span> <span class=o>:=</span> <span class=nx>beam</span><span class=p>.</span><span class=nf>ParDo</span><span class=p>(</span><span class=nx>s</span><span class=p>,</span> <span class=nx>AddTimestampDoFn</span><span class=p>,</span> <span class=nx>unstampedLogs</span><span class=p>)</span></code></pre></div></div></div><h2 id=triggers>9. Triggers</h2><span class=language-go><blockquote><p><strong>Note:</strong> The Trigger API in the Beam SDK for Go is curren [...]
determine when to emit the aggregated results of each window (referred to as a
<em>pane</em>). If you use Beam’s default windowing configuration and <a href=#default-trigger>default
trigger</a>, Beam outputs the aggregated result when it
@@ -2734,8 +2758,10 @@ firings:</p><div class="language-java snippet"><div class="notebook-skip code-sn
<span class=o>.</span><span class=na>plusDuration</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardMinutes</span><span class=o>(</span><span class=n>1</span><span class=o>))</span>
<span class=c1>// Fire on any late data so the bill can be corrected.
</span><span class=c1></span> <span class=o>.</span><span class=na>withLateFirings</span><span class=o>(</span><span class=n>AfterPane</span><span class=o>.</span><span class=na>elementCountAtLeast</span><span class=o>(</span><span class=n>1</span><span class=o>))</span></code></pre></div></div></div><div class="language-py snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/im [...]
- <span class=n>early</span><span class=o>=</span><span class=n>AfterProcessingTime</span><span class=p>(</span><span class=n>delay</span><span class=o>=</span><span class=mi>1</span> <span class=o>*</span> <span class=mi>60</span><span class=p>),</span> <span class=n>late</span><span class=o>=</span><span class=n>AfterCount</span><span class=p>(</span><span class=mi>1</span><span class=p>))</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip [...]
-<span class=nx>trigger</span> <span class=o>:=</span> <span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterEndOfWindow</span><span class=p>().</span><span class=nf>EarlyFiring</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterProcessingTime</span><span class=p>(</span><span class=mi>60000</span><span class=p>)).</span><span class=nf>LateFiring</span><span class=p>(</span><span class=nx>window</span><span class [...]
+ <span class=n>early</span><span class=o>=</span><span class=n>AfterProcessingTime</span><span class=p>(</span><span class=n>delay</span><span class=o>=</span><span class=mi>1</span> <span class=o>*</span> <span class=mi>60</span><span class=p>),</span> <span class=n>late</span><span class=o>=</span><span class=n>AfterCount</span><span class=p>(</span><span class=mi>1</span><span class=p>))</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip [...]
+ <span class=nf>EarlyFiring</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterProcessingTime</span><span class=p>().</span>
+ <span class=nf>PlusDelay</span><span class=p>(</span><span class=mi>60</span> <span class=o>*</span> <span class=nx>time</span><span class=p>.</span><span class=nx>Second</span><span class=p>)).</span>
+ <span class=nf>LateFiring</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerRepeat</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterCount</span><span class=p>(</span><span class=mi>1</span><span class=p>)))</span></code></pre></div></div></div><h4 id=default-trigger>9.1.1. Default trigger</h4><p>The default trigger for a <code>PCollection</code> is based on event time, and emits the
results of the window when the Beam’s watermark passes the end of the window,
and then fires each time late data arrives.</p><p>However, if you are using both the default windowing configuration and the
default trigger, the default trigger emits exactly once, and late data is
@@ -2774,19 +2800,23 @@ when you use the <code>WindowInto</code> transform. This code sample sets a time
trigger for a <code>PCollection</code>, which emits results one minute after the first
element in that window has been processed. The <code>accumulation_mode</code> parameter
sets the window’s <strong>accumulation mode</strong>.</p><p class=language-go>You set the trigger(s) for a <code>PCollection</code> by passing in the <code>beam.Trigger</code> parameter
-when you use the <code>beam.WindowInto</code> transform. This code sample sets an Always
-trigger for a <code>PCollection</code>, which emits results every time an element in that window has been processed. The <code>beam.AccumulationMode</code> parameter
-sets the window’s <strong>accumulation mode</strong>.</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>pc</span> <sp [...]
+when you use the <code>beam.WindowInto</code> transform. This code sample sets a time-based
+trigger for a <code>PCollection</code>, which emits results one minute after the first
+element in that window has been processed.
+The <code>beam.AccumulationMode</code> parameter sets the window’s <strong>accumulation mode</strong>.</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><spa [...]
<span class=n>pc</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span><span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=n>FixedWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>1</span><span class=o>,</span> <span class=n>TimeUnit</span><span class=o>.</span><span class=na>MINUTES</span><span class=o>))</span>
<span class=o>.</span><span class=na>triggering</span><span class=o>(</span><span class=n>AfterProcessingTime</span><span class=o>.</span><span class=na>pastFirstElementInPane</span><span class=o>()</span>
<span class=o>.</span><span class=na>plusDelayOf</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardMinutes</span><span class=o>(</span><span class=n>1</span><span class=o>)))</span>
<span class=o>.</span><span class=na>discardingFiredPanes</span><span class=o>());</span></code></pre></div></div></div><div class="language-py snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-py data-lang=py> <span class=n>pcollection</span> <span class=o>|</span> [...]
<span class=n>FixedWindows</span><span class=p>(</span><span class=mi>1</span> <span class=o>*</span> <span class=mi>60</span><span class=p>),</span>
<span class=n>trigger</span><span class=o>=</span><span class=n>AfterProcessingTime</span><span class=p>(</span><span class=mi>1</span> <span class=o>*</span> <span class=mi>60</span><span class=p>),</span>
- <span class=n>accumulation_mode</span><span class=o>=</span><span class=n>AccumulationMode</span><span class=o>.</span><span class=n>DISCARDING</span><span class=p>)</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-go data-lang=go>
-<span class=nx>windowSize</span> <span class=o>:=</span> <span class=mi>10</span> <span class=o>*</span> <span class=nx>time</span><span class=p>.</span><span class=nx>Second</span>
-<span class=nx>trigger</span> <span class=o>:=</span> <span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAlways</span><span class=p>()</span>
-<span class=nx>beam</span><span class=p>.</span><span class=nf>WindowInto</span><span class=p>(</span><span class=nx>s</span><span class=p>,</span> <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=nx>windowSize</span><span class=p>),</span> <span class=nx>pCollection</span><span class=p>,</span> <span class=nx>beam</span><span class=p>.</span><span class=nf>Trigger</span><span class=p>(</span><span class=nx>trigger</ [...]
+ <span class=n>accumulation_mode</span><span class=o>=</span><span class=n>AccumulationMode</span><span class=o>.</span><span class=n>DISCARDING</span><span class=p>)</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-go data-lang=go>< [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span> <span class=nx>pcollection</span><span class=p>,</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>Trigger</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterProcessingTime</span><span class=p>().</span>
+ <span class=nf>PlusDelay</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>)),</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>AllowedLateness</span><span class=p>(</span><span class=mi>30</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>PanesDiscard</span><span class=p>(),</span>
+<span class=p>)</span></code></pre></div></div></div><h4 id=window-accumulation-modes>9.4.1. Window accumulation modes</h4><p>When you specify a trigger, you must also set the the window’s <strong>accumulation
mode</strong>. When a trigger fires, it emits the current contents of the window as a
pane. Since a trigger can fire multiple times, the accumulation mode determines
whether the system <em>accumulates</em> the window panes as the trigger fires, or
@@ -2818,8 +2848,10 @@ on each firing:</p><pre><code> First trigger firing: [5, 8, 3]
passes the end of the window, you can apply an <em>allowed lateness</em> when you set
your windowing configuration. This gives your trigger the opportunity to react
to the late data. If allowed lateness is set, the default trigger will emit new
-results immediately whenever late data arrives.</p><p>You set the allowed lateness by using <code>.withAllowedLateness()</code> when you set your
-windowing function:</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>pc</span> <span class=o>=</span> <span class=o>...;</span>
+results immediately whenever late data arrives.</p><p>You set the allowed lateness by using <span class=language-java><code>.withAllowedLateness()</code></span>
+<span class=language-py><code>allowed_lateness</code></span>
+<span class=language-go><code>beam.AllowedLateness()</code></span>
+when you set your windowing function:</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>pc</span> <span class=o>=</span> <s [...]
<span class=n>pc</span><span class=o>.</span><span class=na>apply</span><span class=o>(</span><span class=n>Window</span><span class=o>.<</span><span class=n>String</span><span class=o>></span><span class=n>into</span><span class=o>(</span><span class=n>FixedWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>1</span><span class=o>,</span> <span class=n>TimeUnit</span><span class=o>.</span><span class=na>MINUTES</span><span class=o>))</span>
<span class=o>.</span><span class=na>triggering</span><span class=o>(</span><span class=n>AfterProcessingTime</span><span class=o>.</span><span class=na>pastFirstElementInPane</span><span class=o>()</span>
<span class=o>.</span><span class=na>plusDelayOf</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardMinutes</span><span class=o>(</span><span class=n>1</span><span class=o>)))</span>
@@ -2828,10 +2860,18 @@ windowing function:</p><div class="language-java snippet"><div class="notebook-s
<span class=n>FixedWindows</span><span class=p>(</span><span class=mi>60</span><span class=p>),</span>
<span class=n>trigger</span><span class=o>=</span><span class=n>AfterProcessingTime</span><span class=p>(</span><span class=mi>60</span><span class=p>),</span>
<span class=n>allowed_lateness</span><span class=o>=</span><span class=mi>1800</span><span class=p>)</span> <span class=c1># 30 minutes</span>
- <span class=o>|</span> <span class=o>...</span></code></pre></div></div></div><p>This allowed lateness propagates to all <code>PCollection</code>s derived as a result of
+ <span class=o>|</span> <span class=o>...</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-go data-lang=go><span class=nx>allowedToBeLateItems</span> <span class=o>:=</span> <span class=nx>beam</span><span class=p>.</span><span clas [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span> <span class=nx>pcollection</span><span class=p>,</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>Trigger</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterProcessingTime</span><span class=p>().</span>
+ <span class=nf>PlusDelay</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>)),</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>AllowedLateness</span><span class=p>(</span><span class=mi>30</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span>
+<span class=p>)</span></code></pre></div></div></div><p>This allowed lateness propagates to all <code>PCollection</code>s derived as a result of
applying transforms to the original <code>PCollection</code>. If you want to change the
allowed lateness later in your pipeline, you can apply
-<code>Window.configure().withAllowedLateness()</code> again, explicitly.</p><h3 id=composite-triggers>9.5. Composite triggers</h3><p>You can combine multiple triggers to form <strong>composite triggers</strong>, and can
+<span class=language-java><code>Window.configure().withAllowedLateness()</code></span>
+<span class=language-py><code>allowed_lateness</code></span>
+<span class=language-go><code>beam.AllowedLateness()</code></span>
+again, explicitly.</p><h3 id=composite-triggers>9.5. Composite triggers</h3><p>You can combine multiple triggers to form <strong>composite triggers</strong>, and can
specify a trigger to emit results repeatedly, at most once, or under other
custom conditions.</p><h4 id=composite-trigger-types>9.5.1. Composite trigger types</h4><p>Beam includes the following composite triggers:</p><ul><li>You can add additional early firings or late firings to
<code>AfterWatermark.pastEndOfWindow</code> via <code>.withEarlyFirings</code> and
@@ -2864,7 +2904,13 @@ trigger stops executing</li></ul></p><div class="language-java snippet"><div cla
<span class=n>FixedWindows</span><span class=p>(</span><span class=mi>1</span> <span class=o>*</span> <span class=mi>60</span><span class=p>),</span>
<span class=n>trigger</span><span class=o>=</span><span class=n>AfterWatermark</span><span class=p>(</span><span class=n>late</span><span class=o>=</span><span class=n>AfterProcessingTime</span><span class=p>(</span><span class=mi>10</span> <span class=o>*</span> <span class=mi>60</span><span class=p>)),</span>
<span class=n>allowed_lateness</span><span class=o>=</span><span class=mi>10</span><span class=p>,</span>
- <span class=n>accumulation_mode</span><span class=o>=</span><span class=n>AccumulationMode</span><span class=o>.</span><span class=n>DISCARDING</span><span class=p>)</span></code></pre></div></div></div><h4 id=other-composite-triggers>9.5.3. Other composite triggers</h4><p>You can also build other sorts of composite triggers. The following example code
+ <span class=n>accumulation_mode</span><span class=o>=</span><span class=n>AccumulationMode</span><span class=o>.</span><span class=n>DISCARDING</span><span class=p>)</span></code></pre></div></div></div><div class="language-go snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-go data-lang=go>< [...]
+ <span class=nx>window</span><span class=p>.</span><span class=nf>NewFixedWindows</span><span class=p>(</span><span class=mi>1</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>),</span> <span class=nx>pcollection</span><span class=p>,</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>Trigger</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterEndOfWindow</span><span class=p>().</span>
+ <span class=nf>LateFiring</span><span class=p>(</span><span class=nx>window</span><span class=p>.</span><span class=nf>TriggerAfterProcessingTime</span><span class=p>().</span>
+ <span class=nf>PlusDelay</span><span class=p>(</span><span class=mi>10</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Minute</span><span class=p>))),</span>
+ <span class=nx>beam</span><span class=p>.</span><span class=nf>AllowedLateness</span><span class=p>(</span><span class=mi>2</span><span class=o>*</span><span class=mi>24</span><span class=o>*</span><span class=nx>time</span><span class=p>.</span><span class=nx>Hour</span><span class=p>),</span>
+<span class=p>)</span></code></pre></div></div></div><h4 id=other-composite-triggers>9.5.3. Other composite triggers</h4><p>You can also build other sorts of composite triggers. The following example code
shows a simple composite trigger that fires whenever the pane has at least 100
elements, or after a minute.</p><div class="language-java snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-java data-lang=java> <span class=n>Repeatedly</span><span class=o>.</span><span class=na>forever</span><span class=o>(</span><span class=n>AfterFirst</span><span class=o>.</span><span class [...]
<span class=n>AfterPane</span><span class=o>.</span><span class=na>elementCountAtLeast</span><span class=o>(</span><span class=n>100</span><span class=o>),</span>
@@ -4131,7 +4177,7 @@ expansionAddr := "localhost:8097"
outT := beam.UnnamedOutput(typex.New(reflectx.String))
res := beam.CrossLanguage(s, urn, payload, expansionAddr, beam.UnnamedInput(inputPCol), outT)
</code></pre></div></div></li><li><p>After the job has been submitted to the Beam runner, shutdown the expansion service by
-terminating the expansion service process.</p></li></ol><h3 id=x-lang-transform-runner-support>13.3. Runner Support</h3><p>Currently, portable runners such as Flink, Spark, and the Direct runner can be used with multi-language pipelines.</p><p>Google Cloud Dataflow supports multi-language pipelines through the Dataflow Runner v2 backend architecture.</p><div class=feedback><p class=update>Last updated on 2021/10/01</p><h3>Have you found everything you were looking for?</h3><p class=descr [...]
+terminating the expansion service process.</p></li></ol><h3 id=x-lang-transform-runner-support>13.3. Runner Support</h3><p>Currently, portable runners such as Flink, Spark, and the Direct runner can be used with multi-language pipelines.</p><p>Google Cloud Dataflow supports multi-language pipelines through the Dataflow Runner v2 backend architecture.</p><div class=feedback><p class=update>Last updated on 2021/10/04</p><h3>Have you found everything you were looking for?</h3><p class=descr [...]
<a href=http://www.apache.org>The Apache Software Foundation</a>
| <a href=/privacy_policy>Privacy Policy</a>
| <a href=/feed.xml>RSS Feed</a><br><br>Apache Beam, Apache, Beam, the Beam logo, and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation. All other products or name brands are trademarks of their respective holders, including The Apache Software Foundation.</div></div></div></div></footer></body></html>
\ No newline at end of file
diff --git a/website/generated-content/get-started/index.xml b/website/generated-content/get-started/index.xml
index df139c5..bd3de12 100644
--- a/website/generated-content/get-started/index.xml
+++ b/website/generated-content/get-started/index.xml
@@ -152,7 +152,7 @@ looks more like what is depicted by the red squiggly line above the ideal line.&
</span><span class="s2"> &#34;&#34;&#34;</span>
<span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">field</span><span class="p">):</span>
<span class="c1"># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
-<span class="c1"># super(ExtractAndSumScore, self).__init__()</span>
+<span class="c1"># super().__init__()</span>
<span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">field</span> <span class="o">=</span> <span class="n">field</span>
<span class="k">def</span> <span class="nf">expand</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">pcoll</span><span class="p">):</span>
@@ -392,7 +392,7 @@ logical windows based on when those scores occurred in event time.</em></p
<div class="highlight"><pre class="chroma"><code class="language-py" data-lang="py"><span class="k">class</span> <span class="nc">HourlyTeamScore</span><span class="p">(</span><span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="p">):</span>
<span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">start_min</span><span class="p">,</span> <span class="n">stop_min</span><span class="p">,</span> <span class="n">window_duration</span><span class="p">):</span>
<span class="c1"># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
-<span class="c1"># super(HourlyTeamScore, self).__init__()</span>
+<span class="c1"># super().__init__()</span>
<span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">start_timestamp</span> <span class="o">=</span> <span class="n">str2timestamp</span><span class="p">(</span><span class="n">start_min</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">stop_timestamp</span> <span class="o">=</span> <span class="n">str2timestamp</span><span class="p">(</span><span class="n">stop_min</span><span class="p">)</span>
@@ -578,7 +578,7 @@ ten minutes after data is received.</em></p>
</span><span class="s2"> &#34;&#34;&#34;</span>
<span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">allowed_lateness</span><span class="p">):</span>
<span class="c1"># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
-<span class="c1"># super(CalculateUserScores, self).__init__()</span>
+<span class="c1"># super().__init__()</span>
<span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">allowed_lateness_seconds</span> <span class="o">=</span> <span class="n">allowed_lateness</span> <span class="o">*</span> <span class="mi">60</span>
<span class="k">def</span> <span class="nf">expand</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">pcoll</span><span class="p">):</span>
@@ -662,7 +662,7 @@ late results.</em></p>
</span><span class="s2"> &#34;&#34;&#34;</span>
<span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">team_window_duration</span><span class="p">,</span> <span class="n">allowed_lateness</span><span class="p">):</span>
<span class="c1"># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
-<span class="c1"># super(CalculateTeamScores, self).__init__()</span>
+<span class="c1"># super().__init__()</span>
<span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">team_window_duration</span> <span class="o">=</span> <span class="n">team_window_duration</span> <span class="o">*</span> <span class="mi">60</span>
<span class="bp">self</span><span class="o">.</span><span class="n">allowed_lateness_seconds</span> <span class="o">=</span> <span class="n">allowed_lateness</span> <span class="o">*</span> <span class="mi">60</span>
diff --git a/website/generated-content/get-started/mobile-gaming-example/index.html b/website/generated-content/get-started/mobile-gaming-example/index.html
index b8f04d2..d1cfe1a 100644
--- a/website/generated-content/get-started/mobile-gaming-example/index.html
+++ b/website/generated-content/get-started/mobile-gaming-example/index.html
@@ -49,7 +49,7 @@ looks more like what is depicted by the red squiggly line above the ideal line.<
</span><span class=s2> """</span>
<span class=k>def</span> <span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>,</span> <span class=n>field</span><span class=p>):</span>
<span class=c1># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
- <span class=c1># super(ExtractAndSumScore, self).__init__()</span>
+ <span class=c1># super().__init__()</span>
<span class=n>beam</span><span class=o>.</span><span class=n>PTransform</span><span class=o>.</span><span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>)</span>
<span class=bp>self</span><span class=o>.</span><span class=n>field</span> <span class=o>=</span> <span class=n>field</span>
@@ -179,7 +179,7 @@ logical windows based on when those scores occurred in event time.</em></p><p>No
<span class=o>}</span></code></pre></div></div></div><div class="language-py snippet"><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre class=chroma><code class=language-py data-lang=py><span class=k>class</span> <span class=nc>HourlyTeamScore</span><span class=p>(</span><span class=n>beam</span><span class=o>.</span><span class=n>PTransfo [...]
<span class=k>def</span> <span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>,</span> <span class=n>start_min</span><span class=p>,</span> <span class=n>stop_min</span><span class=p>,</span> <span class=n>window_duration</span><span class=p>):</span>
<span class=c1># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
- <span class=c1># super(HourlyTeamScore, self).__init__()</span>
+ <span class=c1># super().__init__()</span>
<span class=n>beam</span><span class=o>.</span><span class=n>PTransform</span><span class=o>.</span><span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>)</span>
<span class=bp>self</span><span class=o>.</span><span class=n>start_timestamp</span> <span class=o>=</span> <span class=n>str2timestamp</span><span class=p>(</span><span class=n>start_min</span><span class=p>)</span>
<span class=bp>self</span><span class=o>.</span><span class=n>stop_timestamp</span> <span class=o>=</span> <span class=n>str2timestamp</span><span class=p>(</span><span class=n>stop_min</span><span class=p>)</span>
@@ -328,7 +328,7 @@ ten minutes after data is received.</em></p><p>As processing time advances and m
</span><span class=s2> """</span>
<span class=k>def</span> <span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>,</span> <span class=n>allowed_lateness</span><span class=p>):</span>
<span class=c1># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
- <span class=c1># super(CalculateUserScores, self).__init__()</span>
+ <span class=c1># super().__init__()</span>
<span class=n>beam</span><span class=o>.</span><span class=n>PTransform</span><span class=o>.</span><span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>)</span>
<span class=bp>self</span><span class=o>.</span><span class=n>allowed_lateness_seconds</span> <span class=o>=</span> <span class=n>allowed_lateness</span> <span class=o>*</span> <span class=mi>60</span>
@@ -387,7 +387,7 @@ late results.</em></p><p>The dotted line in the diagram is the “ideal&rdqu
</span><span class=s2> """</span>
<span class=k>def</span> <span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>,</span> <span class=n>team_window_duration</span><span class=p>,</span> <span class=n>allowed_lateness</span><span class=p>):</span>
<span class=c1># TODO(BEAM-6158): Revert the workaround once we can pickle super() on py3.</span>
- <span class=c1># super(CalculateTeamScores, self).__init__()</span>
+ <span class=c1># super().__init__()</span>
<span class=n>beam</span><span class=o>.</span><span class=n>PTransform</span><span class=o>.</span><span class=fm>__init__</span><span class=p>(</span><span class=bp>self</span><span class=p>)</span>
<span class=bp>self</span><span class=o>.</span><span class=n>team_window_duration</span> <span class=o>=</span> <span class=n>team_window_duration</span> <span class=o>*</span> <span class=mi>60</span>
<span class=bp>self</span><span class=o>.</span><span class=n>allowed_lateness_seconds</span> <span class=o>=</span> <span class=n>allowed_lateness</span> <span class=o>*</span> <span class=mi>60</span>
diff --git a/website/generated-content/sitemap.xml b/website/generated-content/sitemap.xml
index 980082e..4816d58 100644
--- a/website/generated-content/sitemap.xml
+++ b/website/generated-content/sitemap.xml
@@ -1 +1 @@
-<?xml version="1.0" encoding="utf-8" standalone="yes"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:xhtml="http://www.w3.org/1999/xhtml"><url><loc>/blog/beam-2.32.0/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/categories/blog/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/blog/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/categories/</loc><lastmod>2021-09-17T14:05:48-07:00</lastmod></url><url><loc>/blog/b [...]
\ No newline at end of file
+<?xml version="1.0" encoding="utf-8" standalone="yes"?><urlset xmlns="http://www.sitemaps.org/schemas/sitemap/0.9" xmlns:xhtml="http://www.w3.org/1999/xhtml"><url><loc>/blog/beam-2.32.0/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/categories/blog/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/blog/</loc><lastmod>2021-09-16T12:21:14-07:00</lastmod></url><url><loc>/categories/</loc><lastmod>2021-09-17T14:05:48-07:00</lastmod></url><url><loc>/blog/b [...]
\ No newline at end of file