You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by me...@apache.org on 2018/05/07 17:11:25 UTC
[beam-site] branch asf-site updated (d71494f -> ae67bad)
This is an automated email from the ASF dual-hosted git repository.
mergebot-role pushed a change to branch asf-site
in repository https://gitbox.apache.org/repos/asf/beam-site.git.
from d71494f Merge #435: Fix/remove references to old team page
add 6bff9dc Updates for Python streaming
add 4605bf7 This closes #410
new ae67bad Prepare repository for deployment.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
content/documentation/dsls/sql/index.html | 1 +
content/documentation/programming-guide/index.html | 147 ++++++----
content/documentation/runners/dataflow/index.html | 100 ++++---
content/documentation/runners/direct/index.html | 18 +-
.../sdks/feature-comparison/index.html | 1 +
.../documentation/sdks/java-extensions/index.html | 1 +
.../documentation/sdks/java-thirdparty/index.html | 1 +
content/documentation/sdks/java/index.html | 1 +
content/documentation/sdks/java/nexmark/index.html | 1 +
.../documentation/sdks/python-custom-io/index.html | 1 +
.../sdks/python-pipeline-dependencies/index.html | 1 +
.../index.html | 257 +++++++++++-------
.../sdks/python-type-safety/index.html | 1 +
content/documentation/sdks/python/index.html | 30 +-
content/get-started/quickstart-java/index.html | 4 +-
content/get-started/quickstart-py/index.html | 55 +++-
content/get-started/wordcount-example/index.html | 302 +++++++++++++--------
src/_includes/section-menu/sdks.html | 1 +
src/documentation/programming-guide.md | 115 ++++----
src/documentation/runners/dataflow.md | 71 +++--
src/documentation/runners/direct.md | 12 +-
src/documentation/sdks/python-streaming.md | 187 +++++++++++++
src/documentation/sdks/python.md | 20 +-
src/get-started/quickstart-java.md | 3 +-
src/get-started/quickstart-py.md | 54 +++-
src/get-started/wordcount-example.md | 295 +++++++++++++-------
26 files changed, 1144 insertions(+), 536 deletions(-)
copy content/documentation/sdks/{java-thirdparty => python-streaming}/index.html (54%)
create mode 100644 src/documentation/sdks/python-streaming.md
--
To stop receiving notification emails like this one, please contact
mergebot-role@apache.org.
[beam-site] 01/01: Prepare repository for deployment.
Posted by me...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mergebot-role pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/beam-site.git
commit ae67badb85cdf9af320eddb1d2a6745b725516dd
Author: Mergebot <me...@apache.org>
AuthorDate: Mon May 7 10:11:23 2018 -0700
Prepare repository for deployment.
---
content/documentation/dsls/sql/index.html | 1 +
content/documentation/programming-guide/index.html | 147 ++++++----
content/documentation/runners/dataflow/index.html | 100 ++++---
content/documentation/runners/direct/index.html | 18 +-
.../sdks/feature-comparison/index.html | 1 +
.../documentation/sdks/java-extensions/index.html | 1 +
.../documentation/sdks/java-thirdparty/index.html | 1 +
content/documentation/sdks/java/index.html | 1 +
content/documentation/sdks/java/nexmark/index.html | 1 +
.../documentation/sdks/python-custom-io/index.html | 1 +
.../sdks/python-pipeline-dependencies/index.html | 1 +
.../index.html | 257 +++++++++++-------
.../sdks/python-type-safety/index.html | 1 +
content/documentation/sdks/python/index.html | 30 +-
content/get-started/quickstart-java/index.html | 4 +-
content/get-started/quickstart-py/index.html | 55 +++-
content/get-started/wordcount-example/index.html | 302 +++++++++++++--------
17 files changed, 588 insertions(+), 334 deletions(-)
diff --git a/content/documentation/dsls/sql/index.html b/content/documentation/dsls/sql/index.html
index 8ba21e5..e433dc8 100644
--- a/content/documentation/dsls/sql/index.html
+++ b/content/documentation/dsls/sql/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/programming-guide/index.html b/content/documentation/programming-guide/index.html
index 5849cf8..5a47764 100644
--- a/content/documentation/programming-guide/index.html
+++ b/content/documentation/programming-guide/index.html
@@ -2904,11 +2904,6 @@ with a <code class="highlighter-rouge">DoFn</code> to attach the timestamps to e
<h2 id="triggers">8. Triggers</h2>
-<blockquote>
- <p><strong>NOTE:</strong> This content applies only to the Beam SDK for Java. The Beam SDK for
-Python does not support triggers.</p>
-</blockquote>
-
<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’s default windowing configuration and <a href="#default-trigger">default
@@ -2976,12 +2971,12 @@ trigger emits the contents of a window after the
<a href="#watermarks-and-late-data">watermark</a> passes the end of the window, based on the
timestamps attached to the data elements. The watermark is a global progress
metric, and is Beam’s notion of input completeness within your pipeline at any
-given point. <code class="highlighter-rouge">AfterWatermark.pastEndOfWindow()</code> <em>only</em> fires when the watermark
-passes the end of the window.</p>
+given point. <span class="language-java"><code class="highlighter-rouge">AfterWatermark.pastEndOfWindow()</code></span>
+<span class="language-py"><code class="highlighter-rouge">AfterWatermark</code></span> <em>only</em> fires when the
+watermark passes the end of the window.</p>
-<p>In addition, you can use <code class="highlighter-rouge">.withEarlyFirings(trigger)</code> and
-<code class="highlighter-rouge">.withLateFirings(trigger)</code> to configure triggers that fire if your pipeline
-receives data before or after the end of the window.</p>
+<p>In addition, you can configure triggers that fire if your pipeline receives data
+before or after the end of the window.</p>
<p>The following example shows a billing scenario, and uses both early and late
firings:</p>
@@ -2997,7 +2992,9 @@ firings:</p>
<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="mi">1</span><span class="o">))</span>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not support triggers.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="n">AfterWatermark</span><span class="p">(</span>
+ <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>
@@ -3016,9 +3013,10 @@ modifying this behavior.</p>
<h3 id="processing-time-triggers">8.2. Processing time triggers</h3>
<p>The <code class="highlighter-rouge">AfterProcessingTime</code> trigger operates on <em>processing time</em>. For example,
-the <code class="highlighter-rouge">AfterProcessingTime.pastFirstElementInPane() </code> trigger emits a window after
-a certain amount of processing time has passed since data was received. The
-processing time is determined by the system clock, rather than the data
+the <span class="language-java"><code class="highlighter-rouge">AfterProcessingTime.pastFirstElementInPane()</code></span>
+<span class="language-py"><code class="highlighter-rouge">AfterProcessingTime</code></span> trigger emits a window
+after a certain amount of processing time has passed since data was received.
+The processing time is determined by the system clock, rather than the data
element’s timestamp.</p>
<p>The <code class="highlighter-rouge">AfterProcessingTime</code> trigger is useful for triggering early results from a
@@ -3027,26 +3025,39 @@ window.</p>
<h3 id="data-driven-triggers">8.3. Data-driven triggers</h3>
-<p>Beam provides one data-driven trigger, <code class="highlighter-rouge">AfterPane.elementCountAtLeast()</code>. This
-trigger works on an element count; it fires after the current pane has collected
-at least <em>N</em> elements. This allows a window to emit early results (before all
-the data has accumulated), which can be particularly useful if you are using a
-single global window.</p>
-
-<p>It is important to note that if, for example, you use <code class="highlighter-rouge">.elementCountAtLeast(50)</code>
-and only 32 elements arrive, those 32 elements sit around forever. If the 32
-elements are important to you, consider using <a href="#composite-triggers">composite
-triggers</a> to combine multiple conditions. This allows you
-to specify multiple firing conditions such as “fire either when I receive 50
-elements, or every 1 second”.</p>
+<p>Beam provides one data-driven trigger,
+<span class="language-java"><code class="highlighter-rouge">AfterPane.elementCountAtLeast()</code></span>
+<span class="language-py"><code class="highlighter-rouge">AfterCount</code></span>. This trigger works on an element
+count; it fires after the current pane has collected at least <em>N</em> elements. This
+allows a window to emit early results (before all the data has accumulated),
+which can be particularly useful if you are using a single global window.</p>
+
+<p>It is important to note that if, for example, you specify
+<span class="language-java"><code class="highlighter-rouge">.elementCountAtLeast(50)</code></span>
+<span class="language-py">AfterCount(50)</span> and only 32 elements arrive,
+those 32 elements sit around forever. If the 32 elements are important to you,
+consider using <a href="#composite-triggers">composite triggers</a> to combine multiple
+conditions. This allows you to specify multiple firing conditions such as “fire
+either when I receive 50 elements, or every 1 second”.</p>
<h3 id="setting-a-trigger">8.4. Setting a trigger</h3>
-<p>When you set a windowing function for a <code class="highlighter-rouge">PCollection</code> by using the <code class="highlighter-rouge">Window</code>
+<p>When you set a windowing function for a <code class="highlighter-rouge">PCollection</code> by using the
+<span class="language-java"><code class="highlighter-rouge">Window</code></span><span class="language-py"><code class="highlighter-rouge">WindowInto</code></span>
transform, you can also specify a trigger.</p>
-<p>You set the trigger(s) for a <code class="highlighter-rouge">PCollection</code> by invoking the method
-<code class="highlighter-rouge">.triggering()</code> on the result of your <code class="highlighter-rouge">Window.into()</code> transform, as follows:</p>
+<p class="language-java">You set the trigger(s) for a <code class="highlighter-rouge">PCollection</code> by invoking the method
+<code class="highlighter-rouge">.triggering()</code> on the result of your <code class="highlighter-rouge">Window.into()</code> transform. This code
+sample sets a time-based trigger for a <code class="highlighter-rouge">PCollection</code>, which emits results one
+minute after the first element in that window has been processed. The last line
+in the code sample, <code class="highlighter-rouge">.discardingFiredPanes()</code>, sets the window’s <strong>accumulation
+mode</strong>.</p>
+
+<p class="language-py">You set the trigger(s) for a <code class="highlighter-rouge">PCollection</code> by setting the <code class="highlighter-rouge">trigger</code> parameter
+when you use the <code class="highlighter-rouge">WindowInto</code> transform. This code sample sets a time-based
+trigger for a <code class="highlighter-rouge">PCollection</code>, which emits results one minute after the first
+element in that window has been processed. The <code class="highlighter-rouge">accumulation_mode</code> parameter
+sets the window’s <strong>accumulation mode</strong>.</p>
<div class="language-java highlighter-rouge"><pre class="highlight"><code> <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>
<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="mi">1</span><span class="o">,</span> <span class="n">TimeUnit</span><span class="o">.</span><span clas [...]
@@ -3055,15 +3066,13 @@ transform, you can also specify a trigger.</p>
<span class="o">.</span><span class="na">discardingFiredPanes</span><span class="o">());</span>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not support triggers.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="n">pcollection</span> <span class="o">|</span> <span class="n">WindowInto</span><span class="p">(</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">10</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>
-<p>This code sample sets a time-based trigger for a <code class="highlighter-rouge">PCollection</code>, which emits
-results one minute after the first element in that window has been processed.
-The last line in the code sample, <code class="highlighter-rouge">.discardingFiredPanes()</code>, is the window’s
-<strong>accumulation mode</strong>.</p>
-
<h4 id="window-accumulation-modes">8.4.1. Window accumulation modes</h4>
<p>When you specify a trigger, you must also set the the window’s <strong>accumulation
@@ -3072,10 +3081,15 @@ 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
<em>discards</em> them.</p>
-<p>To set a window to accumulate the panes that are produced when the trigger
+<p class="language-java">To set a window to accumulate the panes that are produced when the trigger
fires, invoke<code class="highlighter-rouge">.accumulatingFiredPanes()</code> when you set the trigger. To set a
window to discard fired panes, invoke <code class="highlighter-rouge">.discardingFiredPanes()</code>.</p>
+<p class="language-py">To set a window to accumulate the panes that are produced when the trigger
+fires, set the <code class="highlighter-rouge">accumulation_mode</code> parameter to <code class="highlighter-rouge">ACCUMULATING</code> when you set the
+trigger. To set a window to discard fired panes, set <code class="highlighter-rouge">accumulation_mode</code> to
+<code class="highlighter-rouge">DISCARDING</code>.</p>
+
<p>Let’s look an example that uses a <code class="highlighter-rouge">PCollection</code> with fixed-time windowing and a
data-based trigger. This is something you might do if, for example, each window
represented a ten-minute running average, but you wanted to display the current
@@ -3096,9 +3110,9 @@ we’ll assume that the events all arrive in the pipeline in order.</p>
<h5 id="accumulating-mode">8.4.1.1. Accumulating mode</h5>
-<p>If our trigger is set to <code class="highlighter-rouge">.accumulatingFiredPanes</code>, the trigger emits the
-following values each time it fires. Keep in mind that the trigger fires every
-time three elements arrive:</p>
+<p>If our trigger is set to accumulating mode, the trigger emits the following
+values each time it fires. Keep in mind that the trigger fires every time three
+elements arrive:</p>
<div class="highlighter-rouge"><pre class="highlight"><code> First trigger firing: [5, 8, 3]
Second trigger firing: [5, 8, 3, 15, 19, 23]
@@ -3108,8 +3122,8 @@ time three elements arrive:</p>
<h5 id="discarding-mode">8.4.1.2. Discarding mode</h5>
-<p>If our trigger is set to <code class="highlighter-rouge">.discardingFiredPanes</code>, the trigger emits the
-following values on each firing:</p>
+<p>If our trigger is set to discarding mode, the trigger emits the following values
+on each firing:</p>
<div class="highlighter-rouge"><pre class="highlight"><code> First trigger firing: [5, 8, 3]
Second trigger firing: [15, 19, 23]
@@ -3119,6 +3133,10 @@ following values on each firing:</p>
<h4 id="handling-late-data">8.4.2. Handling late data</h4>
+<blockquote>
+ <p>The Beam SDK for Python does not currently support allowed lateness.</p>
+</blockquote>
+
<p>If you want your pipeline to process data that arrives after the watermark
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
@@ -3135,7 +3153,7 @@ windowing function:</p>
<span class="o">.</span><span class="na">withAllowedLateness</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="mi">30</span><span class="o">));</span>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not support triggers.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not currently support allowed lateness.</span>
</code></pre>
</div>
@@ -3176,26 +3194,37 @@ multiple triggers.</li>
final time and never fire again.</li>
</ul>
-<h4 id="afterwatermark-pastendofwindow">8.5.2. Composition with AfterWatermark.pastEndOfWindow</h4>
+<h4 id="composite-afterwatermark">8.5.2. Composition with AfterWatermark</h4>
<p>Some of the most useful composite triggers fire a single time when Beam
estimates that all the data has arrived (i.e. when the watermark passes the end
of the window) combined with either, or both, of the following:</p>
<ul>
- <li>Speculative firings that precede the watermark passing the end of the window
-to allow faster processing of partial results.</li>
- <li>Late firings that happen after the watermark passes the end of the window,
-to allow for handling late-arriving data</li>
+ <li>
+ <p>Speculative firings that precede the watermark passing the end of the window
+to allow faster processing of partial results.</p>
+ </li>
+ <li>
+ <p>Late firings that happen after the watermark passes the end of the window,
+to allow for handling late-arriving data</p>
+ </li>
</ul>
-<p>You can express this pattern using <code class="highlighter-rouge">AfterWatermark.pastEndOfWindow</code>. For
-example, the following example trigger code fires on the following conditions:</p>
+<p>You can express this pattern using <code class="highlighter-rouge">AfterWatermark</code>. For example, the following
+example trigger code fires on the following conditions:</p>
<ul>
- <li>On Beam’s estimate that all the data has arrived (the watermark passes the
-end of the window)</li>
- <li>Any time late data arrives, after a ten-minute delay</li>
+ <li>
+ <p>On Beam’s estimate that all the data has arrived (the watermark passes the
+end of the window)</p>
+ </li>
+ <li>
+ <p>Any time late data arrives, after a ten-minute delay</p>
+ </li>
+</ul>
+
+<ul class="language-java">
<li>After two days, we assume no more data of interest will arrive, and the
trigger stops executing</li>
</ul>
@@ -3210,7 +3239,11 @@ trigger stops executing</li>
<span class="o">.</span><span class="na">withAllowedLateness</span><span class="o">(</span><span class="n">Duration</span><span class="o">.</span><span class="na">standardDays</span><span class="o">(</span><span class="mi">2</span><span class="o">)));</span>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not support triggers.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="n">pcollection</span> <span class="o">|</span> <span class="n">WindowInto</span><span class="p">(</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">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">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>
@@ -3225,7 +3258,13 @@ elements, or after a minute.</p>
<span class="n">AfterProcessingTime</span><span class="o">.</span><span class="na">pastFirstElementInPane</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="mi">1</span><span class="o">))))</span>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># The Beam SDK for Python does not support triggers.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="n">pcollection</span> <span class="o">|</span> <span class="n">WindowInto</span><span class="p">(</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">Repeatedly</span><span class="p">(</span>
+ <span class="n">AfterAny</span><span class="p">(</span>
+ <span class="n">AfterCount</span><span class="p">(</span><span class="mi">100</span><span class="p">),</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>
diff --git a/content/documentation/runners/dataflow/index.html b/content/documentation/runners/dataflow/index.html
index 9538dbf..9cad557 100644
--- a/content/documentation/runners/dataflow/index.html
+++ b/content/documentation/runners/dataflow/index.html
@@ -102,17 +102,16 @@
<ul class="nav">
- <li><a href="#cloud-dataflow-runner-prerequisites-and-setup">Cloud Dataflow Runner prerequisites and setup</a>
+ <li><a href="#setup">Cloud Dataflow Runner prerequisites and setup</a>
<ul>
- <li><a href="#specify-your-dependency">Specify your dependency</a></li>
+ <li><a href="#dependency">Specify your dependency</a></li>
<li><a href="#self-executing-jar">Self executing JAR</a></li>
- <li><a href="#authentication">Authentication</a></li>
</ul>
</li>
- <li><a href="#pipeline-options-for-the-cloud-dataflow-runner">Pipeline options for the Cloud Dataflow Runner</a></li>
- <li><a href="#additional-information-and-caveats">Additional information and caveats</a>
+ <li><a href="#pipeline-options">Pipeline options for the Cloud Dataflow Runner</a></li>
+ <li><a href="#additional-info">Additional information and caveats</a>
<ul>
- <li><a href="#monitoring-your-job">Monitoring your job</a></li>
+ <li><a href="#monitoring">Monitoring your job</a></li>
<li><a href="#blocking-execution">Blocking Execution</a></li>
<li><a href="#streaming-execution">Streaming Execution</a></li>
</ul>
@@ -145,45 +144,25 @@
<p>The <a href="/documentation/runners/capability-matrix/">Beam Capability Matrix</a> documents the supported capabilities of the Cloud Dataflow Runner.</p>
-<h2 id="cloud-dataflow-runner-prerequisites-and-setup">Cloud Dataflow Runner prerequisites and setup</h2>
-<p>To use the Cloud Dataflow Runner, you must complete the following setup:</p>
+<h2 id="setup">Cloud Dataflow Runner prerequisites and setup</h2>
+
+<p>To use the Cloud Dataflow Runner, you must complete the setup in the <em>Before you
+begin</em> section of the <a href="https://cloud.google.com/dataflow/docs/quickstarts">Cloud Dataflow quickstart</a>
+for your chosen language.</p>
<ol>
- <li>
- <p>Select or create a Google Cloud Platform Console project.</p>
- </li>
- <li>
- <p>Enable billing for your project.</p>
- </li>
- <li>
- <p>Enable the required Google Cloud APIs: Cloud Dataflow, Compute Engine,
+ <li>Select or create a Google Cloud Platform Console project.</li>
+ <li>Enable billing for your project.</li>
+ <li>Enable the required Google Cloud APIs: Cloud Dataflow, Compute Engine,
Stackdriver Logging, Cloud Storage, Cloud Storage JSON, and Cloud Resource
Manager. You may need to enable additional APIs (such as BigQuery, Cloud
-Pub/Sub, or Cloud Datastore) if you use them in your pipeline code.</p>
- </li>
- <li>
- <p>Install the Google Cloud SDK.</p>
- </li>
- <li>
- <p>Create a Cloud Storage bucket.</p>
- <ul>
- <li>In the Google Cloud Platform Console, go to the Cloud Storage browser.</li>
- <li>Click <strong>Create bucket</strong>.</li>
- <li>In the <strong>Create bucket</strong> dialog, specify the following attributes:
- <ul>
- <li><em>Name</em>: A unique bucket name. Do not include sensitive information in the bucket name, as the bucket namespace is global and publicly visible.</li>
- <li><em>Storage class</em>: Multi-Regional</li>
- <li><em>Location</em>: Choose your desired location</li>
- </ul>
- </li>
- <li>Click <strong>Create</strong>.</li>
- </ul>
- </li>
+Pub/Sub, or Cloud Datastore) if you use them in your pipeline code.</li>
+ <li>Authenticate with Google Cloud Platform.</li>
+ <li>Install the Google Cloud SDK.</li>
+ <li>Create a Cloud Storage bucket.</li>
</ol>
-<p>For more information, see the <em>Before you begin</em> section of the <a href="https://cloud.google.com/dataflow/docs/quickstarts">Cloud Dataflow quickstarts</a>.</p>
-
-<h3 id="specify-your-dependency">Specify your dependency</h3>
+<h3 id="dependency">Specify your dependency</h3>
<p><span class="language-java">When using Java, you must specify your dependency on the Cloud Dataflow Runner in your <code class="highlighter-rouge">pom.xml</code>.</span></p>
<div class="language-java highlighter-rouge"><pre class="highlight"><code><span class="o"><</span><span class="n">dependency</span><span class="o">></span>
@@ -246,15 +225,7 @@ Pub/Sub, or Cloud Datastore) if you use them in your pipeline code.</p>
</code></pre>
</div>
-<h3 id="authentication">Authentication</h3>
-
-<p>Before running your pipeline, you must authenticate with the Google Cloud Platform. Run the following command to get <a href="https://developers.google.com/identity/protocols/application-default-credentials">Application Default Credentials</a>.</p>
-
-<div class="highlighter-rouge"><pre class="highlight"><code>gcloud auth application-default login
-</code></pre>
-</div>
-
-<h2 id="pipeline-options-for-the-cloud-dataflow-runner">Pipeline options for the Cloud Dataflow Runner</h2>
+<h2 id="pipeline-options">Pipeline options for the Cloud Dataflow Runner</h2>
<p><span class="language-java">When executing your pipeline with the Cloud Dataflow Runner (Java), consider these common pipeline options.</span>
<span class="language-py">When executing your pipeline with the Cloud Dataflow Runner (Python), consider these common pipeline options.</span></p>
@@ -278,8 +249,7 @@ Pub/Sub, or Cloud Datastore) if you use them in your pipeline code.</p>
<td>If not set, defaults to the default project in the current environment. The default project is set via <code>gcloud</code>.</td>
</tr>
-<!-- Only show for Java -->
-<tr class="language-java">
+<tr>
<td><code>streaming</code></td>
<td>Whether streaming mode is enabled or disabled; <code>true</code> if enabled. Set to <code>true</code> if running pipelines with unbounded <code>PCollection</code>s.</td>
<td><code>false</code></td>
@@ -340,9 +310,9 @@ Pub/Sub, or Cloud Datastore) if you use them in your pipeline code.</p>
<span class="language-py"><a href="/documentation/sdks/pydoc/2.4.0/apache_beam.options.pipeline_options.html#apache_beam.options.pipeline_options.PipelineOptions"><code class="highlighter-rouge">PipelineOptions</code></a></span>
interface (and any subinterfaces) for additional pipeline configuration options.</p>
-<h2 id="additional-information-and-caveats">Additional information and caveats</h2>
+<h2 id="additional-info">Additional information and caveats</h2>
-<h3 id="monitoring-your-job">Monitoring your job</h3>
+<h3 id="monitoring">Monitoring your job</h3>
<p>While your pipeline executes, you can monitor the job’s progress, view details on execution, and receive updates on the pipeline’s results by using the <a href="https://cloud.google.com/dataflow/pipelines/dataflow-monitoring-intf">Dataflow Monitoring Interface</a> or the <a href="https://cloud.google.com/dataflow/pipelines/dataflow-command-line-intf">Dataflow Command-line Interface</a>.</p>
@@ -352,8 +322,30 @@ interface (and any subinterfaces) for additional pipeline configuration options.
<h3 id="streaming-execution">Streaming Execution</h3>
-<p><span class="language-java">If your pipeline uses an unbounded data source or sink, you must set the <code class="highlighter-rouge">streaming</code> option to <code class="highlighter-rouge">true</code>.</span>
-<span class="language-py">The Beam SDK for Python does not currently support streaming pipelines.</span></p>
+<p>If your pipeline uses an unbounded data source or sink, you must set the <code class="highlighter-rouge">streaming</code> option to <code class="highlighter-rouge">true</code>.</p>
+
+<p>When using streaming execution, keep the following considerations in mind.</p>
+
+<ol>
+ <li>
+ <p>Streaming pipelines do not terminate unless explicitly cancelled by the user.
+You can cancel your streaming job from the <a href="https://cloud.google.com/dataflow/pipelines/stopping-a-pipeline">Dataflow Monitoring Interface</a>
+or with the <a href="https://cloud.google.com/dataflow/pipelines/dataflow-command-line-intf">Dataflow Command-line Interface</a>
+(<a href="https://cloud.google.com/sdk/gcloud/reference/dataflow/jobs/cancel">gcloud dataflow jobs cancel</a>
+command).</p>
+ </li>
+ <li>
+ <p>Streaming jobs use a Google Compute Engine <a href="https://cloud.google.com/compute/docs/machine-types">machine type</a>
+of <code class="highlighter-rouge">n1-standard-2</code> or higher by default. You must not override this, as
+<code class="highlighter-rouge">n1-standard-2</code> is the minimum required machine type for running streaming
+jobs.</p>
+ </li>
+ <li>
+ <p>Streaming execution <a href="https://cloud.google.com/dataflow/pricing">pricing</a>
+differs from batch execution.</p>
+ </li>
+</ol>
+
</div>
</div>
diff --git a/content/documentation/runners/direct/index.html b/content/documentation/runners/direct/index.html
index 8794ef4..5556376 100644
--- a/content/documentation/runners/direct/index.html
+++ b/content/documentation/runners/direct/index.html
@@ -108,7 +108,12 @@
</ul>
</li>
<li><a href="#pipeline-options-for-the-direct-runner">Pipeline options for the Direct Runner</a></li>
- <li><a href="#additional-information-and-caveats">Additional information and caveats</a></li>
+ <li><a href="#additional-information-and-caveats">Additional information and caveats</a>
+ <ul>
+ <li><a href="#memory-considerations">Memory considerations</a></li>
+ <li><a href="#streaming-execution">Streaming execution</a></li>
+ </ul>
+ </li>
</ul>
@@ -140,10 +145,10 @@
<ul>
<!-- Java specific links -->
<li class="language-java"><a href="/blog/2016/10/20/test-stream.html">Testing Unbounded Pipelines in Apache Beam</a> talks about the use of Java classes <a href="/documentation/sdks/javadoc/2.4.0/index.html?org/apache/beam/sdk/testing/PAssert.html">PAssert</a> and <a href="/documentation/sdks/javadoc/2.4.0/index.html?org/apache/beam/sdk/testing/TestStream.html">TestStream</a> to test your pipelines.</li>
- <li class="language-java">The <a href="/get-started/wordcount-example/#testing-your-pipeline-via-passert">Apache Beam WordCount Example</a> contains an example of logging and testing a pipeline with <a href="/documentation/sdks/javadoc/2.4.0/index.html?org/apache/beam/sdk/testing/PAssert.html"><code>PAssert</code></a>.</li>
+ <li class="language-java">The <a href="/get-started/wordcount-example/#testing-your-pipeline-with-asserts">Apache Beam WordCount Walkthrough</a> contains an example of logging and testing a pipeline with <a href="/documentation/sdks/javadoc/2.4.0/index.html?org/apache/beam/sdk/testing/PAssert.html">PAssert</a>.</li>
<!-- Python specific links -->
- <li class="language-py">You can use <a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/testing/util.py#L76">assert_that</a> to test your pipeline. The Python <a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount_debugging.py">WordCount Debugging Example</a> contains an example of logging and testing with <code>assert_that</code>.</li>
+ <li class="language-py">The <a href="/get-started/wordcount-example/#testing-your-pipeline-with-asserts">Apache Beam WordCount Walkthrough</a> contains an example of logging and testing a pipeline with <a href="/documentation/sdks/pydoc/2.4.0/apache_beam.testing.util.html#apache_beam.testing.util.assert_that">assert_that</a>.</li>
</ul>
<h2 id="direct-runner-prerequisites-and-setup">Direct Runner prerequisites and setup</h2>
@@ -173,8 +178,15 @@ interface for defaults and additional pipeline configuration options.</p>
<h2 id="additional-information-and-caveats">Additional information and caveats</h2>
+<h3 id="memory-considerations">Memory considerations</h3>
+
<p>Local execution is limited by the memory available in your local environment. It is highly recommended that you run your pipeline with data sets small enough to fit in local memory. You can create a small in-memory data set using a <span class="language-java"><a href="/documentation/sdks/javadoc/2.4.0/index.html?org/apache/beam/sdk/transforms/Create.html"><code class="highlighter-rouge">Create</code></a></span><span class="language-py"><a href="https://github.com/apache/beam/blob/mast [...]
+<h3 id="streaming-execution">Streaming execution</h3>
+
+<p>If your pipeline uses an unbounded data source or sink, you must set the <code class="highlighter-rouge">streaming</code> option to <code class="highlighter-rouge">true</code>.</p>
+
+
</div>
</div>
<footer class="footer">
diff --git a/content/documentation/sdks/feature-comparison/index.html b/content/documentation/sdks/feature-comparison/index.html
index b327c18..efd8734 100644
--- a/content/documentation/sdks/feature-comparison/index.html
+++ b/content/documentation/sdks/feature-comparison/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/java-extensions/index.html b/content/documentation/sdks/java-extensions/index.html
index 4e6c847..48bf68e 100644
--- a/content/documentation/sdks/java-extensions/index.html
+++ b/content/documentation/sdks/java-extensions/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/java-thirdparty/index.html b/content/documentation/sdks/java-thirdparty/index.html
index 010722f..b1ef4eb 100644
--- a/content/documentation/sdks/java-thirdparty/index.html
+++ b/content/documentation/sdks/java-thirdparty/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/java/index.html b/content/documentation/sdks/java/index.html
index 38f7f63..f40843e 100644
--- a/content/documentation/sdks/java/index.html
+++ b/content/documentation/sdks/java/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/java/nexmark/index.html b/content/documentation/sdks/java/nexmark/index.html
index fb3b1e2..04b797b 100644
--- a/content/documentation/sdks/java/nexmark/index.html
+++ b/content/documentation/sdks/java/nexmark/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/python-custom-io/index.html b/content/documentation/sdks/python-custom-io/index.html
index d916d56..0bec3a5 100644
--- a/content/documentation/sdks/python-custom-io/index.html
+++ b/content/documentation/sdks/python-custom-io/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/python-pipeline-dependencies/index.html b/content/documentation/sdks/python-pipeline-dependencies/index.html
index 0bdee1c..3db92cb 100644
--- a/content/documentation/sdks/python-pipeline-dependencies/index.html
+++ b/content/documentation/sdks/python-pipeline-dependencies/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/java-thirdparty/index.html b/content/documentation/sdks/python-streaming/index.html
similarity index 54%
copy from content/documentation/sdks/java-thirdparty/index.html
copy to content/documentation/sdks/python-streaming/index.html
index 010722f..d5f5bc0 100644
--- a/content/documentation/sdks/java-thirdparty/index.html
+++ b/content/documentation/sdks/python-streaming/index.html
@@ -4,7 +4,7 @@
<meta charset="utf-8">
<meta http-equiv="X-UA-Compatible" content="IE=edge">
<meta name="viewport" content="width=device-width, initial-scale=1">
- <title>Beam 3rd Party Java Extensions</title>
+ <title>Apache Beam Python Streaming Pipelines</title>
<meta name="description" content="Apache Beam is an open source, unified model and set of language-specific SDKs for defining and executing data processing workflows, and also data ingestion and integration flows, supporting Enterprise Integration Patterns (EIPs) and Domain Specific Languages (DSLs). Dataflow pipelines simplify the mechanics of large-scale batch and streaming data processing and can run on a number of runtimes like Apache Flink, Apache Spark, and Google Cloud Dataflow [...]
">
<link href="https://fonts.googleapis.com/css?family=Roboto:100,300,400" rel="stylesheet">
@@ -15,7 +15,7 @@
<script src="/js/fix-menu.js"></script>
<script src="/js/section-nav.js"></script>
<script src="/js/page-nav.js"></script>
- <link rel="canonical" href="https://beam.apache.org/documentation/sdks/java-thirdparty/" data-proofer-ignore>
+ <link rel="canonical" href="https://beam.apache.org/documentation/sdks/python-streaming/" data-proofer-ignore>
<link rel="shortcut icon" type="image/x-icon" href="/images/favicon.ico">
<link rel="alternate" type="application/rss+xml" title="Apache Beam" href="https://beam.apache.org/feed.xml">
<script>
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
@@ -130,22 +131,13 @@
<ul class="nav">
- <li><a href="#parsing-httpdnginx-access-logs">Parsing HTTPD/NGINX access logs.</a>
+ <li><a href="#why-use-streaming-execution">Why use streaming execution?</a></li>
+ <li><a href="#modifying-a-pipeline-to-use-stream-processing">Modifying a pipeline to use stream processing</a></li>
+ <li><a href="#running-a-streaming-pipeline">Running a streaming pipeline</a></li>
+ <li><a href="#unsupported-features">Unsupported features</a>
<ul>
- <li><a href="#summary">Summary</a></li>
- <li><a href="#project-page">Project page</a></li>
- <li><a href="#license">License</a></li>
- <li><a href="#download">Download</a></li>
- <li><a href="#code-example">Code example</a></li>
- </ul>
- </li>
- <li><a href="#analyzing-the-useragent-string">Analyzing the Useragent string</a>
- <ul>
- <li><a href="#summary-1">Summary</a></li>
- <li><a href="#project-page-1">Project page</a></li>
- <li><a href="#license-1">License</a></li>
- <li><a href="#download-1">Download</a></li>
- <li><a href="#code-example-1">Code example</a></li>
+ <li><a href="#general-beam-features">General Beam features</a></li>
+ <li><a href="#dataflowrunner-specific-features">DataflowRunner specific features</a></li>
</ul>
</li>
</ul>
@@ -154,107 +146,184 @@
</nav>
<div class="body__contained body__section-nav">
- <h1 id="apache-beam-3rd-party-java-extensions">Apache Beam 3rd Party Java Extensions</h1>
+ <h1 id="python-streaming-pipelines">Python Streaming Pipelines</h1>
-<p>These are some of the 3rd party Java libaries that may be useful for specific applications.</p>
+<p>Python streaming pipeline execution is experimentally available (with some
+<a href="#unsupported-features">limitations</a>) starting with Beam SDK version 2.5.0.</p>
-<h2 id="parsing-httpdnginx-access-logs">Parsing HTTPD/NGINX access logs.</h2>
+<h2 id="why-use-streaming-execution">Why use streaming execution?</h2>
-<h3 id="summary">Summary</h3>
-<p>The Apache HTTPD webserver creates logfiles that contain valuable information about the requests that have been done to
-the webserver. The format of these log files is a configuration option in the Apache HTTPD server so parsing this
-into useful data elements is normally very hard to do.</p>
+<p>Beam creates an unbounded PCollection if your pipeline reads from a streaming or
+continously-updating data source (such as Cloud Pub/Sub). A runner must
+process an unbounded PCollection using a streaming job that runs continuously,
+as the entire collection is never available for processing at any one time.
+<a href="/documentation/programming-guide/#size-and-boundedness">Size and boundedness</a>
+has more information about bounded and unbounded collections.</p>
-<p>To solve this problem in an easy way a library was created that works in combination with Apache Beam
-and is capable of doing this for both the Apache HTTPD and NGINX.</p>
+<h2 id="modifying-a-pipeline-to-use-stream-processing">Modifying a pipeline to use stream processing</h2>
+
+<p>To modify a batch pipeline to support streaming, you must make the following
+code changes:</p>
+
+<ul>
+ <li>Use an I/O connector that supports reading from an unbounded source.</li>
+ <li>Use an I/O connector that supports writing to an unbounded source.</li>
+ <li>Choose a <a href="/documentation/programming-guide/index.html#windowing">windowing strategy</a>.</li>
+</ul>
-<p>The basic idea is that the logformat specification is the schema used to create the line.
-This parser is simply initialized with this schema and the list of fields you want to extract.</p>
+<p>The Beam SDK for Python includes two I/O connectors that support unbounded
+PCollections: Google Cloud Pub/Sub (reading and writing) and Google BigQuery
+(writing).</p>
-<h3 id="project-page">Project page</h3>
-<p><a href="https://github.com/nielsbasjes/logparser">https://github.com/nielsbasjes/logparser</a></p>
+<p>The following snippets show the necessary code changes to modify the batch
+WordCount example to support streaming:</p>
-<h3 id="license">License</h3>
-<p>Apache License 2.0</p>
+<p>These batch WordCount snippets are from
+<a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount.py">wordcount.py</a>.
+This code uses the TextIO I/O connector to read from and write to a bounded
+collection.</p>
-<h3 id="download">Download</h3>
-<div class="highlighter-rouge"><pre class="highlight"><code><dependency>
- <groupId>nl.basjes.parse.httpdlog</groupId>
- <artifactId>httpdlog-parser</artifactId>
- <version>5.0</version>
-</dependency>
+<div class="highlighter-rouge"><pre class="highlight"><code> lines = p | 'read' >> ReadFromText(known_args.input)
+ ...
+
+ counts = (lines
+ | 'split' >> (beam.ParDo(WordExtractingDoFn())
+ .with_output_types(six.text_type))
+ | 'pair_with_one' >> beam.Map(lambda x: (x, 1))
+ | 'group' >> beam.GroupByKey()
+ | 'count' >> beam.Map(count_ones))
+ ...
+
+ output = counts | 'format' >> beam.Map(format_result)
+
+ # Write the output using a "Write" transform that has side effects.
+ output | 'write' >> WriteToText(known_args.output)
</code></pre>
</div>
-<h3 id="code-example">Code example</h3>
-
-<p>Assuming a WebEvent class that has a the setters setIP, setQueryImg and setQueryStringValues</p>
-
-<div class="highlighter-rouge"><pre class="highlight"><code>PCollection<WebEvent> filledWebEvents = input
- .apply("Extract Elements from logline",
- ParDo.of(new DoFn<String, WebEvent>() {
- private Parser<WebEvent> parser;
-
- @Setup
- public void setup() throws NoSuchMethodException {
- parser = new HttpdLoglineParser<>(WebEvent.class,
- "%h %l %u %t \"%r\" %>s %b \"%{Referer}i\" \"%{User-Agent}i\" \"%{Cookie}i\"");
- parser.addParseTarget("setIP", "IP:connection.client.host");
- parser.addParseTarget("setQueryImg", "STRING:request.firstline.uri.query.img");
- parser.addParseTarget("setQueryStringValues", "STRING:request.firstline.uri.query.*");
- }
-
- @ProcessElement
- public void processElement(ProcessContext c) throws InvalidDissectorException, MissingDissectorsException, DissectionFailure {
- c.output(parser.parse(c.element()));
- }
- })
- );
+<p>These streaming WordCount snippets are from
+<a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py">streaming_wordcount.py</a>.
+This code uses an I/O connector that reads from and writes to an unbounded
+source (Cloud Pub/Sub) and specifies a fixed windowing strategy.</p>
+
+<div class="highlighter-rouge"><pre class="highlight"><code> lines = p | beam.io.ReadFromPubSub(topic=known_args.input_topic)
+ ...
+
+ counts = (lines
+ | 'split' >> (beam.ParDo(WordExtractingDoFn())
+ .with_output_types(six.text_type))
+ | 'pair_with_one' >> beam.Map(lambda x: (x, 1))
+ | beam.WindowInto(window.FixedWindows(15, 0))
+ | 'group' >> beam.GroupByKey()
+ | 'count' >> beam.Map(count_ones))
+
+ ...
+
+ output = counts | 'format' >> beam.Map(format_result)
+
+ # Write to Pub/Sub
+ output | beam.io.WriteStringsToPubSub(known_args.output_topic)
</code></pre>
</div>
-<h2 id="analyzing-the-useragent-string">Analyzing the Useragent string</h2>
+<h2 id="running-a-streaming-pipeline">Running a streaming pipeline</h2>
-<h3 id="summary-1">Summary</h3>
-<p>Parse and analyze the useragent string and extract as many relevant attributes as possible.</p>
+<p>To run the example streaming WordCount pipeline, you must have a Cloud Pub/Sub
+input topic and output topic. To create, subscribe to, and pull from a topic for
+testing purposes, you can use the commands in the <a href="https://cloud.google.com/pubsub/docs/quickstart-cli">Cloud Pub/Sub quickstart</a>.</p>
-<h3 id="project-page-1">Project page</h3>
-<p><a href="https://github.com/nielsbasjes/yauaa">https://github.com/nielsbasjes/yauaa</a></p>
+<p>The following simple bash script feeds lines of an input text file to your input
+topic:</p>
+
+<div class="highlighter-rouge"><pre class="highlight"><code>cat <YOUR_LOCAL_TEXT_FILE> | while read line; do gcloud pubsub topics publish <YOUR_INPUT_TOPIC_NAME> “ $line”; done
+</code></pre>
+</div>
-<h3 id="license-1">License</h3>
-<p>Apache License 2.0</p>
+<p>Alternately, you can read from a publicly available Cloud Pub/Sub stream, such
+as <code class="highlighter-rouge">projects/pubsub-public-data/topics/taxirides-realtime</code>. However, you must
+create your own output topic to test writes.</p>
-<h3 id="download-1">Download</h3>
-<div class="highlighter-rouge"><pre class="highlight"><code><dependency>
- <groupId>nl.basjes.parse.useragent</groupId>
- <artifactId>yauaa-beam</artifactId>
- <version>4.2</version>
-</dependency>
+<p>The following commands run the
+<a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py">streaming_wordcount.py</a>
+example streaming pipeline. Specify your Cloud Pub/Sub project and input topic
+(<code class="highlighter-rouge">--input_topic</code>), output Cloud Pub/Sub project and topic (<code class="highlighter-rouge">--output_topic</code>).</p>
+
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code># DirectRunner requires the --streaming option
+python -m apache_beam.examples.streaming_wordcount \
+ --input_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_INPUT_TOPIC" \
+ --output_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_OUTPUT_TOPIC" \
+ --streaming
</code></pre>
</div>
-<h3 id="code-example-1">Code example</h3>
-<div class="highlighter-rouge"><pre class="highlight"><code>PCollection<WebEvent> filledWebEvents = input
- .apply("Extract Elements from Useragent",
- ParDo.of(new UserAgentAnalysisDoFn<WebEvent>() {
- @Override
- public String getUserAgentString(WebEvent record) {
- return record.useragent;
- }
-
- @YauaaField("DeviceClass")
- public void setDC(WebEvent record, String value) {
- record.deviceClass = value;
- }
-
- @YauaaField("AgentNameVersion")
- public void setANV(WebEvent record, String value) {
- record.agentNameVersion = value;
- }
- }));
+<div class="runner-apex highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
</code></pre>
</div>
+<div class="runner-flink-local highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-cluster highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-spark highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
+pip install apache-beam[gcp]
+
+# DataflowRunner requires the --streaming option
+python -m apache_beam.examples.streaming_wordcount \
+ --runner DataflowRunner \
+ --project YOUR_GCP_PROJECT \
+ --temp_location gs://YOUR_GCS_BUCKET/tmp/ \
+ --input_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_INPUT_TOPIC" \
+ --output_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_OUTPUT_TOPIC" \
+ --streaming
+</code></pre>
+</div>
+
+<p>Check your runner’s documentation for any additional runner-specific information
+about executing streaming pipelines:</p>
+
+<ul>
+ <li><a href="/documentation/runners/direct/#streaming-execution">DirectRunner streaming execution</a></li>
+ <li><a href="/documentation/runners/dataflow/#streaming-execution">DataflowRunner streaming execution</a></li>
+</ul>
+
+<h2 id="unsupported-features">Unsupported features</h2>
+
+<p>Python streaming execution does not currently support the following features.</p>
+
+<h3 id="general-beam-features">General Beam features</h3>
+
+<p>These unsupported Beam features apply to all runners.</p>
+
+<ul>
+ <li><code class="highlighter-rouge">State</code> and <code class="highlighter-rouge">Timers</code> APIs</li>
+ <li>Custom source API</li>
+ <li>Splittable <code class="highlighter-rouge">DoFn</code> API</li>
+ <li>Handling of late data</li>
+ <li>User-defined custom <code class="highlighter-rouge">WindowFn</code></li>
+</ul>
+
+<h3 id="dataflowrunner-specific-features">DataflowRunner specific features</h3>
+
+<p>Additionally, <code class="highlighter-rouge">DataflowRunner</code> does not currently support the following Cloud
+Dataflow specific features with Python streaming execution.</p>
+
+<ul>
+ <li>Streaming autoscaling</li>
+ <li>Updating existing pipelines</li>
+ <li>Cloud Dataflow Templates</li>
+ <li>Some monitoring features, such as msec counters, display data, metrics, and
+element counts for transforms. However, logging, watermarks, and element
+counts for sources are supported.</li>
+</ul>
+
</div>
</div>
diff --git a/content/documentation/sdks/python-type-safety/index.html b/content/documentation/sdks/python-type-safety/index.html
index b6aab73..e804af2 100644
--- a/content/documentation/sdks/python-type-safety/index.html
+++ b/content/documentation/sdks/python-type-safety/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
diff --git a/content/documentation/sdks/python/index.html b/content/documentation/sdks/python/index.html
index fffa8d8..22ac521 100644
--- a/content/documentation/sdks/python/index.html
+++ b/content/documentation/sdks/python/index.html
@@ -109,6 +109,7 @@
width="14" height="14"
alt="External link."></a>
</li>
+ <li><a href="/documentation/sdks/python-streaming/">Python streaming pipelines</a></li>
<li><a href="/documentation/sdks/python-type-safety/">Ensuring Python type safety</a></li>
<li><a href="/documentation/sdks/python-pipeline-dependencies/">Managing pipeline dependencies</a></li>
<li><a href="/documentation/sdks/python-custom-io/">Creating new sources and sinks</a></li>
@@ -130,10 +131,11 @@
<ul class="nav">
- <li><a href="#get-started-with-the-python-sdk">Get Started with the Python SDK</a></li>
- <li><a href="#python-type-safety">Python Type Safety</a></li>
- <li><a href="#managing-python-pipeline-dependencies">Managing Python Pipeline Dependencies</a></li>
- <li><a href="#creating-new-sources-and-sinks">Creating New Sources and Sinks</a></li>
+ <li><a href="#get-started-with-the-python-sdk">Get started with the Python SDK</a></li>
+ <li><a href="#python-streaming-pipelines">Python streaming pipelines</a></li>
+ <li><a href="#python-type-safety">Python type safety</a></li>
+ <li><a href="#managing-python-pipeline-dependencies">Managing Python pipeline dependencies</a></li>
+ <li><a href="#creating-new-sources-and-sinks">Creating new sources and Sinks</a></li>
</ul>
@@ -142,23 +144,29 @@
<div class="body__contained body__section-nav">
<h1 id="apache-beam-python-sdk">Apache Beam Python SDK</h1>
-<p>The Python SDK for Apache Beam provides a simple, powerful API for building batch data processing pipelines in Python.</p>
+<p>The Python SDK for Apache Beam provides a simple, powerful API for building batch and streaming data processing pipelines.</p>
-<h2 id="get-started-with-the-python-sdk">Get Started with the Python SDK</h2>
+<h2 id="get-started-with-the-python-sdk">Get started with the Python SDK</h2>
-<p>Get started with the <a href="/documentation/programming-guide">Beam Programming Guide</a> to learn the basic concepts that apply to all SDKs in Beam. Then, follow the <a href="/get-started/quickstart-py">Beam Python SDK Quickstart</a> to set up your Python development environment, get the Beam SDK for Python, and run an example pipeline.</p>
+<p>Get started with the <a href="/get-started/quickstart-py">Beam Python SDK quickstart</a> to set up your Python development environment, get the Beam SDK for Python, and run an example pipeline. Then, read through the <a href="/documentation/programming-guide">Beam programming guide</a> to learn the basic concepts that apply to all SDKs in Beam.</p>
-<p>See the <a href="/documentation/sdks/pydoc/">Python API Reference</a> for more information on individual APIs.</p>
+<p>See the <a href="/documentation/sdks/pydoc/">Python API reference</a> for more information on individual APIs.</p>
-<h2 id="python-type-safety">Python Type Safety</h2>
+<h2 id="python-streaming-pipelines">Python streaming pipelines</h2>
+
+<p>Python <a href="/documentation/sdks/python-streaming">streaming pipeline execution</a>
+is experimentally available (with some <a href="/documentation/sdks/python-streaming/#unsupported-features">limitations</a>)
+starting with Beam SDK version 2.5.0.</p>
+
+<h2 id="python-type-safety">Python type safety</h2>
<p>Python is a dynamically-typed language with no static type checking. The Beam SDK for Python uses type hints during pipeline construction and runtime to try to emulate the correctness guarantees achieved by true static typing. <a href="/documentation/sdks/python-type-safety">Ensuring Python Type Safety</a> walks through how to use type hints, which help you to catch potential bugs up front with the <a href="/documentation/runners/direct/">Direct Runner</a>.</p>
-<h2 id="managing-python-pipeline-dependencies">Managing Python Pipeline Dependencies</h2>
+<h2 id="managing-python-pipeline-dependencies">Managing Python pipeline dependencies</h2>
<p>When you run your pipeline locally, the packages that your pipeline depends on are available because they are installed on your local machine. However, when you want to run your pipeline remotely, you must make sure these dependencies are available on the remote machines. <a href="/documentation/sdks/python-pipeline-dependencies">Managing Python Pipeline Dependencies</a> shows you how to make your dependencies available to the remote workers.</p>
-<h2 id="creating-new-sources-and-sinks">Creating New Sources and Sinks</h2>
+<h2 id="creating-new-sources-and-sinks">Creating new sources and Sinks</h2>
<p>The Beam SDK for Python provides an extensible API that you can use to create new data sources and sinks. <a href="/documentation/sdks/python-custom-io">Creating New Sources and Sinks with the Python SDK</a> shows how to create new sources and sinks using <a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/iobase.py">Beam’s Source and Sink API</a>.</p>
diff --git a/content/get-started/quickstart-java/index.html b/content/get-started/quickstart-java/index.html
index d046707..642122b 100644
--- a/content/get-started/quickstart-java/index.html
+++ b/content/get-started/quickstart-java/index.html
@@ -225,7 +225,9 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
</code></pre>
</div>
-<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code>$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
+<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code>Make sure you complete the setup steps at https://beam.apache.org/documentation/runners/dataflow/#setup
+
+$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
-Dexec.args="--runner=DataflowRunner --project=<your-gcp-project> \
--gcpTempLocation=gs://<your-gcs-bucket>/tmp \
--inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://<your-gcs-bucket>/counts" \
diff --git a/content/get-started/quickstart-py/index.html b/content/get-started/quickstart-py/index.html
index a38389a..7cc21c0 100644
--- a/content/get-started/quickstart-py/index.html
+++ b/content/get-started/quickstart-py/index.html
@@ -127,7 +127,7 @@
<li><a href="#download-and-install">Download and install</a></li>
</ul>
</li>
- <li><a href="#execute-a-pipeline-locally">Execute a pipeline locally</a></li>
+ <li><a href="#execute-a-pipeline">Execute a pipeline</a></li>
<li><a href="#next-steps">Next Steps</a></li>
</ul>
@@ -149,12 +149,12 @@
<li><a href="#get-apache-beam" id="markdown-toc-get-apache-beam">Get Apache Beam</a> <ul>
<li><a href="#create-and-activate-a-virtual-environment" id="markdown-toc-create-and-activate-a-virtual-environment">Create and activate a virtual environment</a></li>
<li><a href="#download-and-install" id="markdown-toc-download-and-install">Download and install</a> <ul>
- <li><a href="#extra-requirements" id="markdown-toc-extra-requirements">Extra Requirements</a></li>
+ <li><a href="#extra-requirements" id="markdown-toc-extra-requirements">Extra requirements</a></li>
</ul>
</li>
</ul>
</li>
- <li><a href="#execute-a-pipeline-locally" id="markdown-toc-execute-a-pipeline-locally">Execute a pipeline locally</a></li>
+ <li><a href="#execute-a-pipeline" id="markdown-toc-execute-a-pipeline">Execute a pipeline</a></li>
<li><a href="#next-steps" id="markdown-toc-next-steps">Next Steps</a></li>
</ul>
@@ -176,16 +176,27 @@
</code></pre>
</div>
+<p>If you do not have <code class="highlighter-rouge">pip</code> version 7.0.0 or newer, run the following command to
+install it. This command might require administrative privileges.</p>
+
+<div class="highlighter-rouge"><pre class="highlight"><code>pip install --upgrade pip
+</code></pre>
+</div>
+
<h3 id="install-python-virtual-environment">Install Python virtual environment</h3>
<p>It is recommended that you install a <a href="http://docs.python-guide.org/en/latest/dev/virtualenvs/">Python virtual environment</a>
-for initial experiments. If you do not have <code class="highlighter-rouge">virtualenv</code> version 13.1.0 or newer, install it by running:</p>
+for initial experiments. If you do not have <code class="highlighter-rouge">virtualenv</code> version 13.1.0 or
+newer, run the following command to install it. This command might require
+administrative privileges.</p>
<div class="highlighter-rouge"><pre class="highlight"><code>pip install --upgrade virtualenv
</code></pre>
</div>
-<p>If you do not want to use a Python virtual environment (not recommended), ensure <code class="highlighter-rouge">setuptools</code> is installed on your machine. If you do not have <code class="highlighter-rouge">setuptools</code> version 17.1 or newer, install it by running:</p>
+<p>If you do not want to use a Python virtual environment (not recommended), ensure
+<code class="highlighter-rouge">setuptools</code> is installed on your machine. If you do not have <code class="highlighter-rouge">setuptools</code>
+version 17.1 or newer, run the following command to install it.</p>
<div class="highlighter-rouge"><pre class="highlight"><code>pip install --upgrade setuptools
</code></pre>
@@ -223,9 +234,9 @@ environment’s directories.</p>
</code></pre>
</div>
-<h4 id="extra-requirements">Extra Requirements</h4>
+<h4 id="extra-requirements">Extra requirements</h4>
-<p>The above installation will not install all the extra dependencies for using features like the Google Cloud Dataflow runner. Information on what extra packages are required for different features are highlighted below. It is possible to install multitple extra requirements using something like <code class="highlighter-rouge">pip install apache-beam[feature1, feature2]</code>.</p>
+<p>The above installation will not install all the extra dependencies for using features like the Google Cloud Dataflow runner. Information on what extra packages are required for different features are highlighted below. It is possible to install multitple extra requirements using something like <code class="highlighter-rouge">pip install apache-beam[feature1,feature2]</code>.</p>
<ul>
<li><strong>Google Cloud Platform</strong>
@@ -255,17 +266,34 @@ environment’s directories.</p>
</li>
</ul>
-<h2 id="execute-a-pipeline-locally">Execute a pipeline locally</h2>
+<h2 id="execute-a-pipeline">Execute a pipeline</h2>
<p>The Apache Beam <a href="https://github.com/apache/beam/tree/master/sdks/python/apache_beam/examples">examples</a> directory has many examples. All examples can be run locally by passing the required arguments described in the example script.</p>
-<p>For example, to run <code class="highlighter-rouge">wordcount.py</code>, run:</p>
+<p>For example, run <code class="highlighter-rouge">wordcount.py</code> with the following command:</p>
-<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount --input <PATH_TO_INPUT_FILE> --output counts
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount --input /path/to/inputfile --output /path/to/write/counts
</code></pre>
</div>
-<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
+<div class="runner-apex highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-local highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-cluster highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-spark highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components. Make sure you
+# complete the setup steps at https://beam.apache.org/documentation/runners/dataflow/#setup
pip install apache-beam[gcp]
python -m apache_beam.examples.wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
--output gs://<your-gcs-bucket>/counts \
@@ -275,6 +303,11 @@ python -m apache_beam.examples.wordcount --input gs://dataflow-samples/shakespea
</code></pre>
</div>
+<p>After the pipeline completes, you can view the output files at your specified
+output path. For example, if you specify <code class="highlighter-rouge">/dir1/counts</code> for the <code class="highlighter-rouge">--output</code>
+parameter, the pipeline writes the files to <code class="highlighter-rouge">/dir1/</code> and names the files
+sequentially in the format <code class="highlighter-rouge">counts-0000-of-0001</code>.</p>
+
<h2 id="next-steps">Next Steps</h2>
<ul>
diff --git a/content/get-started/wordcount-example/index.html b/content/get-started/wordcount-example/index.html
index 575bae7..e7b460f 100644
--- a/content/get-started/wordcount-example/index.html
+++ b/content/get-started/wordcount-example/index.html
@@ -131,7 +131,7 @@
<li><a href="#debuggingwordcount-example">DebuggingWordCount example</a>
<ul>
<li><a href="#logging">Logging</a></li>
- <li><a href="#testing-your-pipeline-via-passert">Testing your pipeline via PAssert</a></li>
+ <li><a href="#testing-your-pipeline-with-asserts">Testing your pipeline with asserts</a></li>
</ul>
</li>
<li><a href="#windowedwordcount-example">WindowedWordCount example</a>
@@ -140,7 +140,12 @@
<li><a href="#adding-timestamps-to-data">Adding timestamps to data</a></li>
<li><a href="#windowing">Windowing</a></li>
<li><a href="#reusing-ptransforms-over-windowed-pcollections">Reusing PTransforms over windowed PCollections</a></li>
- <li><a href="#writing-results-to-an-unbounded-sink">Writing results to an unbounded sink</a></li>
+ </ul>
+ </li>
+ <li><a href="#streamingwordcount-example">StreamingWordCount example</a>
+ <ul>
+ <li><a href="#reading-an-unbounded-data-set">Reading an unbounded data set</a></li>
+ <li><a href="#writing-unbounded-results">Writing unbounded results</a></li>
</ul>
</li>
</ul>
@@ -173,7 +178,7 @@
<li><a href="#apache-apex-runner" id="markdown-toc-apache-apex-runner">Apache Apex Runner</a></li>
</ul>
</li>
- <li><a href="#testing-your-pipeline-via-passert" id="markdown-toc-testing-your-pipeline-via-passert">Testing your pipeline via PAssert</a></li>
+ <li><a href="#testing-your-pipeline-with-asserts" id="markdown-toc-testing-your-pipeline-with-asserts">Testing your pipeline with asserts</a></li>
</ul>
</li>
<li><a href="#windowedwordcount-example" id="markdown-toc-windowedwordcount-example">WindowedWordCount example</a> <ul>
@@ -181,7 +186,11 @@
<li><a href="#adding-timestamps-to-data" id="markdown-toc-adding-timestamps-to-data">Adding timestamps to data</a></li>
<li><a href="#windowing" id="markdown-toc-windowing">Windowing</a></li>
<li><a href="#reusing-ptransforms-over-windowed-pcollections" id="markdown-toc-reusing-ptransforms-over-windowed-pcollections">Reusing PTransforms over windowed PCollections</a></li>
- <li><a href="#writing-results-to-an-unbounded-sink" id="markdown-toc-writing-results-to-an-unbounded-sink">Writing results to an unbounded sink</a></li>
+ </ul>
+ </li>
+ <li><a href="#streamingwordcount-example" id="markdown-toc-streamingwordcount-example">StreamingWordCount example</a> <ul>
+ <li><a href="#reading-an-unbounded-data-set" id="markdown-toc-reading-an-unbounded-data-set">Reading an unbounded data set</a></li>
+ <li><a href="#writing-unbounded-results" id="markdown-toc-writing-unbounded-results">Writing unbounded results</a></li>
</ul>
</li>
</ul>
@@ -217,57 +226,29 @@ to handle both bounded and unbounded datasets.</li>
<h2 id="minimalwordcount-example">MinimalWordCount example</h2>
-<p>MinimalWordCount demonstrates a simple pipeline that can read from a text file,
-apply transforms to tokenize and count the words, and write the data to an
-output text file. This example hard-codes the locations for its input and output
-files and doesn’t perform any error checking; it is intended to only show you
-the “bare bones” of creating a Beam pipeline. This lack of parameterization
-makes this particular pipeline less portable across different runners than
-standard Beam pipelines. In later examples, we will parameterize the pipeline’s
-input and output sources and show other best practices.</p>
-
-<p><strong>To run this example in Java:</strong></p>
-
-<div class="highlighter-rouge"><pre class="highlight"><code>$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.MinimalWordCount
-</code></pre>
-</div>
-
-<p>To view the full code in Java, see
-<strong><a href="https://github.com/apache/beam/blob/master/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java">MinimalWordCount</a>.</strong></p>
-
-<p><strong>To run this example in Python:</strong></p>
-
-<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount_minimal --input README.md --output counts
-</code></pre>
-</div>
+<p>MinimalWordCount demonstrates a simple pipeline that uses the Direct Runner to
+read from a text file, apply transforms to tokenize and count the words, and
+write the data to an output text file.</p>
-<div class="runner-apex highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
-</code></pre>
-</div>
+<p class="language-java">This example hard-codes the locations for its input and output files and doesn’t
+perform any error checking; it is intended to only show you the “bare bones” of
+creating a Beam pipeline. This lack of parameterization makes this particular
+pipeline less portable across different runners than standard Beam pipelines. In
+later examples, we will parameterize the pipeline’s input and output sources and
+show other best practices.</p>
-<div class="runner-flink-local highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+<div class="language-java highlighter-rouge"><pre class="highlight"><code><span class="err">$</span> <span class="n">mvn</span> <span class="n">compile</span> <span class="nl">exec:</span><span class="n">java</span> <span class="o">-</span><span class="n">Dexec</span><span class="o">.</span><span class="na">mainClass</span><span class="o">=</span><span class="n">org</span><span class="o">.</span><span class="na">apache</span><span class="o">.</span><span class="na">beam</span><span class [...]
</code></pre>
</div>
-<div class="runner-flink-cluster highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="n">python</span> <span class="o">-</span><span class="n">m</span> <span class="n">apache_beam</span><span class="o">.</span><span class="n">examples</span><span class="o">.</span><span class="n">wordcount_minimal</span> <span class="o">--</span><span class="nb">input</span> <span class="n">YOUR_INPUT_FILE</span> <span class="o">--</span><span class="n">output</span> <span class="n">counts</span>
</code></pre>
</div>
-<div class="runner-spark highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
-</code></pre>
-</div>
-
-<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
-pip install apache-beam[gcp]
-python -m apache_beam.examples.wordcount_minimal --input gs://dataflow-samples/shakespeare/kinglear.txt \
- --output gs://<your-gcs-bucket>/counts \
- --runner DataflowRunner \
- --project your-gcp-project \
- --temp_location gs://<your-gcs-bucket>/tmp/
-</code></pre>
-</div>
+<p class="language-java">To view the full code in Java, see
+<strong><a href="https://github.com/apache/beam/blob/master/examples/java/src/main/java/org/apache/beam/examples/MinimalWordCount.java">MinimalWordCount</a>.</strong></p>
-<p>To view the full code in Python, see
+<p class="language-py">To view the full code in Python, see
<strong><a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/wordcount_minimal.py">wordcount_minimal.py</a>.</strong></p>
<p><strong>Key Concepts:</strong></p>
@@ -303,19 +284,6 @@ sections, we will specify the pipeline’s runner.</p>
<span class="c1">// will run with the DirectRunner by default, based on the class path configured</span>
<span class="c1">// in its dependencies.</span>
<span class="n">PipelineOptions</span> <span class="n">options</span> <span class="o">=</span> <span class="n">PipelineOptionsFactory</span><span class="o">.</span><span class="na">create</span><span class="o">();</span>
-
- <span class="c1">// In order to run your pipeline, you need to make following runner specific changes:</span>
- <span class="c1">//</span>
- <span class="c1">// CHANGE 1/3: Select a Beam runner, such as DataflowRunner or FlinkRunner.</span>
- <span class="c1">// CHANGE 2/3: Specify runner-required options.</span>
- <span class="c1">// For DataflowRunner, set project and temp location as follows:</span>
- <span class="c1">// DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);</span>
- <span class="c1">// dataflowOptions.setRunner(DataflowRunner.class);</span>
- <span class="c1">// dataflowOptions.setProject("SET_YOUR_PROJECT_ID_HERE");</span>
- <span class="c1">// dataflowOptions.setTempLocation("gs://SET_YOUR_BUCKET_NAME_HERE/AND_TEMP_DIRECTORY");</span>
- <span class="c1">// For FlinkRunner, set the runner as follows. See {@code FlinkPipelineOptions}</span>
- <span class="c1">// for more details.</span>
- <span class="c1">// options.setRunner(FlinkRunner.class);</span>
</code></pre>
</div>
@@ -376,15 +344,15 @@ data stored in a publicly accessible Google Cloud Storage bucket (“gs://”).<
</div>
</li>
<li>
- <p>This transform splits the lines in PCollection<String>, where each element
-is an individual word in Shakespeare's collected texts.
-As an alternative, it would have been possible to use a
-[ParDo](/documentation/programming-guide/#pardo)
-transform that invokes a `DoFn` (defined in-line as an anonymous class) on
+ <p>This transform splits the lines in <code class="highlighter-rouge">PCollection<String></code>, where each element
+is an individual word in Shakespeare’s collected texts.
+As an alternative, it would have been possible to use a
+<a href="/documentation/programming-guide/#pardo">ParDo</a>
+transform that invokes a <code class="highlighter-rouge">DoFn</code> (defined in-line as an anonymous class) on
each element that tokenizes the text lines into individual words. The input
-for this transform is the `PCollection` of text lines generated by the
-previous `TextIO.Read` transform. The `ParDo` transform outputs a new
-`PCollection`, where each element represents an individual word in the text.</String></p>
+for this transform is the <code class="highlighter-rouge">PCollection</code> of text lines generated by the
+previous <code class="highlighter-rouge">TextIO.Read</code> transform. The <code class="highlighter-rouge">ParDo</code> transform outputs a new
+<code class="highlighter-rouge">PCollection</code>, where each element represents an individual word in the text.</p>
<div class="language-java highlighter-rouge"><pre class="highlight"><code> <span class="o">.</span><span class="na">apply</span><span class="o">(</span><span class="s">"ExtractWords"</span><span class="o">,</span> <span class="n">FlatMapElements</span>
<span class="o">.</span><span class="na">into</span><span class="o">(</span><span class="n">TypeDescriptors</span><span class="o">.</span><span class="na">strings</span><span class="o">())</span>
@@ -515,8 +483,8 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
</div>
<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code>$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WordCount \
- -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://<your-gcs-bucket>/tmp \
- --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://<your-gcs-bucket>/counts" \
+ -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://YOUR_GCS_BUCKET/tmp \
+ --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://YOUR_GCS_BUCKET/counts" \
-Pdataflow-runner
</code></pre>
</div>
@@ -526,7 +494,7 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
<p><strong>To run this example in Python:</strong></p>
-<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount --input README.md --output counts
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount --input YOUR_INPUT_FILE --output counts
</code></pre>
</div>
@@ -549,10 +517,10 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
pip install apache-beam[gcp]
python -m apache_beam.examples.wordcount --input gs://dataflow-samples/shakespeare/kinglear.txt \
- --output gs://<your-gcs-bucket>/counts \
+ --output gs://YOUR_GCS_BUCKET/counts \
--runner DataflowRunner \
- --project your-gcp-project \
- --temp_location gs://<your-gcs-bucket>/tmp/
+ --project YOUR_GCP_PROJECT \
+ --temp_location gs://YOUR_GCS_BUCKET/tmp/
</code></pre>
</div>
@@ -749,7 +717,7 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
<p><strong>To run this example in Python:</strong></p>
-<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount_debugging --input README.md --output counts
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.wordcount_debugging --input YOUR_INPUT_FILE --output counts
</code></pre>
</div>
@@ -772,10 +740,10 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
pip install apache-beam[gcp]
python -m apache_beam.examples.wordcount_debugging --input gs://dataflow-samples/shakespeare/kinglear.txt \
- --output gs://<your-gcs-bucket>/counts \
+ --output gs://YOUR_GCS_BUCKET/counts \
--runner DataflowRunner \
- --project your-gcp-project \
- --temp_location gs://<your-gcs-bucket>/tmp/
+ --project YOUR_GCP_PROJECT \
+ --temp_location gs://YOUR_GCS_BUCKET/tmp/
</code></pre>
</div>
@@ -908,18 +876,21 @@ or DEBUG significantly increases the amount of logs output.</p>
(<a href="https://issues.apache.org/jira/browse/BEAM-2285">BEAM-2285</a>).</p>
</blockquote>
-<h3 id="testing-your-pipeline-via-passert">Testing your pipeline via PAssert</h3>
+<h3 id="testing-your-pipeline-with-asserts">Testing your pipeline with asserts</h3>
-<p><code class="highlighter-rouge">PAssert</code> is a set of convenient PTransforms in the style of Hamcrest’s
-collection matchers that can be used when writing Pipeline level tests to
-validate the contents of PCollections. <code class="highlighter-rouge">PAssert</code> is best used in unit tests with
-small data sets, but is demonstrated here as a teaching tool.</p>
+<p><span class="language-java"><code class="highlighter-rouge">PAssert</code></span><span class="language-py"><code class="highlighter-rouge">assert_that</code></span>
+is a set of convenient PTransforms in the style of Hamcrest’s collection
+matchers that can be used when writing pipeline level tests to validate the
+contents of PCollections. Asserts are best used in unit tests with small data
+sets.</p>
-<p>Below, we verify that the set of filtered words matches our expected counts.
-Note that <code class="highlighter-rouge">PAssert</code> does not produce any output, and the pipeline only succeeds
-if all of the expectations are met. See
-<a href="https://github.com/apache/beam/blob/master/examples/java/src/test/java/org/apache/beam/examples/DebuggingWordCountTest.java">DebuggingWordCountTest</a>
-for an example unit test.</p>
+<p class="language-java">The following example verifies that the set of filtered words matches our
+expected counts. The assert does not produce any output, and the pipeline only
+succeeds if all of the expectations are met.</p>
+
+<p class="language-py">The following example verifies that two collections contain the same values. The
+assert does not produce any output, and the pipeline only succeeds if all of the
+expectations are met.</p>
<div class="language-java highlighter-rouge"><pre class="highlight"><code><span class="kd">public</span> <span class="kd">static</span> <span class="kt">void</span> <span class="nf">main</span><span class="o">(</span><span class="n">String</span><span class="o">[]</span> <span class="n">args</span><span class="o">)</span> <span class="o">{</span>
<span class="o">...</span>
@@ -932,10 +903,17 @@ for an example unit test.</p>
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="c"># This feature is not yet available in the Beam SDK for Python.</span>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="kn">from</span> <span class="nn">apache_beam.testing.util</span> <span class="kn">import</span> <span class="n">assert_that</span>
+<span class="kn">from</span> <span class="nn">apache_beam.testing.util</span> <span class="kn">import</span> <span class="n">equal_to</span>
+
+<span class="k">with</span> <span class="n">TestPipeline</span><span class="p">()</span> <span class="k">as</span> <span class="n">p</span><span class="p">:</span>
+ <span class="n">assert_that</span><span class="p">(</span><span class="n">p</span> <span class="o">|</span> <span class="n">Create</span><span class="p">([</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">,</span> <span class="mi">3</span><span class="p">]),</span> <span class="n">equal_to</span><span class="p">([</span><span class="mi">1</span><span class="p">,</span> <span class="mi">2</span><span class="p">,</span> <span class="mi">3</ [...]
</code></pre>
</div>
+<p class="language-java">See <a href="https://github.com/apache/beam/blob/master/examples/java/src/test/java/org/apache/beam/examples/DebuggingWordCountTest.java">DebuggingWordCountTest</a>
+for an example unit test.</p>
+
<h2 id="windowedwordcount-example">WindowedWordCount example</h2>
<p>The WindowedWordCount example counts words in text just as the previous
@@ -984,8 +962,8 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
</div>
<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code>$ mvn compile exec:java -Dexec.mainClass=org.apache.beam.examples.WindowedWordCount \
- -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://<your-gcs-bucket>/tmp \
- --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://<your-gcs-bucket>/counts" \
+ -Dexec.args="--runner=DataflowRunner --gcpTempLocation=gs://YOUR_GCS_BUCKET/tmp \
+ --inputFile=gs://apache-beam-samples/shakespeare/* --output=gs://YOUR_GCS_BUCKET/counts" \
-Pdataflow-runner
</code></pre>
</div>
@@ -993,9 +971,44 @@ You can monitor the running job by visiting the Flink dashboard at http://<fl
<p>To view the full code in Java, see
<strong><a href="https://github.com/apache/beam/blob/master/examples/java/src/main/java/org/apache/beam/examples/WindowedWordCount.java">WindowedWordCount</a>.</strong></p>
-<blockquote>
- <p><strong>Note:</strong> WindowedWordCount is not yet available for the Python SDK.</p>
-</blockquote>
+<p><strong>To run this example in Python:</strong></p>
+
+<p>This pipeline writes its results to a BigQuery table <code class="highlighter-rouge">--output_table</code>
+parameter. using the format <code class="highlighter-rouge">PROJECT:DATASET.TABLE</code> or
+<code class="highlighter-rouge">DATASET.TABLE</code>.</p>
+
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.windowed_wordcount --input YOUR_INPUT_FILE --output_table PROJECT:DATASET.TABLE
+</code></pre>
+</div>
+
+<div class="runner-apex highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-local highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-cluster highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-spark highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
+pip install apache-beam[gcp]
+python -m apache_beam.examples.windowed_wordcount --input YOUR_INPUT_FILE \
+ --output_table PROJECT:DATASET.TABLE \
+ --runner DataflowRunner \
+ --project YOUR_GCP_PROJECT \
+ --temp_location gs://YOUR_GCS_BUCKET/tmp/
+</code></pre>
+</div>
+
+<p>To view the full code in Python, see
+<strong><a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/windowed_wordcount.py">windowed_wordcount.py</a>.</strong></p>
<h3 id="unbounded-and-bounded-pipeline-input-modes">Unbounded and bounded pipeline input modes</h3>
@@ -1119,28 +1132,105 @@ PCollections over windowed PCollections as well.</p>
</code></pre>
</div>
-<h3 id="writing-results-to-an-unbounded-sink">Writing results to an unbounded sink</h3>
+<h2 id="streamingwordcount-example">StreamingWordCount example</h2>
+
+<p>The StreamingWordCount example is a streaming pipeline that reads Pub/Sub
+messages from a Pub/Sub subscription or topic, and performs a frequency count on
+the words in each message. Similar to WindowedWordCount, this example applies
+fixed-time windowing, wherein each window represents a fixed time interval. The
+fixed window size for this example is 15 seconds. The pipeline outputs the
+frequency count of the words seen in each 15 second window.</p>
+
+<p><strong>New Concepts:</strong></p>
+
+<ul>
+ <li>Reading an unbounded data set</li>
+ <li>Writing unbounded results</li>
+</ul>
-<p>When our input is unbounded, the same is true of our output <code class="highlighter-rouge">PCollection</code>. We
-need to make sure that we choose an appropriate, unbounded sink. Some output
-sinks support only bounded output, while others support both bounded and
-unbounded outputs. By using a <code class="highlighter-rouge">FilenamePolicy</code>, we can use <code class="highlighter-rouge">TextIO</code> to files
-that are partitioned by windows. We use a composite <code class="highlighter-rouge">PTransform</code> that uses such
-a policy internally to write a single sharded file per window.</p>
+<p><strong>To run this example in Java:</strong></p>
-<p>In this example, we stream the results to Google BigQuery. The code formats the
-results and writes them to a BigQuery table using <code class="highlighter-rouge">BigQueryIO.Write</code>.</p>
+<blockquote>
+ <p><strong>Note:</strong> StreamingWordCount is not yet available for the Java SDK.</p>
+</blockquote>
-<div class="language-java highlighter-rouge"><pre class="highlight"><code> <span class="n">wordCounts</span>
- <span class="o">.</span><span class="na">apply</span><span class="o">(</span><span class="n">MapElements</span><span class="o">.</span><span class="na">via</span><span class="o">(</span><span class="k">new</span> <span class="n">WordCount</span><span class="o">.</span><span class="na">FormatAsTextFn</span><span class="o">()))</span>
- <span class="o">.</span><span class="na">apply</span><span class="o">(</span><span class="k">new</span> <span class="n">WriteOneFilePerWindow</span><span class="o">(</span><span class="n">output</span><span class="o">,</span> <span class="n">options</span><span class="o">.</span><span class="na">getNumShards</span><span class="o">()));</span>
+<p><strong>To run this example in Python:</strong></p>
+
+<div class="runner-direct highlighter-rouge"><pre class="highlight"><code>python -m apache_beam.examples.streaming_wordcount \
+ --input_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_INPUT_TOPIC" \
+ --output_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_OUTPUT_TOPIC" \
+ --streaming
</code></pre>
</div>
-<div class="language-py highlighter-rouge"><pre class="highlight"><code><span class="c"># This feature is not yet available in the Beam SDK for Python.</span>
+<div class="runner-apex highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-flink-local highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
</code></pre>
</div>
+<div class="runner-flink-cluster highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-spark highlighter-rouge"><pre class="highlight"><code>This runner is not yet available for the Python SDK.
+</code></pre>
+</div>
+
+<div class="runner-dataflow highlighter-rouge"><pre class="highlight"><code># As part of the initial setup, install Google Cloud Platform specific extra components.
+pip install apache-beam[gcp]
+python -m apache_beam.examples.streaming_wordcount \
+ --runner DataflowRunner \
+ --project YOUR_GCP_PROJECT \
+ --temp_location gs://YOUR_GCS_BUCKET/tmp/ \
+ --input_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_INPUT_TOPIC" \
+ --output_topic "projects/YOUR_PUBSUB_PROJECT_NAME/topics/YOUR_OUTPUT_TOPIC" \
+ --streaming
+</code></pre>
+</div>
+
+<p>To view the full code in Python, see
+<strong><a href="https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/streaming_wordcount.py">streaming_wordcount.py</a>.</strong></p>
+
+<h3 id="reading-an-unbounded-data-set">Reading an unbounded data set</h3>
+
+<p>This example uses an unbounded data set as input. The code reads Pub/Sub
+messages from a Pub/Sub subscription or topic using
+<a href="/documentation/sdks/pydoc/2.4.0/apache_beam.io.gcp.pubsub.html#apache_beam.io.gcp.pubsub.ReadStringsFromPubSub"><code class="highlighter-rouge">beam.io.ReadStringsFromPubSub</code></a>.</p>
+
+<div class="language-java highlighter-rouge"><pre class="highlight"><code> <span class="c1">// This example is not currently available for the Beam SDK for Java.</span>
+</code></pre>
+</div>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># Read from Pub/Sub into a PCollection.</span>
+ <span class="k">if</span> <span class="n">known_args</span><span class="o">.</span><span class="n">input_subscription</span><span class="p">:</span>
+ <span class="n">lines</span> <span class="o">=</span> <span class="n">p</span> <span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">io</span><span class="o">.</span><span class="n">ReadStringsFromPubSub</span><span class="p">(</span>
+ <span class="n">subscription</span><span class="o">=</span><span class="n">known_args</span><span class="o">.</span><span class="n">input_subscription</span><span class="p">)</span>
+ <span class="k">else</span><span class="p">:</span>
+ <span class="n">lines</span> <span class="o">=</span> <span class="n">p</span> <span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">io</span><span class="o">.</span><span class="n">ReadStringsFromPubSub</span><span class="p">(</span><span class="n">topic</span><span class="o">=</span><span class="n">known_args</span><span class="o">.</span><span class="n">input_topic</span><span class="p">)</span>
+</code></pre>
+</div>
+<h3 id="writing-unbounded-results">Writing unbounded results</h3>
+
+<p>When the input is unbounded, the same is true of the output <code class="highlighter-rouge">PCollection</code>. As
+such, you must make sure to choose an appropriate I/O for the results. Some I/Os
+support only bounded output, while others support both bounded and unbounded
+outputs.</p>
+
+<p>This example uses an unbounded <code class="highlighter-rouge">PCollection</code> and streams the results to
+Google Pub/Sub. The code formats the results and writes them to a Pub/Sub topic
+using <a href="/documentation/sdks/pydoc/2.4.0/apache_beam.io.gcp.pubsub.html#apache_beam.io.gcp.pubsub.WriteStringsToPubSub"><code class="highlighter-rouge">beam.io.WriteStringsToPubSub</code></a>.</p>
+
+<div class="language-java highlighter-rouge"><pre class="highlight"><code> <span class="c1">// This example is not currently available for the Beam SDK for Java.</span>
+</code></pre>
+</div>
+<div class="language-py highlighter-rouge"><pre class="highlight"><code> <span class="c"># Write to Pub/Sub</span>
+ <span class="n">output</span> <span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">io</span><span class="o">.</span><span class="n">WriteStringsToPubSub</span><span class="p">(</span><span class="n">known_args</span><span class="o">.</span><span class="n">output_topic</span><span class="p">)</span>
+</code></pre>
+</div>
+
+
</div>
</div>
<footer class="footer">
--
To stop receiving notification emails like this one, please contact
mergebot-role@apache.org.