Google Git
Sign in
apache / beam-site / 2124d9d3064f7803e25018ca63200f5af5857a1f / . / pydoc / 2.8.0 / _modules / apache_beam / testing / synthetic_pipeline.html
blob: 2d12b8d4b77e4fcf01bf12a8eda3aaaa58f3685a [file] [log] [blame]
<!DOCTYPE html>
<!--[if IE 8]><html class="no-js lt-ie9" lang="en" > <![endif]-->
<!--[if gt IE 8]><!--> <html class="no-js" lang="en" > <!--<![endif]-->
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>apache_beam.testing.synthetic_pipeline &mdash; Apache Beam documentation</title>
<link rel="stylesheet" href="../../../_static/css/theme.css" type="text/css" />
<link rel="index" title="Index"
href="../../../genindex.html"/>
<link rel="search" title="Search" href="../../../search.html"/>
<link rel="top" title="Apache Beam documentation" href="../../../index.html"/>
<link rel="up" title="Module code" href="../../index.html"/>
<script src="../../../_static/js/modernizr.min.js"></script>
</head>
<body class="wy-body-for-nav" role="document">
<div class="wy-grid-for-nav">
<nav data-toggle="wy-nav-shift" class="wy-nav-side">
<div class="wy-side-scroll">
<div class="wy-side-nav-search">
<a href="../../../index.html" class="icon icon-home"> Apache Beam
</a>
<div role="search">
<form id="rtd-search-form" class="wy-form" action="../../../search.html" method="get">
<input type="text" name="q" placeholder="Search docs" />
<input type="hidden" name="check_keywords" value="yes" />
<input type="hidden" name="area" value="default" />
</form>
</div>
</div>
<div class="wy-menu wy-menu-vertical" data-spy="affix" role="navigation" aria-label="main navigation">
<ul>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.coders.html">apache_beam.coders package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.internal.html">apache_beam.internal package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.io.html">apache_beam.io package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.metrics.html">apache_beam.metrics package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.options.html">apache_beam.options package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.portability.html">apache_beam.portability package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.runners.html">apache_beam.runners package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.testing.html">apache_beam.testing package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.tools.html">apache_beam.tools package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.transforms.html">apache_beam.transforms package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.typehints.html">apache_beam.typehints package</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.utils.html">apache_beam.utils package</a></li>
</ul>
<ul>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.error.html">apache_beam.error module</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.pipeline.html">apache_beam.pipeline module</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.pvalue.html">apache_beam.pvalue module</a></li>
<li class="toctree-l1"><a class="reference internal" href="../../../apache_beam.version.html">apache_beam.version module</a></li>
</ul>
</div>
</div>
</nav>
<section data-toggle="wy-nav-shift" class="wy-nav-content-wrap">
<nav class="wy-nav-top" role="navigation" aria-label="top navigation">
<i data-toggle="wy-nav-top" class="fa fa-bars"></i>
<a href="../../../index.html">Apache Beam</a>
</nav>
<div class="wy-nav-content">
<div class="rst-content">
<div role="navigation" aria-label="breadcrumbs navigation">
<ul class="wy-breadcrumbs">
<li><a href="../../../index.html">Docs</a> &raquo;</li>
<li><a href="../../index.html">Module code</a> &raquo;</li>
<li>apache_beam.testing.synthetic_pipeline</li>
<li class="wy-breadcrumbs-aside">
</li>
</ul>
<hr/>
</div>
<div role="main" class="document" itemscope="itemscope" itemtype="http://schema.org/Article">
<div itemprop="articleBody">
<h1>Source code for apache_beam.testing.synthetic_pipeline</h1><div class="highlight"><pre>
<span></span><span class="c1">#</span>
<span class="c1"># Licensed to the Apache Software Foundation (ASF) under one or more</span>
<span class="c1"># contributor license agreements. See the NOTICE file distributed with</span>
<span class="c1"># this work for additional information regarding copyright ownership.</span>
<span class="c1"># The ASF licenses this file to You under the Apache License, Version 2.0</span>
<span class="c1"># (the &quot;License&quot;); you may not use this file except in compliance with</span>
<span class="c1"># the License. You may obtain a copy of the License at</span>
<span class="c1">#</span>
<span class="c1"># http://www.apache.org/licenses/LICENSE-2.0</span>
<span class="c1">#</span>
<span class="c1"># Unless required by applicable law or agreed to in writing, software</span>
<span class="c1"># distributed under the License is distributed on an &quot;AS IS&quot; BASIS,</span>
<span class="c1"># WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.</span>
<span class="c1"># See the License for the specific language governing permissions and</span>
<span class="c1"># limitations under the License.</span>
<span class="c1">#</span>
<span class="sd">&quot;&quot;&quot;A set of utilities to write pipelines for performance tests.</span>
<span class="sd">This module offers a way to create pipelines using synthetic sources and steps.</span>
<span class="sd">Exact shape of the pipeline and the behaviour of sources and steps can be</span>
<span class="sd">controlled through arguments. Please see function &#39;parse_args()&#39; for more</span>
<span class="sd">details about the arguments.</span>
<span class="sd">Shape of the pipeline is primariy controlled through two arguments. Argument</span>
<span class="sd">&#39;steps&#39; can be used to define a list of steps as a JSON string. Argument</span>
<span class="sd">&#39;barrier&#39; describes how these steps are separated from each other. Argument</span>
<span class="sd">&#39;barrier&#39; can be use to build a pipeline as a a series of steps or a tree of</span>
<span class="sd">steps with a fanin or a fanout of size 2.</span>
<span class="sd">Other arguments describe what gets generated by synthetic sources that produce</span>
<span class="sd">data for the pipeline.</span>
<span class="sd">&quot;&quot;&quot;</span>
<span class="kn">from</span> <span class="nn">__future__</span> <span class="k">import</span> <span class="n">absolute_import</span>
<span class="kn">from</span> <span class="nn">__future__</span> <span class="k">import</span> <span class="n">division</span>
<span class="kn">import</span> <span class="nn">argparse</span>
<span class="kn">import</span> <span class="nn">json</span>
<span class="kn">import</span> <span class="nn">logging</span>
<span class="kn">import</span> <span class="nn">math</span>
<span class="kn">import</span> <span class="nn">time</span>
<span class="kn">import</span> <span class="nn">apache_beam</span> <span class="k">as</span> <span class="nn">beam</span>
<span class="kn">from</span> <span class="nn">apache_beam.io</span> <span class="k">import</span> <span class="n">WriteToText</span>
<span class="kn">from</span> <span class="nn">apache_beam.io</span> <span class="k">import</span> <span class="n">iobase</span>
<span class="kn">from</span> <span class="nn">apache_beam.io</span> <span class="k">import</span> <span class="n">range_trackers</span>
<span class="kn">from</span> <span class="nn">apache_beam.options.pipeline_options</span> <span class="k">import</span> <span class="n">PipelineOptions</span>
<span class="kn">from</span> <span class="nn">apache_beam.options.pipeline_options</span> <span class="k">import</span> <span class="n">SetupOptions</span>
<span class="kn">from</span> <span class="nn">apache_beam.testing.test_pipeline</span> <span class="k">import</span> <span class="n">TestPipeline</span>
<span class="k">try</span><span class="p">:</span>
<span class="kn">import</span> <span class="nn">numpy</span> <span class="k">as</span> <span class="nn">np</span>
<span class="k">except</span> <span class="ne">ImportError</span><span class="p">:</span>
<span class="n">np</span> <span class="o">=</span> <span class="kc">None</span>
<div class="viewcode-block" id="parse_byte_size"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.parse_byte_size">[docs]</a><span class="k">def</span> <span class="nf">parse_byte_size</span><span class="p">(</span><span class="n">s</span><span class="p">):</span>
<span class="n">suffixes</span> <span class="o">=</span> <span class="s1">&#39;BKMGTP&#39;</span>
<span class="k">if</span> <span class="n">s</span><span class="p">[</span><span class="o">-</span><span class="mi">1</span><span class="p">]</span> <span class="ow">in</span> <span class="n">suffixes</span><span class="p">:</span>
<span class="k">return</span> <span class="nb">int</span><span class="p">(</span><span class="nb">float</span><span class="p">(</span><span class="n">s</span><span class="p">[:</span><span class="o">-</span><span class="mi">1</span><span class="p">])</span> <span class="o">*</span> <span class="mi">1024</span><span class="o">**</span><span class="n">suffixes</span><span class="o">.</span><span class="n">index</span><span class="p">(</span><span class="n">s</span><span class="p">[</span><span class="o">-</span><span class="mi">1</span><span class="p">]))</span>
<span class="k">return</span> <span class="nb">int</span><span class="p">(</span><span class="n">s</span><span class="p">)</span></div>
<div class="viewcode-block" id="div_round_up"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.div_round_up">[docs]</a><span class="k">def</span> <span class="nf">div_round_up</span><span class="p">(</span><span class="n">a</span><span class="p">,</span> <span class="n">b</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Return ceil(a/b).&quot;&quot;&quot;</span>
<span class="k">return</span> <span class="nb">int</span><span class="p">(</span><span class="n">math</span><span class="o">.</span><span class="n">ceil</span><span class="p">(</span><span class="nb">float</span><span class="p">(</span><span class="n">a</span><span class="p">)</span> <span class="o">/</span> <span class="n">b</span><span class="p">))</span></div>
<div class="viewcode-block" id="rotate_key"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.rotate_key">[docs]</a><span class="k">def</span> <span class="nf">rotate_key</span><span class="p">(</span><span class="n">element</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Returns a new key-value pair of the same size but with a different key.&quot;&quot;&quot;</span>
<span class="p">(</span><span class="n">key</span><span class="p">,</span> <span class="n">value</span><span class="p">)</span> <span class="o">=</span> <span class="n">element</span>
<span class="k">return</span> <span class="n">key</span><span class="p">[</span><span class="o">-</span><span class="mi">1</span><span class="p">]</span> <span class="o">+</span> <span class="n">key</span><span class="p">[:</span><span class="o">-</span><span class="mi">1</span><span class="p">],</span> <span class="n">value</span></div>
<div class="viewcode-block" id="SyntheticStep"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticStep">[docs]</a><span class="k">class</span> <span class="nc">SyntheticStep</span><span class="p">(</span><span class="n">beam</span><span class="o">.</span><span class="n">DoFn</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;A DoFn of which behavior can be controlled through prespecified parameters.</span>
<span class="sd"> &quot;&quot;&quot;</span>
<span class="k">def</span> <span class="nf">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">per_element_delay_sec</span><span class="o">=</span><span class="mi">0</span><span class="p">,</span> <span class="n">per_bundle_delay_sec</span><span class="o">=</span><span class="mi">0</span><span class="p">,</span>
<span class="n">output_records_per_input_record</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span> <span class="n">output_filter_ratio</span><span class="o">=</span><span class="mi">0</span><span class="p">):</span>
<span class="k">if</span> <span class="n">per_element_delay_sec</span> <span class="ow">and</span> <span class="n">per_element_delay_sec</span> <span class="o">&lt;</span> <span class="mf">1e-3</span><span class="p">:</span>
<span class="k">raise</span> <span class="ne">ValueError</span><span class="p">(</span><span class="s1">&#39;Per element sleep time must be at least 1e-3. &#39;</span>
<span class="s1">&#39;Received: </span><span class="si">%r</span><span class="s1">&#39;</span><span class="p">,</span> <span class="n">per_element_delay_sec</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_per_element_delay_sec</span> <span class="o">=</span> <span class="n">per_element_delay_sec</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_per_bundle_delay_sec</span> <span class="o">=</span> <span class="n">per_bundle_delay_sec</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_output_records_per_input_record</span> <span class="o">=</span> <span class="n">output_records_per_input_record</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_output_filter_ratio</span> <span class="o">=</span> <span class="n">output_filter_ratio</span>
<div class="viewcode-block" id="SyntheticStep.start_bundle"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticStep.start_bundle">[docs]</a> <span class="k">def</span> <span class="nf">start_bundle</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_start_time</span> <span class="o">=</span> <span class="n">time</span><span class="o">.</span><span class="n">time</span><span class="p">()</span></div>
<div class="viewcode-block" id="SyntheticStep.finish_bundle"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticStep.finish_bundle">[docs]</a> <span class="k">def</span> <span class="nf">finish_bundle</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="c1"># The target is for the enclosing stage to take as close to as possible</span>
<span class="c1"># the given number of seconds, so we only sleep enough to make up for</span>
<span class="c1"># overheads not incurred elsewhere.</span>
<span class="n">to_sleep</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_per_bundle_delay_sec</span> <span class="o">-</span> <span class="p">(</span><span class="n">time</span><span class="o">.</span><span class="n">time</span><span class="p">()</span> <span class="o">-</span> <span class="bp">self</span><span class="o">.</span><span class="n">_start_time</span><span class="p">)</span>
<span class="c1"># Ignoring sub-millisecond sleep times.</span>
<span class="k">if</span> <span class="n">to_sleep</span> <span class="o">&gt;=</span> <span class="mf">1e-3</span><span class="p">:</span>
<span class="n">time</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="n">to_sleep</span><span class="p">)</span></div>
<div class="viewcode-block" id="SyntheticStep.process"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticStep.process">[docs]</a> <span class="k">def</span> <span class="nf">process</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">element</span><span class="p">):</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_per_element_delay_sec</span> <span class="o">&gt;=</span> <span class="mf">1e-3</span><span class="p">:</span>
<span class="n">time</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_per_element_delay_sec</span><span class="p">)</span>
<span class="n">filter_element</span> <span class="o">=</span> <span class="kc">False</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_output_filter_ratio</span> <span class="o">&gt;</span> <span class="mi">0</span><span class="p">:</span>
<span class="k">if</span> <span class="n">np</span><span class="o">.</span><span class="n">random</span><span class="o">.</span><span class="n">random</span><span class="p">()</span> <span class="o">&lt;</span> <span class="bp">self</span><span class="o">.</span><span class="n">_output_filter_ratio</span><span class="p">:</span>
<span class="n">filter_element</span> <span class="o">=</span> <span class="kc">True</span>
<span class="k">if</span> <span class="ow">not</span> <span class="n">filter_element</span><span class="p">:</span>
<span class="k">for</span> <span class="n">_</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_output_records_per_input_record</span><span class="p">):</span>
<span class="k">yield</span> <span class="n">element</span></div></div>
<div class="viewcode-block" id="SyntheticSource"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource">[docs]</a><span class="k">class</span> <span class="nc">SyntheticSource</span><span class="p">(</span><span class="n">iobase</span><span class="o">.</span><span class="n">BoundedSource</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;A custom source of a specified size.</span>
<span class="sd"> &quot;&quot;&quot;</span>
<span class="k">def</span> <span class="nf">__init__</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">input_spec</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Initiates a synthetic source.</span>
<span class="sd"> Args:</span>
<span class="sd"> input_spec: Input specification of the source. See corresponding option in</span>
<span class="sd"> function &#39;parse_args()&#39; below for more details.</span>
<span class="sd"> Raises:</span>
<span class="sd"> ValueError: if input parameters are invalid.</span>
<span class="sd"> &quot;&quot;&quot;</span>
<span class="k">def</span> <span class="nf">maybe_parse_byte_size</span><span class="p">(</span><span class="n">s</span><span class="p">):</span>
<span class="k">return</span> <span class="n">parse_byte_size</span><span class="p">(</span><span class="n">s</span><span class="p">)</span> <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">s</span><span class="p">,</span> <span class="nb">str</span><span class="p">)</span> <span class="k">else</span> <span class="nb">int</span><span class="p">(</span><span class="n">s</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span> <span class="o">=</span> <span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;numRecords&#39;</span><span class="p">]</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_key_size</span> <span class="o">=</span> <span class="n">maybe_parse_byte_size</span><span class="p">(</span><span class="n">input_spec</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">&#39;keySizeBytes&#39;</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_value_size</span> <span class="o">=</span> <span class="n">maybe_parse_byte_size</span><span class="p">(</span>
<span class="n">input_spec</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">&#39;valueSizeBytes&#39;</span><span class="p">,</span> <span class="mi">1</span><span class="p">))</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_total_size</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">element_size</span> <span class="o">*</span> <span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span> <span class="o">=</span> <span class="p">(</span>
<span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;bundleSizeDistribution&#39;</span><span class="p">][</span><span class="s1">&#39;type&#39;</span><span class="p">]</span>
<span class="k">if</span> <span class="s1">&#39;bundleSizeDistribution&#39;</span> <span class="ow">in</span> <span class="n">input_spec</span> <span class="k">else</span> <span class="s1">&#39;const&#39;</span><span class="p">)</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span> <span class="o">!=</span> <span class="s1">&#39;const&#39;</span> <span class="ow">and</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span> <span class="o">!=</span> <span class="s1">&#39;zipf&#39;</span><span class="p">:</span>
<span class="k">raise</span> <span class="ne">ValueError</span><span class="p">(</span>
<span class="s1">&#39;Only const and zipf distributions are supported for determining &#39;</span>
<span class="s1">&#39;sizes of bundles produced by initial splitting. Received: </span><span class="si">%s</span><span class="s1">&#39;</span><span class="p">,</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_num_bundles</span> <span class="o">=</span> <span class="p">(</span>
<span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;forceNumInitialBundles&#39;</span><span class="p">]</span>
<span class="k">if</span> <span class="s1">&#39;forceNumInitialBundles&#39;</span> <span class="ow">in</span> <span class="n">input_spec</span> <span class="k">else</span> <span class="mi">0</span><span class="p">)</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span> <span class="o">==</span> <span class="s1">&#39;zipf&#39;</span><span class="p">:</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_distribution_parameter</span> <span class="o">=</span> <span class="p">(</span>
<span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;bundleSizeDistribution&#39;</span><span class="p">][</span><span class="s1">&#39;param&#39;</span><span class="p">])</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_distribution_parameter</span> <span class="o">&lt;</span> <span class="mi">1</span><span class="p">:</span>
<span class="k">raise</span> <span class="ne">ValueError</span><span class="p">(</span>
<span class="s1">&#39;Parameter for a Zipf distribution must be larger than 1. &#39;</span>
<span class="s1">&#39;Received </span><span class="si">%r</span><span class="s1">.&#39;</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_distribution_parameter</span><span class="p">)</span>
<span class="k">else</span><span class="p">:</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_distribution_parameter</span> <span class="o">=</span> <span class="mi">0</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_dynamic_splitting</span> <span class="o">=</span> <span class="p">(</span>
<span class="s1">&#39;none&#39;</span> <span class="k">if</span> <span class="p">(</span>
<span class="s1">&#39;splitPointFrequencyRecords&#39;</span> <span class="ow">in</span> <span class="n">input_spec</span>
<span class="ow">and</span> <span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;splitPointFrequencyRecords&#39;</span><span class="p">]</span> <span class="o">==</span> <span class="mi">0</span><span class="p">)</span>
<span class="k">else</span> <span class="s1">&#39;perfect&#39;</span><span class="p">)</span>
<span class="k">if</span> <span class="s1">&#39;delayDistribution&#39;</span> <span class="ow">in</span> <span class="n">input_spec</span><span class="p">:</span>
<span class="k">if</span> <span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;delayDistribution&#39;</span><span class="p">][</span><span class="s1">&#39;type&#39;</span><span class="p">]</span> <span class="o">!=</span> <span class="s1">&#39;const&#39;</span><span class="p">:</span>
<span class="k">raise</span> <span class="ne">ValueError</span><span class="p">(</span><span class="s1">&#39;SyntheticSource currently only supports delay &#39;</span>
<span class="s1">&#39;distributions of type </span><span class="se">\&#39;</span><span class="s1">const</span><span class="se">\&#39;</span><span class="s1">. Received </span><span class="si">%s</span><span class="s1">.&#39;</span><span class="p">,</span>
<span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;delayDistribution&#39;</span><span class="p">][</span><span class="s1">&#39;type&#39;</span><span class="p">])</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span> <span class="o">=</span> <span class="p">(</span>
<span class="nb">float</span><span class="p">(</span><span class="n">input_spec</span><span class="p">[</span><span class="s1">&#39;delayDistribution&#39;</span><span class="p">][</span><span class="s1">&#39;const&#39;</span><span class="p">])</span> <span class="o">/</span> <span class="mi">1000</span><span class="p">)</span>
<span class="k">if</span> <span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span> <span class="ow">and</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span> <span class="o">&lt;</span> <span class="mf">1e-3</span><span class="p">):</span>
<span class="k">raise</span> <span class="ne">ValueError</span><span class="p">(</span><span class="s1">&#39;Sleep time per input record must be at least 1e-3.&#39;</span>
<span class="s1">&#39; Received: </span><span class="si">%r</span><span class="s1">&#39;</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span><span class="p">)</span>
<span class="k">else</span><span class="p">:</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span> <span class="o">=</span> <span class="mi">0</span>
<span class="nd">@property</span>
<span class="k">def</span> <span class="nf">element_size</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">_key_size</span> <span class="o">+</span> <span class="bp">self</span><span class="o">.</span><span class="n">_value_size</span>
<div class="viewcode-block" id="SyntheticSource.estimate_size"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource.estimate_size">[docs]</a> <span class="k">def</span> <span class="nf">estimate_size</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">_total_size</span></div>
<div class="viewcode-block" id="SyntheticSource.split"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource.split">[docs]</a> <span class="k">def</span> <span class="nf">split</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">desired_bundle_size</span><span class="p">,</span> <span class="n">start_position</span><span class="o">=</span><span class="mi">0</span><span class="p">,</span> <span class="n">stop_position</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
<span class="c1"># Performs initial splitting of SyntheticSource.</span>
<span class="c1">#</span>
<span class="c1"># Exact sizes and distribution of initial splits generated here depends on</span>
<span class="c1"># the input specification of the SyntheticSource.</span>
<span class="k">if</span> <span class="n">stop_position</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span>
<span class="n">stop_position</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting</span> <span class="o">==</span> <span class="s1">&#39;zipf&#39;</span><span class="p">:</span>
<span class="n">desired_num_bundles</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_num_bundles</span> <span class="ow">or</span> <span class="n">math</span><span class="o">.</span><span class="n">ceil</span><span class="p">(</span>
<span class="nb">float</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">estimate_size</span><span class="p">())</span> <span class="o">/</span> <span class="n">desired_bundle_size</span><span class="p">)</span>
<span class="n">samples</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">random</span><span class="o">.</span><span class="n">zipf</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_distribution_parameter</span><span class="p">,</span>
<span class="n">desired_num_bundles</span><span class="p">)</span>
<span class="n">total</span> <span class="o">=</span> <span class="nb">sum</span><span class="p">(</span><span class="n">samples</span><span class="p">)</span>
<span class="n">relative_bundle_sizes</span> <span class="o">=</span> <span class="p">[(</span><span class="nb">float</span><span class="p">(</span><span class="n">sample</span><span class="p">)</span> <span class="o">/</span> <span class="n">total</span><span class="p">)</span> <span class="k">for</span> <span class="n">sample</span> <span class="ow">in</span> <span class="n">samples</span><span class="p">]</span>
<span class="n">bundle_ranges</span> <span class="o">=</span> <span class="p">[]</span>
<span class="n">start</span> <span class="o">=</span> <span class="n">start_position</span>
<span class="n">index</span> <span class="o">=</span> <span class="mi">0</span>
<span class="k">while</span> <span class="n">start</span> <span class="o">&lt;</span> <span class="n">stop_position</span><span class="p">:</span>
<span class="k">if</span> <span class="n">index</span> <span class="o">==</span> <span class="n">desired_num_bundles</span> <span class="o">-</span> <span class="mi">1</span><span class="p">:</span>
<span class="n">bundle_ranges</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">start</span><span class="p">,</span> <span class="n">stop_position</span><span class="p">))</span>
<span class="k">break</span>
<span class="n">stop</span> <span class="o">=</span> <span class="n">start</span> <span class="o">+</span> <span class="nb">int</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span> <span class="o">*</span> <span class="n">relative_bundle_sizes</span><span class="p">[</span><span class="n">index</span><span class="p">])</span>
<span class="n">bundle_ranges</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">start</span><span class="p">,</span> <span class="n">stop</span><span class="p">))</span>
<span class="n">start</span> <span class="o">=</span> <span class="n">stop</span>
<span class="n">index</span> <span class="o">+=</span> <span class="mi">1</span>
<span class="k">else</span><span class="p">:</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_num_bundles</span><span class="p">:</span>
<span class="n">bundle_size_in_elements</span> <span class="o">=</span> <span class="nb">max</span><span class="p">(</span><span class="mi">1</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span> <span class="o">/</span>
<span class="bp">self</span><span class="o">.</span><span class="n">_initial_splitting_num_bundles</span><span class="p">)</span>
<span class="k">else</span><span class="p">:</span>
<span class="n">bundle_size_in_elements</span> <span class="o">=</span> <span class="p">(</span><span class="nb">max</span><span class="p">(</span>
<span class="n">div_round_up</span><span class="p">(</span><span class="n">desired_bundle_size</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">element_size</span><span class="p">),</span>
<span class="n">math</span><span class="o">.</span><span class="n">floor</span><span class="p">(</span><span class="n">math</span><span class="o">.</span><span class="n">sqrt</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span><span class="p">))))</span>
<span class="n">bundle_ranges</span> <span class="o">=</span> <span class="p">[]</span>
<span class="k">for</span> <span class="n">start</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">start_position</span><span class="p">,</span> <span class="n">stop_position</span><span class="p">,</span>
<span class="n">bundle_size_in_elements</span><span class="p">):</span>
<span class="n">stop</span> <span class="o">=</span> <span class="nb">min</span><span class="p">(</span><span class="n">start</span> <span class="o">+</span> <span class="n">bundle_size_in_elements</span><span class="p">,</span> <span class="n">stop_position</span><span class="p">)</span>
<span class="n">bundle_ranges</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">start</span><span class="p">,</span> <span class="n">stop</span><span class="p">))</span>
<span class="k">for</span> <span class="n">start</span><span class="p">,</span> <span class="n">stop</span> <span class="ow">in</span> <span class="n">bundle_ranges</span><span class="p">:</span>
<span class="k">yield</span> <span class="n">iobase</span><span class="o">.</span><span class="n">SourceBundle</span><span class="p">(</span><span class="n">stop</span> <span class="o">-</span> <span class="n">start</span><span class="p">,</span> <span class="bp">self</span><span class="p">,</span> <span class="n">start</span><span class="p">,</span> <span class="n">stop</span><span class="p">)</span></div>
<div class="viewcode-block" id="SyntheticSource.get_range_tracker"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource.get_range_tracker">[docs]</a> <span class="k">def</span> <span class="nf">get_range_tracker</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">start_position</span><span class="p">,</span> <span class="n">stop_position</span><span class="p">):</span>
<span class="k">if</span> <span class="n">start_position</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span>
<span class="n">start_position</span> <span class="o">=</span> <span class="mi">0</span>
<span class="k">if</span> <span class="n">stop_position</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span>
<span class="n">stop_position</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_num_records</span>
<span class="n">tracker</span> <span class="o">=</span> <span class="n">range_trackers</span><span class="o">.</span><span class="n">OffsetRangeTracker</span><span class="p">(</span><span class="n">start_position</span><span class="p">,</span> <span class="n">stop_position</span><span class="p">)</span>
<span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_dynamic_splitting</span> <span class="o">==</span> <span class="s1">&#39;none&#39;</span><span class="p">:</span>
<span class="n">tracker</span> <span class="o">=</span> <span class="n">range_trackers</span><span class="o">.</span><span class="n">UnsplittableRangeTracker</span><span class="p">(</span><span class="n">tracker</span><span class="p">)</span>
<span class="k">return</span> <span class="n">tracker</span></div>
<div class="viewcode-block" id="SyntheticSource.read"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource.read">[docs]</a> <span class="k">def</span> <span class="nf">read</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">range_tracker</span><span class="p">):</span>
<span class="n">index</span> <span class="o">=</span> <span class="n">range_tracker</span><span class="o">.</span><span class="n">start_position</span><span class="p">()</span>
<span class="k">while</span> <span class="n">range_tracker</span><span class="o">.</span><span class="n">try_claim</span><span class="p">(</span><span class="n">index</span><span class="p">):</span>
<span class="n">r</span> <span class="o">=</span> <span class="n">np</span><span class="o">.</span><span class="n">random</span><span class="o">.</span><span class="n">RandomState</span><span class="p">(</span><span class="n">index</span><span class="p">)</span>
<span class="n">time</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_sleep_per_input_record_sec</span><span class="p">)</span>
<span class="k">yield</span> <span class="n">r</span><span class="o">.</span><span class="n">bytes</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_key_size</span><span class="p">),</span> <span class="n">r</span><span class="o">.</span><span class="n">bytes</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_value_size</span><span class="p">)</span>
<span class="n">index</span> <span class="o">+=</span> <span class="mi">1</span></div>
<div class="viewcode-block" id="SyntheticSource.default_output_coder"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SyntheticSource.default_output_coder">[docs]</a> <span class="k">def</span> <span class="nf">default_output_coder</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="k">return</span> <span class="n">beam</span><span class="o">.</span><span class="n">coders</span><span class="o">.</span><span class="n">TupleCoder</span><span class="p">(</span>
<span class="p">[</span><span class="n">beam</span><span class="o">.</span><span class="n">coders</span><span class="o">.</span><span class="n">BytesCoder</span><span class="p">(),</span> <span class="n">beam</span><span class="o">.</span><span class="n">coders</span><span class="o">.</span><span class="n">BytesCoder</span><span class="p">()])</span></div></div>
<div class="viewcode-block" id="ShuffleBarrier"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.ShuffleBarrier">[docs]</a><span class="k">class</span> <span class="nc">ShuffleBarrier</span><span class="p">(</span><span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="p">):</span>
<div class="viewcode-block" id="ShuffleBarrier.expand"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.ShuffleBarrier.expand">[docs]</a> <span class="k">def</span> <span class="nf">expand</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">pc</span><span class="p">):</span>
<span class="k">return</span> <span class="p">(</span><span class="n">pc</span>
<span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="n">rotate_key</span><span class="p">)</span>
<span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">GroupByKey</span><span class="p">()</span>
<span class="o">|</span> <span class="s1">&#39;Ungroup&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">FlatMap</span><span class="p">(</span>
<span class="k">lambda</span> <span class="n">elm</span><span class="p">:</span> <span class="p">[(</span><span class="n">elm</span><span class="p">[</span><span class="mi">0</span><span class="p">],</span> <span class="n">v</span><span class="p">)</span> <span class="k">for</span> <span class="n">v</span> <span class="ow">in</span> <span class="n">elm</span><span class="p">[</span><span class="mi">1</span><span class="p">]]))</span></div></div>
<div class="viewcode-block" id="SideInputBarrier"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SideInputBarrier">[docs]</a><span class="k">class</span> <span class="nc">SideInputBarrier</span><span class="p">(</span><span class="n">beam</span><span class="o">.</span><span class="n">PTransform</span><span class="p">):</span>
<div class="viewcode-block" id="SideInputBarrier.expand"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.SideInputBarrier.expand">[docs]</a> <span class="k">def</span> <span class="nf">expand</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">pc</span><span class="p">):</span>
<span class="k">return</span> <span class="p">(</span><span class="n">pc</span>
<span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="n">rotate_key</span><span class="p">)</span>
<span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span>
<span class="k">lambda</span> <span class="n">elem</span><span class="p">,</span> <span class="n">ignored</span><span class="p">:</span> <span class="n">elem</span><span class="p">,</span>
<span class="n">beam</span><span class="o">.</span><span class="n">pvalue</span><span class="o">.</span><span class="n">AsIter</span><span class="p">(</span><span class="n">pc</span> <span class="o">|</span> <span class="n">beam</span><span class="o">.</span><span class="n">FlatMap</span><span class="p">(</span><span class="k">lambda</span> <span class="n">elem</span><span class="p">:</span> <span class="kc">None</span><span class="p">))))</span></div></div>
<div class="viewcode-block" id="merge_using_gbk"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.merge_using_gbk">[docs]</a><span class="k">def</span> <span class="nf">merge_using_gbk</span><span class="p">(</span><span class="n">name</span><span class="p">,</span> <span class="n">pc1</span><span class="p">,</span> <span class="n">pc2</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Merges two given PCollections using a CoGroupByKey.&quot;&quot;&quot;</span>
<span class="n">pc1_with_key</span> <span class="o">=</span> <span class="n">pc1</span> <span class="o">|</span> <span class="p">(</span><span class="n">name</span> <span class="o">+</span> <span class="s1">&#39;AttachKey1&#39;</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="k">lambda</span> <span class="n">x</span><span class="p">:</span> <span class="p">(</span><span class="n">x</span><span class="p">,</span> <span class="n">x</span><span class="p">))</span>
<span class="n">pc2_with_key</span> <span class="o">=</span> <span class="n">pc2</span> <span class="o">|</span> <span class="p">(</span><span class="n">name</span> <span class="o">+</span> <span class="s1">&#39;AttachKey2&#39;</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="k">lambda</span> <span class="n">x</span><span class="p">:</span> <span class="p">(</span><span class="n">x</span><span class="p">,</span> <span class="n">x</span><span class="p">))</span>
<span class="n">grouped</span> <span class="o">=</span> <span class="p">(</span>
<span class="p">{</span><span class="s1">&#39;pc1&#39;</span><span class="p">:</span> <span class="n">pc1_with_key</span><span class="p">,</span> <span class="s1">&#39;pc2&#39;</span><span class="p">:</span> <span class="n">pc2_with_key</span><span class="p">}</span> <span class="o">|</span>
<span class="p">(</span><span class="n">name</span> <span class="o">+</span> <span class="s1">&#39;Group&#39;</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">CoGroupByKey</span><span class="p">())</span>
<span class="k">return</span> <span class="p">(</span><span class="n">grouped</span> <span class="o">|</span>
<span class="p">(</span><span class="n">name</span> <span class="o">+</span> <span class="s1">&#39;DeDup&#39;</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="k">lambda</span> <span class="n">elm</span><span class="p">:</span> <span class="n">elm</span><span class="p">[</span><span class="mi">0</span><span class="p">]))</span> <span class="c1"># Ignoring values</span></div>
<div class="viewcode-block" id="merge_using_side_input"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.merge_using_side_input">[docs]</a><span class="k">def</span> <span class="nf">merge_using_side_input</span><span class="p">(</span><span class="n">name</span><span class="p">,</span> <span class="n">pc1</span><span class="p">,</span> <span class="n">pc2</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Merges two given PCollections using side inputs.&quot;&quot;&quot;</span>
<span class="k">def</span> <span class="nf">join_fn</span><span class="p">(</span><span class="n">val</span><span class="p">,</span> <span class="n">_</span><span class="p">):</span> <span class="c1"># Ignoring side input</span>
<span class="k">return</span> <span class="n">val</span>
<span class="k">return</span> <span class="n">pc1</span> <span class="o">|</span> <span class="n">name</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">core</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="n">join_fn</span><span class="p">,</span> <span class="n">beam</span><span class="o">.</span><span class="n">pvalue</span><span class="o">.</span><span class="n">AsIter</span><span class="p">(</span><span class="n">pc2</span><span class="p">))</span></div>
<div class="viewcode-block" id="expand_using_gbk"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.expand_using_gbk">[docs]</a><span class="k">def</span> <span class="nf">expand_using_gbk</span><span class="p">(</span><span class="n">name</span><span class="p">,</span> <span class="n">pc</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Expands a given PCollection into two copies using GroupByKey.&quot;&quot;&quot;</span>
<span class="n">ret</span> <span class="o">=</span> <span class="p">[]</span>
<span class="n">ret</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">pc</span> <span class="o">|</span> <span class="p">(</span><span class="s1">&#39;</span><span class="si">%s</span><span class="s1">.a&#39;</span> <span class="o">%</span> <span class="n">name</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">ShuffleBarrier</span><span class="p">()))</span>
<span class="n">ret</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">pc</span> <span class="o">|</span> <span class="p">(</span><span class="s1">&#39;</span><span class="si">%s</span><span class="s1">.b&#39;</span> <span class="o">%</span> <span class="n">name</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">ShuffleBarrier</span><span class="p">()))</span>
<span class="k">return</span> <span class="n">ret</span></div>
<div class="viewcode-block" id="expand_using_second_output"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.expand_using_second_output">[docs]</a><span class="k">def</span> <span class="nf">expand_using_second_output</span><span class="p">(</span><span class="n">name</span><span class="p">,</span> <span class="n">pc</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Expands a given PCollection into two copies using side outputs.&quot;&quot;&quot;</span>
<span class="k">class</span> <span class="nc">ExpandFn</span><span class="p">(</span><span class="n">beam</span><span class="o">.</span><span class="n">DoFn</span><span class="p">):</span>
<span class="k">def</span> <span class="nf">process</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">element</span><span class="p">):</span>
<span class="k">yield</span> <span class="n">beam</span><span class="o">.</span><span class="n">pvalue</span><span class="o">.</span><span class="n">TaggedOutput</span><span class="p">(</span><span class="s1">&#39;second_out&#39;</span><span class="p">,</span> <span class="n">element</span><span class="p">)</span>
<span class="k">yield</span> <span class="n">element</span>
<span class="n">pc1</span><span class="p">,</span> <span class="n">pc2</span> <span class="o">=</span> <span class="p">(</span><span class="n">pc</span> <span class="o">|</span> <span class="n">name</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">ParDo</span><span class="p">(</span>
<span class="n">ExpandFn</span><span class="p">())</span><span class="o">.</span><span class="n">with_outputs</span><span class="p">(</span><span class="s1">&#39;second_out&#39;</span><span class="p">,</span> <span class="n">main</span><span class="o">=</span><span class="s1">&#39;main_out&#39;</span><span class="p">))</span>
<span class="k">return</span> <span class="p">[</span><span class="n">pc1</span><span class="p">,</span> <span class="n">pc2</span><span class="p">]</span></div>
<span class="k">def</span> <span class="nf">_parse_steps</span><span class="p">(</span><span class="n">json_str</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Converts the JSON step description into Python objects.</span>
<span class="sd"> See property &#39;steps&#39; for more details about the JSON step description.</span>
<span class="sd"> Args:</span>
<span class="sd"> json_str: a JSON string that describes the steps.</span>
<span class="sd"> Returns:</span>
<span class="sd"> Information about steps as a list of dictionaries. Each dictionary may have</span>
<span class="sd"> following properties.</span>
<span class="sd"> (1) per_element_delay - amount of delay for each element in seconds.</span>
<span class="sd"> (2) per_bundle_delay - minimum amount of delay for a given step in seconds.</span>
<span class="sd"> (3) output_records_per_input_record - number of output elements generated</span>
<span class="sd"> for each input element to a step.</span>
<span class="sd"> (4) output_filter_ratio - the probability at which a step may filter out a</span>
<span class="sd"> given element by not producing any output for that element.</span>
<span class="sd"> &quot;&quot;&quot;</span>
<span class="n">all_steps</span> <span class="o">=</span> <span class="p">[]</span>
<span class="n">json_data</span> <span class="o">=</span> <span class="n">json</span><span class="o">.</span><span class="n">loads</span><span class="p">(</span><span class="n">json_str</span><span class="p">)</span>
<span class="k">for</span> <span class="n">val</span> <span class="ow">in</span> <span class="n">json_data</span><span class="p">:</span>
<span class="n">steps</span> <span class="o">=</span> <span class="p">{}</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;per_element_delay&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="p">(</span>
<span class="p">(</span><span class="nb">float</span><span class="p">(</span><span class="n">val</span><span class="p">[</span><span class="s1">&#39;per_element_delay_msec&#39;</span><span class="p">])</span> <span class="o">/</span> <span class="mi">1000</span><span class="p">)</span>
<span class="k">if</span> <span class="s1">&#39;per_element_delay_msec&#39;</span> <span class="ow">in</span> <span class="n">val</span> <span class="k">else</span> <span class="mi">0</span><span class="p">)</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;per_bundle_delay&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="p">(</span>
<span class="nb">float</span><span class="p">(</span><span class="n">val</span><span class="p">[</span><span class="s1">&#39;per_bundle_delay_sec&#39;</span><span class="p">])</span>
<span class="k">if</span> <span class="s1">&#39;per_bundle_delay_sec&#39;</span> <span class="ow">in</span> <span class="n">val</span> <span class="k">else</span> <span class="mi">0</span><span class="p">)</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;output_records_per_input_record&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="p">(</span>
<span class="nb">int</span><span class="p">(</span><span class="n">val</span><span class="p">[</span><span class="s1">&#39;output_records_per_input_record&#39;</span><span class="p">])</span>
<span class="k">if</span> <span class="s1">&#39;output_records_per_input_record&#39;</span> <span class="ow">in</span> <span class="n">val</span> <span class="k">else</span> <span class="mi">1</span><span class="p">)</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;output_filter_ratio&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="p">(</span>
<span class="nb">float</span><span class="p">(</span><span class="n">val</span><span class="p">[</span><span class="s1">&#39;output_filter_ratio&#39;</span><span class="p">])</span>
<span class="k">if</span> <span class="s1">&#39;output_filter_ratio&#39;</span> <span class="ow">in</span> <span class="n">val</span> <span class="k">else</span> <span class="mi">0</span><span class="p">)</span>
<span class="n">all_steps</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="n">steps</span><span class="p">)</span>
<span class="k">return</span> <span class="n">all_steps</span>
<div class="viewcode-block" id="parse_args"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.parse_args">[docs]</a><span class="k">def</span> <span class="nf">parse_args</span><span class="p">(</span><span class="n">args</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Parses a given set of arguments.</span>
<span class="sd"> Args:</span>
<span class="sd"> args: set of arguments to be passed.</span>
<span class="sd"> Returns:</span>
<span class="sd"> a tuple where first item gives the set of arguments defined and parsed</span>
<span class="sd"> within this method and second item gives the set of unknown arguments.</span>
<span class="sd"> &quot;&quot;&quot;</span>
<span class="n">parser</span> <span class="o">=</span> <span class="n">argparse</span><span class="o">.</span><span class="n">ArgumentParser</span><span class="p">()</span>
<span class="n">parser</span><span class="o">.</span><span class="n">add_argument</span><span class="p">(</span>
<span class="s1">&#39;--steps&#39;</span><span class="p">,</span>
<span class="n">dest</span><span class="o">=</span><span class="s1">&#39;steps&#39;</span><span class="p">,</span>
<span class="nb">type</span><span class="o">=</span><span class="n">_parse_steps</span><span class="p">,</span>
<span class="n">help</span><span class="o">=</span><span class="s1">&#39;A JSON string that gives a list where each entry of the list is &#39;</span>
<span class="s1">&#39;configuration information for a step. Configuration for each step &#39;</span>
<span class="s1">&#39;consists of &#39;</span>
<span class="s1">&#39;(1) A float &quot;per_bundle_delay_sec&quot; (in seconds). Defaults to 0.&#39;</span>
<span class="s1">&#39;(2) A float &quot;per_element_delay_msec&quot; (in milli seconds). &#39;</span>
<span class="s1">&#39; Defaults to 0.&#39;</span>
<span class="s1">&#39;(3) An integer &quot;output_records_per_input_record&quot;. Defaults to 1.&#39;</span>
<span class="s1">&#39;(4) A float &quot;output_filter_ratio&quot; in the range [0, 1] . &#39;</span>
<span class="s1">&#39; Defaults to 0.&#39;</span><span class="p">)</span>
<span class="n">parser</span><span class="o">.</span><span class="n">add_argument</span><span class="p">(</span>
<span class="s1">&#39;--input&#39;</span><span class="p">,</span>
<span class="n">dest</span><span class="o">=</span><span class="s1">&#39;input&#39;</span><span class="p">,</span>
<span class="nb">type</span><span class="o">=</span><span class="n">json</span><span class="o">.</span><span class="n">loads</span><span class="p">,</span>
<span class="n">help</span><span class="o">=</span><span class="s1">&#39;A JSON string that describes the properties of the SyntheticSource &#39;</span>
<span class="s1">&#39;used by the pipeline. Configuration is similar to Java &#39;</span>
<span class="s1">&#39;SyntheticBoundedInput.&#39;</span>
<span class="s1">&#39;Currently supports following properties. &#39;</span>
<span class="s1">&#39;(1) An integer &quot;numRecords&quot;. &#39;</span>
<span class="s1">&#39;(2) An integer &quot;keySize&quot;. &#39;</span>
<span class="s1">&#39;(3) An integer &quot;valueSize&quot;. &#39;</span>
<span class="s1">&#39;(4) A tuple &quot;bundleSizeDistribution&quot; with following values. &#39;</span>
<span class="s1">&#39; A string &quot;type&quot;. Allowed values are &quot;const&quot; and &quot;zipf&quot;. &#39;</span>
<span class="s1">&#39; An float &quot;param&quot;. Only used if &quot;type&quot;==&quot;zipf&quot;. Must be &#39;</span>
<span class="s1">&#39; larger than 1. &#39;</span>
<span class="s1">&#39;(5) An integer &quot;forceNumInitialBundles&quot;. &#39;</span>
<span class="s1">&#39;(6) An integer &quot;splitPointFrequencyRecords&quot;. &#39;</span>
<span class="s1">&#39;(7) A tuple &quot;delayDistribution&quot; with following values. &#39;</span>
<span class="s1">&#39; A string &quot;type&quot;. Only allowed value is &quot;const&quot;. &#39;</span>
<span class="s1">&#39; An integer &quot;const&quot;. &#39;</span><span class="p">)</span>
<span class="n">parser</span><span class="o">.</span><span class="n">add_argument</span><span class="p">(</span><span class="s1">&#39;--barrier&#39;</span><span class="p">,</span>
<span class="n">dest</span><span class="o">=</span><span class="s1">&#39;barrier&#39;</span><span class="p">,</span>
<span class="n">default</span><span class="o">=</span><span class="s1">&#39;shuffle&#39;</span><span class="p">,</span>
<span class="n">choices</span><span class="o">=</span><span class="p">[</span><span class="s1">&#39;shuffle&#39;</span><span class="p">,</span> <span class="s1">&#39;side-input&#39;</span><span class="p">,</span> <span class="s1">&#39;expand-gbk&#39;</span><span class="p">,</span>
<span class="s1">&#39;expand-second-output&#39;</span><span class="p">,</span> <span class="s1">&#39;merge-gbk&#39;</span><span class="p">,</span>
<span class="s1">&#39;merge-side-input&#39;</span><span class="p">],</span>
<span class="n">help</span><span class="o">=</span><span class="s1">&#39;Whether to use shuffle as the barrier &#39;</span>
<span class="s1">&#39;(as opposed to side inputs).&#39;</span><span class="p">)</span>
<span class="n">parser</span><span class="o">.</span><span class="n">add_argument</span><span class="p">(</span><span class="s1">&#39;--output&#39;</span><span class="p">,</span>
<span class="n">dest</span><span class="o">=</span><span class="s1">&#39;output&#39;</span><span class="p">,</span>
<span class="n">default</span><span class="o">=</span><span class="s1">&#39;&#39;</span><span class="p">,</span>
<span class="n">help</span><span class="o">=</span><span class="s1">&#39;Destination to write output.&#39;</span><span class="p">)</span>
<span class="k">return</span> <span class="n">parser</span><span class="o">.</span><span class="n">parse_known_args</span><span class="p">(</span><span class="n">args</span><span class="p">)</span></div>
<div class="viewcode-block" id="run"><a class="viewcode-back" href="../../../apache_beam.testing.synthetic_pipeline.html#apache_beam.testing.synthetic_pipeline.run">[docs]</a><span class="k">def</span> <span class="nf">run</span><span class="p">(</span><span class="n">argv</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
<span class="sd">&quot;&quot;&quot;Runs the workflow.&quot;&quot;&quot;</span>
<span class="n">known_args</span><span class="p">,</span> <span class="n">pipeline_args</span> <span class="o">=</span> <span class="n">parse_args</span><span class="p">(</span><span class="n">argv</span><span class="p">)</span>
<span class="n">pipeline_options</span> <span class="o">=</span> <span class="n">PipelineOptions</span><span class="p">(</span><span class="n">pipeline_args</span><span class="p">)</span>
<span class="n">pipeline_options</span><span class="o">.</span><span class="n">view_as</span><span class="p">(</span><span class="n">SetupOptions</span><span class="p">)</span><span class="o">.</span><span class="n">save_main_session</span> <span class="o">=</span> <span class="kc">True</span>
<span class="n">input_info</span> <span class="o">=</span> <span class="n">known_args</span><span class="o">.</span><span class="n">input</span>
<span class="k">with</span> <span class="n">TestPipeline</span><span class="p">(</span><span class="n">options</span><span class="o">=</span><span class="n">pipeline_options</span><span class="p">)</span> <span class="k">as</span> <span class="n">p</span><span class="p">:</span>
<span class="n">source</span> <span class="o">=</span> <span class="n">SyntheticSource</span><span class="p">(</span><span class="n">input_info</span><span class="p">)</span>
<span class="c1"># pylint: disable=expression-not-assigned</span>
<span class="n">barrier</span> <span class="o">=</span> <span class="n">known_args</span><span class="o">.</span><span class="n">barrier</span>
<span class="n">pc_list</span> <span class="o">=</span> <span class="p">[]</span>
<span class="n">num_roots</span> <span class="o">=</span> <span class="mi">2</span> <span class="o">**</span> <span class="p">(</span><span class="nb">len</span><span class="p">(</span><span class="n">known_args</span><span class="o">.</span><span class="n">steps</span><span class="p">)</span> <span class="o">-</span> <span class="mi">1</span><span class="p">)</span> <span class="k">if</span> <span class="p">(</span>
<span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;merge-gbk&#39;</span> <span class="ow">or</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;merge-side-input&#39;</span><span class="p">)</span> <span class="k">else</span> <span class="mi">1</span>
<span class="k">for</span> <span class="n">read_no</span> <span class="ow">in</span> <span class="nb">range</span><span class="p">(</span><span class="n">num_roots</span><span class="p">):</span>
<span class="n">pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">((</span><span class="n">p</span> <span class="o">|</span> <span class="p">(</span><span class="s1">&#39;Read </span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="n">read_no</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">io</span><span class="o">.</span><span class="n">Read</span><span class="p">(</span><span class="n">source</span><span class="p">)))</span>
<span class="k">for</span> <span class="n">step_no</span><span class="p">,</span> <span class="n">steps</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">known_args</span><span class="o">.</span><span class="n">steps</span><span class="p">):</span>
<span class="k">if</span> <span class="n">step_no</span> <span class="o">!=</span> <span class="mi">0</span><span class="p">:</span>
<span class="n">new_pc_list</span> <span class="o">=</span> <span class="p">[]</span>
<span class="k">for</span> <span class="n">pc_no</span><span class="p">,</span> <span class="n">pc</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">pc_list</span><span class="p">):</span>
<span class="k">if</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;shuffle&#39;</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
<span class="p">(</span><span class="n">pc</span> <span class="o">|</span>
<span class="p">(</span><span class="s1">&#39;shuffle </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">))</span> <span class="o">&gt;&gt;</span> <span class="n">ShuffleBarrier</span><span class="p">()))</span>
<span class="k">elif</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;side-input&#39;</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
<span class="p">(</span><span class="n">pc</span> <span class="o">|</span>
<span class="p">(</span><span class="s1">&#39;side-input </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">))</span> <span class="o">&gt;&gt;</span> <span class="n">SideInputBarrier</span><span class="p">()))</span>
<span class="k">elif</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;expand-gbk&#39;</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">extend</span><span class="p">(</span>
<span class="n">expand_using_gbk</span><span class="p">((</span><span class="s1">&#39;expand-gbk </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">)),</span> <span class="n">pc</span><span class="p">))</span>
<span class="k">elif</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;expand-second-output&#39;</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">extend</span><span class="p">(</span>
<span class="n">expand_using_second_output</span><span class="p">(</span>
<span class="p">(</span><span class="s1">&#39;expand-second-output </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">)),</span> <span class="n">pc</span><span class="p">))</span>
<span class="k">elif</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;merge-gbk&#39;</span><span class="p">:</span>
<span class="k">if</span> <span class="n">pc_no</span> <span class="o">%</span> <span class="mi">2</span> <span class="o">==</span> <span class="mi">0</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
<span class="n">merge_using_gbk</span><span class="p">((</span><span class="s1">&#39;merge-gbk </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">)),</span>
<span class="n">pc</span><span class="p">,</span> <span class="n">pc_list</span><span class="p">[</span><span class="n">pc_no</span> <span class="o">+</span> <span class="mi">1</span><span class="p">]))</span>
<span class="k">else</span><span class="p">:</span>
<span class="k">continue</span>
<span class="k">elif</span> <span class="n">barrier</span> <span class="o">==</span> <span class="s1">&#39;merge-side-input&#39;</span><span class="p">:</span>
<span class="k">if</span> <span class="n">pc_no</span> <span class="o">%</span> <span class="mi">2</span> <span class="o">==</span> <span class="mi">0</span><span class="p">:</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
<span class="n">merge_using_side_input</span><span class="p">(</span>
<span class="p">(</span><span class="s1">&#39;merge-side-input </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">)),</span>
<span class="n">pc</span><span class="p">,</span> <span class="n">pc_list</span><span class="p">[</span><span class="n">pc_no</span> <span class="o">+</span> <span class="mi">1</span><span class="p">]))</span>
<span class="k">else</span><span class="p">:</span>
<span class="k">continue</span>
<span class="n">pc_list</span> <span class="o">=</span> <span class="n">new_pc_list</span>
<span class="n">new_pc_list</span> <span class="o">=</span> <span class="p">[]</span>
<span class="k">for</span> <span class="n">pc_no</span><span class="p">,</span> <span class="n">pc</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">pc_list</span><span class="p">):</span>
<span class="n">new_pc</span> <span class="o">=</span> <span class="n">pc</span> <span class="o">|</span> <span class="s1">&#39;SyntheticStep </span><span class="si">%d</span><span class="s1">.</span><span class="si">%d</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="n">step_no</span><span class="p">,</span> <span class="n">pc_no</span><span class="p">)</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">ParDo</span><span class="p">(</span>
<span class="n">SyntheticStep</span><span class="p">(</span>
<span class="n">per_element_delay_sec</span><span class="o">=</span><span class="n">steps</span><span class="p">[</span><span class="s1">&#39;per_element_delay&#39;</span><span class="p">],</span>
<span class="n">per_bundle_delay_sec</span><span class="o">=</span><span class="n">steps</span><span class="p">[</span><span class="s1">&#39;per_bundle_delay&#39;</span><span class="p">],</span>
<span class="n">output_records_per_input_record</span><span class="o">=</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;output_records_per_input_record&#39;</span><span class="p">],</span>
<span class="n">output_filter_ratio</span><span class="o">=</span>
<span class="n">steps</span><span class="p">[</span><span class="s1">&#39;output_filter_ratio&#39;</span><span class="p">]))</span>
<span class="n">new_pc_list</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="n">new_pc</span><span class="p">)</span>
<span class="n">pc_list</span> <span class="o">=</span> <span class="n">new_pc_list</span>
<span class="k">if</span> <span class="n">known_args</span><span class="o">.</span><span class="n">output</span><span class="p">:</span>
<span class="c1"># If an output location is provided we format and write output.</span>
<span class="k">if</span> <span class="nb">len</span><span class="p">(</span><span class="n">pc_list</span><span class="p">)</span> <span class="o">==</span> <span class="mi">1</span><span class="p">:</span>
<span class="p">(</span><span class="n">pc_list</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o">|</span>
<span class="s1">&#39;FormatOutput&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">beam</span><span class="o">.</span><span class="n">Map</span><span class="p">(</span><span class="k">lambda</span> <span class="n">elm</span><span class="p">:</span> <span class="p">(</span><span class="n">elm</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o">+</span> <span class="n">elm</span><span class="p">[</span><span class="mi">1</span><span class="p">]))</span> <span class="o">|</span>
<span class="s1">&#39;WriteOutput&#39;</span> <span class="o">&gt;&gt;</span> <span class="n">WriteToText</span><span class="p">(</span><span class="n">known_args</span><span class="o">.</span><span class="n">output</span><span class="p">))</span>
<span class="n">logging</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s1">&#39;Pipeline run completed.&#39;</span><span class="p">)</span></div>
<span class="k">if</span> <span class="vm">__name__</span> <span class="o">==</span> <span class="s1">&#39;__main__&#39;</span><span class="p">:</span>
<span class="n">logging</span><span class="o">.</span><span class="n">getLogger</span><span class="p">()</span><span class="o">.</span><span class="n">setLevel</span><span class="p">(</span><span class="n">logging</span><span class="o">.</span><span class="n">INFO</span><span class="p">)</span>
<span class="n">run</span><span class="p">()</span>
</pre></div>
</div>
<div class="articleComments">
</div>
</div>
<footer>
<hr/>
<div role="contentinfo">
<p>
&copy; Copyright .
</p>
</div>
Built with <a href="http://sphinx-doc.org/">Sphinx</a> using a <a href="https://github.com/snide/sphinx_rtd_theme">theme</a> provided by <a href="https://readthedocs.org">Read the Docs</a>.
</footer>
</div>
</div>
</section>
</div>
<script type="text/javascript">
var DOCUMENTATION_OPTIONS = {
URL_ROOT:'../../../',
VERSION:'',
COLLAPSE_INDEX:false,
FILE_SUFFIX:'.html',
HAS_SOURCE: true,
SOURCELINK_SUFFIX: '.txt'
};
</script>
<script type="text/javascript" src="../../../_static/jquery.js"></script>
<script type="text/javascript" src="../../../_static/underscore.js"></script>
<script type="text/javascript" src="../../../_static/doctools.js"></script>
<script type="text/javascript" src="../../../_static/js/theme.js"></script>
<script type="text/javascript">
jQuery(function () {
SphinxRtdTheme.StickyNav.enable();
});
</script>
</body>
</html>