pydoc/2.47.0/_modules/apache_beam/runners/interactive/display/pipeline_graph.html - beam-site - Git at Google



 <!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.runners.interactive.display.pipeline_graph &mdash; Apache Beam 2.47.0 documentation</title>


   <script type="text/javascript" src="../../../../../_static/js/modernizr.min.js"></script>


       <script type="text/javascript" id="documentation_options" data-url_root="../../../../../" src="../../../../../_static/documentation_options.js"></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/language_data.js"></script>
         <script async="async" type="text/javascript" src="https://cdnjs.cloudflare.com/ajax/libs/mathjax/2.7.5/latest.js?config=TeX-AMS-MML_HTMLorMML"></script>

     <script type="text/javascript" src="../../../../../_static/js/theme.js"></script>


   <link rel="stylesheet" href="../../../../../_static/css/theme.css" type="text/css" />
   <link rel="stylesheet" href="../../../../../_static/pygments.css" type="text/css" />
     <link rel="index" title="Index" href="../../../../../genindex.html" />
     <link rel="search" title="Search" href="../../../../../search.html" />
 </head>

 <body class="wy-body-for-nav">


   <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 class="version">
                 2.47.0
               </div>


 <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.dataframe.html">apache_beam.dataframe 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.ml.html">apache_beam.ml 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.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>
 <li class="toctree-l1"><a class="reference internal" href="../../../../../apache_beam.yaml.html">apache_beam.yaml 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>
 </ul>


         </div>
       </div>
     </nav>

     <section data-toggle="wy-nav-shift" class="wy-nav-content-wrap">


       <nav class="wy-nav-top" 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.runners.interactive.display.pipeline_graph</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.runners.interactive.display.pipeline_graph</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;For generating Beam pipeline graph in DOT representation.</span>

 <span class="sd">This module is experimental. No backwards-compatibility guarantees.</span>
 <span class="sd">&quot;&quot;&quot;</span>

 <span class="c1"># pytype: skip-file</span>

 <span class="kn">import</span> <span class="nn">collections</span>
 <span class="kn">import</span> <span class="nn">logging</span>
 <span class="kn">import</span> <span class="nn">threading</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">DefaultDict</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">Dict</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">Iterator</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">List</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">Tuple</span>
 <span class="kn">from</span> <span class="nn">typing</span> <span class="kn">import</span> <span class="n">Union</span>

 <span class="kn">import</span> <span class="nn">pydot</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.portability.api</span> <span class="kn">import</span> <span class="n">beam_runner_api_pb2</span>
 <span class="kn">from</span> <span class="nn">apache_beam.runners.interactive</span> <span class="kn">import</span> <span class="n">interactive_environment</span> <span class="k">as</span> <span class="n">ie</span>
 <span class="kn">from</span> <span class="nn">apache_beam.runners.interactive</span> <span class="kn">import</span> <span class="n">pipeline_instrument</span> <span class="k">as</span> <span class="n">inst</span>
 <span class="kn">from</span> <span class="nn">apache_beam.runners.interactive.display</span> <span class="kn">import</span> <span class="n">pipeline_graph_renderer</span>

 <span class="c1"># pylint does not understand context</span>
 <span class="c1"># pylint:disable=dangerous-default-value</span>


 <div class="viewcode-block" id="PipelineGraph"><a class="viewcode-back" href="../../../../../apache_beam.runners.interactive.display.pipeline_graph.html#apache_beam.runners.interactive.display.pipeline_graph.PipelineGraph">[docs]</a><span class="k">class</span> <span class="nc">PipelineGraph</span><span class="p">(</span><span class="nb">object</span><span class="p">):</span>
 <span class="w">  </span><span class="sd">&quot;&quot;&quot;Creates a DOT representing the pipeline. Thread-safe. Runner agnostic.&quot;&quot;&quot;</span>
   <span class="k">def</span> <span class="fm">__init__</span><span class="p">(</span>
       <span class="bp">self</span><span class="p">,</span>
       <span class="n">pipeline</span><span class="p">,</span>  <span class="c1"># type: Union[beam_runner_api_pb2.Pipeline, beam.Pipeline]</span>
       <span class="n">default_vertex_attrs</span><span class="o">=</span><span class="p">{</span><span class="s1">&#39;shape&#39;</span><span class="p">:</span> <span class="s1">&#39;box&#39;</span><span class="p">},</span>
       <span class="n">default_edge_attrs</span><span class="o">=</span><span class="kc">None</span><span class="p">,</span>
       <span class="n">render_option</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Constructor of PipelineGraph.</span>

 <span class="sd">    Examples:</span>
 <span class="sd">      graph = pipeline_graph.PipelineGraph(pipeline_proto)</span>
 <span class="sd">      graph.get_dot()</span>

 <span class="sd">      or</span>

 <span class="sd">      graph = pipeline_graph.PipelineGraph(pipeline)</span>
 <span class="sd">      graph.get_dot()</span>

 <span class="sd">    Args:</span>
 <span class="sd">      pipeline: (Pipeline proto) or (Pipeline) pipeline to be rendered.</span>
 <span class="sd">      default_vertex_attrs: (Dict[str, str]) a dict of default vertex attributes</span>
 <span class="sd">      default_edge_attrs: (Dict[str, str]) a dict of default edge attributes</span>
 <span class="sd">      render_option: (str) this parameter decides how the pipeline graph is</span>
 <span class="sd">          rendered. See display.pipeline_graph_renderer for available options.</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_lock</span> <span class="o">=</span> <span class="n">threading</span><span class="o">.</span><span class="n">Lock</span><span class="p">()</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span> <span class="o">=</span> <span class="kc">None</span>  <span class="c1"># type: pydot.Dot</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_instrument</span> <span class="o">=</span> <span class="kc">None</span>
     <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">pipeline</span><span class="p">,</span> <span class="n">beam</span><span class="o">.</span><span class="n">Pipeline</span><span class="p">):</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_instrument</span> <span class="o">=</span> <span class="n">inst</span><span class="o">.</span><span class="n">PipelineInstrument</span><span class="p">(</span>
           <span class="n">pipeline</span><span class="p">,</span> <span class="n">pipeline</span><span class="o">.</span><span class="n">_options</span><span class="p">)</span>
       <span class="c1"># The pre-process links user pipeline to runner pipeline through analysis</span>
       <span class="c1"># but without mutating runner pipeline.</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_instrument</span><span class="o">.</span><span class="n">preprocess</span><span class="p">()</span>

     <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">pipeline</span><span class="p">,</span> <span class="n">beam_runner_api_pb2</span><span class="o">.</span><span class="n">Pipeline</span><span class="p">):</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_proto</span> <span class="o">=</span> <span class="n">pipeline</span>
     <span class="k">elif</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">pipeline</span><span class="p">,</span> <span class="n">beam</span><span class="o">.</span><span class="n">Pipeline</span><span class="p">):</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_proto</span> <span class="o">=</span> <span class="n">pipeline</span><span class="o">.</span><span class="n">to_runner_api</span><span class="p">()</span>
     <span class="k">else</span><span class="p">:</span>
       <span class="k">raise</span> <span class="ne">TypeError</span><span class="p">(</span>
           <span class="s1">&#39;pipeline should either be a </span><span class="si">%s</span><span class="s1"> or </span><span class="si">%s</span><span class="s1">, while </span><span class="si">%s</span><span class="s1"> is given&#39;</span> <span class="o">%</span>
           <span class="p">(</span><span class="n">beam_runner_api_pb2</span><span class="o">.</span><span class="n">Pipeline</span><span class="p">,</span> <span class="n">beam</span><span class="o">.</span><span class="n">Pipeline</span><span class="p">,</span> <span class="nb">type</span><span class="p">(</span><span class="n">pipeline</span><span class="p">)))</span>

     <span class="c1"># A dict from PCollection ID to a list of its consuming Transform IDs</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_consumers</span> <span class="o">=</span> <span class="n">collections</span><span class="o">.</span><span class="n">defaultdict</span><span class="p">(</span>
         <span class="nb">list</span><span class="p">)</span>  <span class="c1"># type: DefaultDict[str, List[str]]</span>
     <span class="c1"># A dict from PCollection ID to its producing Transform ID</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_producers</span> <span class="o">=</span> <span class="p">{}</span>  <span class="c1"># type: Dict[str, str]</span>

     <span class="k">for</span> <span class="n">transform_id</span><span class="p">,</span> <span class="n">transform_proto</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_top_level_transforms</span><span class="p">():</span>
       <span class="k">for</span> <span class="n">pcoll_id</span> <span class="ow">in</span> <span class="n">transform_proto</span><span class="o">.</span><span class="n">inputs</span><span class="o">.</span><span class="n">values</span><span class="p">():</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_consumers</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="n">transform_id</span><span class="p">)</span>
       <span class="k">for</span> <span class="n">pcoll_id</span> <span class="ow">in</span> <span class="n">transform_proto</span><span class="o">.</span><span class="n">outputs</span><span class="o">.</span><span class="n">values</span><span class="p">():</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_producers</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]</span> <span class="o">=</span> <span class="n">transform_id</span>

     <span class="n">default_vertex_attrs</span> <span class="o">=</span> <span class="n">default_vertex_attrs</span> <span class="ow">or</span> <span class="p">{</span><span class="s1">&#39;shape&#39;</span><span class="p">:</span> <span class="s1">&#39;box&#39;</span><span class="p">}</span>
     <span class="k">if</span> <span class="s1">&#39;color&#39;</span> <span class="ow">not</span> <span class="ow">in</span> <span class="n">default_vertex_attrs</span><span class="p">:</span>
       <span class="n">default_vertex_attrs</span><span class="p">[</span><span class="s1">&#39;color&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="s1">&#39;blue&#39;</span>
     <span class="k">if</span> <span class="s1">&#39;fontcolor&#39;</span> <span class="ow">not</span> <span class="ow">in</span> <span class="n">default_vertex_attrs</span><span class="p">:</span>
       <span class="n">default_vertex_attrs</span><span class="p">[</span><span class="s1">&#39;fontcolor&#39;</span><span class="p">]</span> <span class="o">=</span> <span class="s1">&#39;blue&#39;</span>

     <span class="n">vertex_dict</span><span class="p">,</span> <span class="n">edge_dict</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_generate_graph_dicts</span><span class="p">()</span>
     <span class="bp">self</span><span class="o">.</span><span class="n">_construct_graph</span><span class="p">(</span>
         <span class="n">vertex_dict</span><span class="p">,</span> <span class="n">edge_dict</span><span class="p">,</span> <span class="n">default_vertex_attrs</span><span class="p">,</span> <span class="n">default_edge_attrs</span><span class="p">)</span>

     <span class="bp">self</span><span class="o">.</span><span class="n">_renderer</span> <span class="o">=</span> <span class="n">pipeline_graph_renderer</span><span class="o">.</span><span class="n">get_renderer</span><span class="p">(</span><span class="n">render_option</span><span class="p">)</span>

 <div class="viewcode-block" id="PipelineGraph.get_dot"><a class="viewcode-back" href="../../../../../apache_beam.runners.interactive.display.pipeline_graph.html#apache_beam.runners.interactive.display.pipeline_graph.PipelineGraph.get_dot">[docs]</a>  <span class="k">def</span> <span class="nf">get_dot</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
     <span class="c1"># type: () -&gt; str</span>
     <span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">_get_graph</span><span class="p">()</span><span class="o">.</span><span class="n">to_string</span><span class="p">()</span></div>

 <div class="viewcode-block" id="PipelineGraph.display_graph"><a class="viewcode-back" href="../../../../../apache_beam.runners.interactive.display.pipeline_graph.html#apache_beam.runners.interactive.display.pipeline_graph.PipelineGraph.display_graph">[docs]</a>  <span class="k">def</span> <span class="nf">display_graph</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Displays the graph generated.&quot;&quot;&quot;</span>
     <span class="n">rendered_graph</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_renderer</span><span class="o">.</span><span class="n">render_pipeline_graph</span><span class="p">(</span><span class="bp">self</span><span class="p">)</span>
     <span class="k">if</span> <span class="n">ie</span><span class="o">.</span><span class="n">current_env</span><span class="p">()</span><span class="o">.</span><span class="n">is_in_notebook</span><span class="p">:</span>
       <span class="k">try</span><span class="p">:</span>
         <span class="kn">from</span> <span class="nn">IPython</span> <span class="kn">import</span> <span class="n">display</span>
         <span class="n">display</span><span class="o">.</span><span class="n">display</span><span class="p">(</span><span class="n">display</span><span class="o">.</span><span class="n">HTML</span><span class="p">(</span><span class="n">rendered_graph</span><span class="p">))</span>
       <span class="k">except</span> <span class="ne">ImportError</span><span class="p">:</span>  <span class="c1"># Unlikely to happen when is_in_notebook.</span>
         <span class="n">logging</span><span class="o">.</span><span class="n">warning</span><span class="p">(</span>
             <span class="s1">&#39;Failed to import IPython display module when current &#39;</span>
             <span class="s1">&#39;environment is in a notebook. Cannot display the &#39;</span>
             <span class="s1">&#39;pipeline graph.&#39;</span><span class="p">)</span></div>

   <span class="k">def</span> <span class="nf">_top_level_transforms</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
     <span class="c1"># type: () -&gt; Iterator[Tuple[str, beam_runner_api_pb2.PTransform]]</span>

 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Yields all top level PTransforms (subtransforms of the root PTransform).</span>

 <span class="sd">    Yields: (str, PTransform proto) ID, proto pair of top level PTransforms.</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="n">transforms</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_proto</span><span class="o">.</span><span class="n">components</span><span class="o">.</span><span class="n">transforms</span>
     <span class="k">for</span> <span class="n">root_transform_id</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_proto</span><span class="o">.</span><span class="n">root_transform_ids</span><span class="p">:</span>
       <span class="n">root_transform_proto</span> <span class="o">=</span> <span class="n">transforms</span><span class="p">[</span><span class="n">root_transform_id</span><span class="p">]</span>
       <span class="k">for</span> <span class="n">top_level_transform_id</span> <span class="ow">in</span> <span class="n">root_transform_proto</span><span class="o">.</span><span class="n">subtransforms</span><span class="p">:</span>
         <span class="n">top_level_transform_proto</span> <span class="o">=</span> <span class="n">transforms</span><span class="p">[</span><span class="n">top_level_transform_id</span><span class="p">]</span>
         <span class="k">yield</span> <span class="n">top_level_transform_id</span><span class="p">,</span> <span class="n">top_level_transform_proto</span>

   <span class="k">def</span> <span class="nf">_decorate</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">value</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Decorates label-ish values used for rendering in dot language.</span>

 <span class="sd">    Escapes special characters in the given str value for dot language. All</span>
 <span class="sd">    PTransform unique names are escaped implicitly in this module when building</span>
 <span class="sd">    dot representation. Otherwise, special characters will break the graph</span>
 <span class="sd">    rendered or cause runtime errors.</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="c1"># Replace py str literal `\\` which is `\` in dot with py str literal</span>
     <span class="c1"># `\\\\` which is `\\` in dot so that dot `\\` can be rendered as `\`. Then</span>
     <span class="c1"># replace `&quot;` with `\\&quot;` so that the dot generated will be `\&quot;` and be</span>
     <span class="c1"># rendered as `&quot;`.</span>
     <span class="k">return</span> <span class="s1">&#39;&quot;</span><span class="si">{}</span><span class="s1">&quot;&#39;</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">value</span><span class="o">.</span><span class="n">replace</span><span class="p">(</span><span class="s1">&#39;</span><span class="se">\\</span><span class="s1">&#39;</span><span class="p">,</span> <span class="s1">&#39;</span><span class="se">\\\\</span><span class="s1">&#39;</span><span class="p">)</span><span class="o">.</span><span class="n">replace</span><span class="p">(</span><span class="s1">&#39;&quot;&#39;</span><span class="p">,</span> <span class="s1">&#39;</span><span class="se">\\</span><span class="s1">&quot;&#39;</span><span class="p">))</span>

   <span class="k">def</span> <span class="nf">_generate_graph_dicts</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;From pipeline_proto and other info, generate the graph.</span>

 <span class="sd">    Returns:</span>
 <span class="sd">      vertex_dict: (Dict[str, Dict[str, str]]) vertex mapped to attributes.</span>
 <span class="sd">      edge_dict: (Dict[(str, str), Dict[str, str]]) vertex pair mapped to the</span>
 <span class="sd">          edge&#39;s attribute.</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="n">transforms</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_proto</span><span class="o">.</span><span class="n">components</span><span class="o">.</span><span class="n">transforms</span>

     <span class="c1"># A dict from vertex name (i.e. PCollection ID) to its attributes.</span>
     <span class="n">vertex_dict</span> <span class="o">=</span> <span class="n">collections</span><span class="o">.</span><span class="n">defaultdict</span><span class="p">(</span><span class="nb">dict</span><span class="p">)</span>
     <span class="c1"># A dict from vertex name pairs defining the edge (i.e. a pair of PTransform</span>
     <span class="c1"># IDs defining the PCollection) to its attributes.</span>
     <span class="n">edge_dict</span> <span class="o">=</span> <span class="n">collections</span><span class="o">.</span><span class="n">defaultdict</span><span class="p">(</span><span class="nb">dict</span><span class="p">)</span>

     <span class="bp">self</span><span class="o">.</span><span class="n">_edge_to_vertex_pairs</span> <span class="o">=</span> <span class="n">collections</span><span class="o">.</span><span class="n">defaultdict</span><span class="p">(</span><span class="nb">list</span><span class="p">)</span>

     <span class="k">for</span> <span class="n">_</span><span class="p">,</span> <span class="n">transform</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_top_level_transforms</span><span class="p">():</span>
       <span class="n">vertex_dict</span><span class="p">[</span><span class="bp">self</span><span class="o">.</span><span class="n">_decorate</span><span class="p">(</span><span class="n">transform</span><span class="o">.</span><span class="n">unique_name</span><span class="p">)]</span> <span class="o">=</span> <span class="p">{}</span>

       <span class="k">for</span> <span class="n">pcoll_id</span> <span class="ow">in</span> <span class="n">transform</span><span class="o">.</span><span class="n">outputs</span><span class="o">.</span><span class="n">values</span><span class="p">():</span>
         <span class="n">pcoll_node</span> <span class="o">=</span> <span class="kc">None</span>
         <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_instrument</span><span class="p">:</span>
           <span class="n">cacheable</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_pipeline_instrument</span><span class="o">.</span><span class="n">cacheables</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="n">pcoll_id</span><span class="p">)</span>
           <span class="n">pcoll_node</span> <span class="o">=</span> <span class="n">cacheable</span><span class="o">.</span><span class="n">var</span> <span class="k">if</span> <span class="n">cacheable</span> <span class="k">else</span> <span class="kc">None</span>
         <span class="c1"># If no PipelineInstrument is available or the PCollection is not</span>
         <span class="c1"># watched.</span>
         <span class="k">if</span> <span class="ow">not</span> <span class="n">pcoll_node</span><span class="p">:</span>
           <span class="n">pcoll_node</span> <span class="o">=</span> <span class="s1">&#39;pcoll</span><span class="si">%s</span><span class="s1">&#39;</span> <span class="o">%</span> <span class="p">(</span><span class="nb">hash</span><span class="p">(</span><span class="n">pcoll_id</span><span class="p">)</span> <span class="o">%</span> <span class="mi">10000</span><span class="p">)</span>
           <span class="n">vertex_dict</span><span class="p">[</span><span class="n">pcoll_node</span><span class="p">]</span> <span class="o">=</span> <span class="p">{</span>
               <span class="s1">&#39;shape&#39;</span><span class="p">:</span> <span class="s1">&#39;circle&#39;</span><span class="p">,</span>
               <span class="s1">&#39;label&#39;</span><span class="p">:</span> <span class="s1">&#39;&#39;</span><span class="p">,</span>  <span class="c1"># The pcoll node has no name.</span>
           <span class="p">}</span>
         <span class="c1"># There is PipelineInstrument and the PCollection is watched with an</span>
         <span class="c1"># assigned variable.</span>
         <span class="k">else</span><span class="p">:</span>
           <span class="n">vertex_dict</span><span class="p">[</span><span class="n">pcoll_node</span><span class="p">]</span> <span class="o">=</span> <span class="p">{</span><span class="s1">&#39;shape&#39;</span><span class="p">:</span> <span class="s1">&#39;circle&#39;</span><span class="p">}</span>
         <span class="k">if</span> <span class="n">pcoll_id</span> <span class="ow">not</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_consumers</span><span class="p">:</span>
           <span class="bp">self</span><span class="o">.</span><span class="n">_edge_to_vertex_pairs</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
               <span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_decorate</span><span class="p">(</span><span class="n">transform</span><span class="o">.</span><span class="n">unique_name</span><span class="p">),</span> <span class="n">pcoll_node</span><span class="p">))</span>
           <span class="n">edge_dict</span><span class="p">[(</span><span class="bp">self</span><span class="o">.</span><span class="n">_decorate</span><span class="p">(</span><span class="n">transform</span><span class="o">.</span><span class="n">unique_name</span><span class="p">),</span> <span class="n">pcoll_node</span><span class="p">)]</span> <span class="o">=</span> <span class="p">{}</span>
         <span class="k">else</span><span class="p">:</span>
           <span class="k">for</span> <span class="n">consumer</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_consumers</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]:</span>
             <span class="n">producer_name</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_decorate</span><span class="p">(</span><span class="n">transform</span><span class="o">.</span><span class="n">unique_name</span><span class="p">)</span>
             <span class="n">consumer_name</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_decorate</span><span class="p">(</span><span class="n">transforms</span><span class="p">[</span><span class="n">consumer</span><span class="p">]</span><span class="o">.</span><span class="n">unique_name</span><span class="p">)</span>
             <span class="bp">self</span><span class="o">.</span><span class="n">_edge_to_vertex_pairs</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
                 <span class="p">(</span><span class="n">producer_name</span><span class="p">,</span> <span class="n">pcoll_node</span><span class="p">))</span>
             <span class="n">edge_dict</span><span class="p">[(</span><span class="n">producer_name</span><span class="p">,</span> <span class="n">pcoll_node</span><span class="p">)]</span> <span class="o">=</span> <span class="p">{}</span>
             <span class="bp">self</span><span class="o">.</span><span class="n">_edge_to_vertex_pairs</span><span class="p">[</span><span class="n">pcoll_id</span><span class="p">]</span><span class="o">.</span><span class="n">append</span><span class="p">(</span>
                 <span class="p">(</span><span class="n">pcoll_node</span><span class="p">,</span> <span class="n">consumer_name</span><span class="p">))</span>
             <span class="n">edge_dict</span><span class="p">[(</span><span class="n">pcoll_node</span><span class="p">,</span> <span class="n">consumer_name</span><span class="p">)]</span> <span class="o">=</span> <span class="p">{}</span>

     <span class="k">return</span> <span class="n">vertex_dict</span><span class="p">,</span> <span class="n">edge_dict</span>

   <span class="k">def</span> <span class="nf">_get_graph</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Returns pydot.Dot object for the pipeline graph.</span>

 <span class="sd">    The purpose of this method is to avoid accessing the graph while it is</span>
 <span class="sd">    updated. No one except for this method should be accessing _graph directly.</span>

 <span class="sd">    Returns:</span>
 <span class="sd">      (pydot.Dot)</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="k">with</span> <span class="bp">self</span><span class="o">.</span><span class="n">_lock</span><span class="p">:</span>
       <span class="k">return</span> <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span>

   <span class="k">def</span> <span class="nf">_construct_graph</span><span class="p">(</span>
       <span class="bp">self</span><span class="p">,</span> <span class="n">vertex_dict</span><span class="p">,</span> <span class="n">edge_dict</span><span class="p">,</span> <span class="n">default_vertex_attrs</span><span class="p">,</span> <span class="n">default_edge_attrs</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Constructs the pydot.Dot object for the pipeline graph.</span>

 <span class="sd">    Args:</span>
 <span class="sd">      vertex_dict: (Dict[str, Dict[str, str]]) maps vertex names to attributes</span>
 <span class="sd">      edge_dict: (Dict[(str, str), Dict[str, str]]) maps vertex name pairs to</span>
 <span class="sd">          attributes</span>
 <span class="sd">      default_vertex_attrs: (Dict[str, str]) a dict of attributes</span>
 <span class="sd">      default_edge_attrs: (Dict[str, str]) a dict of attributes</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="k">with</span> <span class="bp">self</span><span class="o">.</span><span class="n">_lock</span><span class="p">:</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span> <span class="o">=</span> <span class="n">pydot</span><span class="o">.</span><span class="n">Dot</span><span class="p">()</span>

       <span class="k">if</span> <span class="n">default_vertex_attrs</span><span class="p">:</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span><span class="o">.</span><span class="n">set_node_defaults</span><span class="p">(</span><span class="o">**</span><span class="n">default_vertex_attrs</span><span class="p">)</span>
       <span class="k">if</span> <span class="n">default_edge_attrs</span><span class="p">:</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span><span class="o">.</span><span class="n">set_edge_defaults</span><span class="p">(</span><span class="o">**</span><span class="n">default_edge_attrs</span><span class="p">)</span>

       <span class="bp">self</span><span class="o">.</span><span class="n">_vertex_refs</span> <span class="o">=</span> <span class="p">{}</span>  <span class="c1"># Maps vertex name to pydot.Node</span>
       <span class="bp">self</span><span class="o">.</span><span class="n">_edge_refs</span> <span class="o">=</span> <span class="p">{}</span>  <span class="c1"># Maps vertex name pairs to pydot.Edge</span>

       <span class="k">for</span> <span class="n">vertex</span><span class="p">,</span> <span class="n">vertex_attrs</span> <span class="ow">in</span> <span class="n">vertex_dict</span><span class="o">.</span><span class="n">items</span><span class="p">():</span>
         <span class="n">vertex_ref</span> <span class="o">=</span> <span class="n">pydot</span><span class="o">.</span><span class="n">Node</span><span class="p">(</span><span class="n">vertex</span><span class="p">,</span> <span class="o">**</span><span class="n">vertex_attrs</span><span class="p">)</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_vertex_refs</span><span class="p">[</span><span class="n">vertex</span><span class="p">]</span> <span class="o">=</span> <span class="n">vertex_ref</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span><span class="o">.</span><span class="n">add_node</span><span class="p">(</span><span class="n">vertex_ref</span><span class="p">)</span>

       <span class="k">for</span> <span class="n">edge</span><span class="p">,</span> <span class="n">edge_attrs</span> <span class="ow">in</span> <span class="n">edge_dict</span><span class="o">.</span><span class="n">items</span><span class="p">():</span>
         <span class="n">vertex_src</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_vertex_refs</span><span class="p">[</span><span class="n">edge</span><span class="p">[</span><span class="mi">0</span><span class="p">]]</span>
         <span class="n">vertex_dst</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">_vertex_refs</span><span class="p">[</span><span class="n">edge</span><span class="p">[</span><span class="mi">1</span><span class="p">]]</span>

         <span class="n">edge_ref</span> <span class="o">=</span> <span class="n">pydot</span><span class="o">.</span><span class="n">Edge</span><span class="p">(</span><span class="n">vertex_src</span><span class="p">,</span> <span class="n">vertex_dst</span><span class="p">,</span> <span class="o">**</span><span class="n">edge_attrs</span><span class="p">)</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_edge_refs</span><span class="p">[</span><span class="n">edge</span><span class="p">]</span> <span class="o">=</span> <span class="n">edge_ref</span>
         <span class="bp">self</span><span class="o">.</span><span class="n">_graph</span><span class="o">.</span><span class="n">add_edge</span><span class="p">(</span><span class="n">edge_ref</span><span class="p">)</span>

   <span class="k">def</span> <span class="nf">_update_graph</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">vertex_dict</span><span class="o">=</span><span class="kc">None</span><span class="p">,</span> <span class="n">edge_dict</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
 <span class="w">    </span><span class="sd">&quot;&quot;&quot;Updates the pydot.Dot object with the given attribute update</span>

 <span class="sd">    Args:</span>
 <span class="sd">      vertex_dict: (Dict[str, Dict[str, str]]) maps vertex names to attributes</span>
 <span class="sd">      edge_dict: This should be</span>
 <span class="sd">          Either (Dict[str, Dict[str, str]]) which maps edge names to attributes</span>
 <span class="sd">          Or (Dict[(str, str), Dict[str, str]]) which maps vertex pairs to edge</span>
 <span class="sd">          attributes</span>
 <span class="sd">    &quot;&quot;&quot;</span>
     <span class="k">def</span> <span class="nf">set_attrs</span><span class="p">(</span><span class="n">ref</span><span class="p">,</span> <span class="n">attrs</span><span class="p">):</span>
       <span class="k">for</span> <span class="n">attr_name</span><span class="p">,</span> <span class="n">attr_val</span> <span class="ow">in</span> <span class="n">attrs</span><span class="o">.</span><span class="n">items</span><span class="p">():</span>
         <span class="n">ref</span><span class="o">.</span><span class="n">set</span><span class="p">(</span><span class="n">attr_name</span><span class="p">,</span> <span class="n">attr_val</span><span class="p">)</span>

     <span class="k">with</span> <span class="bp">self</span><span class="o">.</span><span class="n">_lock</span><span class="p">:</span>
       <span class="k">if</span> <span class="n">vertex_dict</span><span class="p">:</span>
         <span class="k">for</span> <span class="n">vertex</span><span class="p">,</span> <span class="n">vertex_attrs</span> <span class="ow">in</span> <span class="n">vertex_dict</span><span class="o">.</span><span class="n">items</span><span class="p">():</span>
           <span class="n">set_attrs</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_vertex_refs</span><span class="p">[</span><span class="n">vertex</span><span class="p">],</span> <span class="n">vertex_attrs</span><span class="p">)</span>
       <span class="k">if</span> <span class="n">edge_dict</span><span class="p">:</span>
         <span class="k">for</span> <span class="n">edge</span><span class="p">,</span> <span class="n">edge_attrs</span> <span class="ow">in</span> <span class="n">edge_dict</span><span class="o">.</span><span class="n">items</span><span class="p">():</span>
           <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">edge</span><span class="p">,</span> <span class="nb">tuple</span><span class="p">):</span>
             <span class="n">set_attrs</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_edge_refs</span><span class="p">[</span><span class="n">edge</span><span class="p">],</span> <span class="n">edge_attrs</span><span class="p">)</span>
           <span class="k">else</span><span class="p">:</span>
             <span class="k">for</span> <span class="n">vertex_pair</span> <span class="ow">in</span> <span class="bp">self</span><span class="o">.</span><span class="n">_edge_to_vertex_pairs</span><span class="p">[</span><span class="n">edge</span><span class="p">]:</span>
               <span class="n">set_attrs</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_edge_refs</span><span class="p">[</span><span class="n">vertex_pair</span><span class="p">],</span> <span class="n">edge_attrs</span><span class="p">)</span></div>
 </pre></div>

            </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/rtfd/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">
       jQuery(function () {
           SphinxRtdTheme.Navigation.enable(true);
       });
   </script>


 </body>
 </html>