You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by cr...@apache.org on 2018/01/03 17:48:17 UTC
[17/35] incubator-airflow-site git commit: 1.9.0
http://git-wip-us.apache.org/repos/asf/incubator-airflow-site/blob/28a3eb60/_modules/dagrun_operator.html
----------------------------------------------------------------------
diff --git a/_modules/dagrun_operator.html b/_modules/dagrun_operator.html
index ec09c2c..8dd70d8 100644
--- a/_modules/dagrun_operator.html
+++ b/_modules/dagrun_operator.html
@@ -13,6 +13,8 @@
+
+
@@ -80,7 +82,10 @@
- <ul>
+
+
+
+ <ul>
<li class="toctree-l1"><a class="reference internal" href="../project.html">Project</a></li>
<li class="toctree-l1"><a class="reference internal" href="../license.html">License</a></li>
<li class="toctree-l1"><a class="reference internal" href="../start.html">Quick Start</a></li>
@@ -177,13 +182,11 @@
<span class="c1"># limitations under the License.</span>
<span class="kn">from</span> <span class="nn">datetime</span> <span class="k">import</span> <span class="n">datetime</span>
-<span class="kn">import</span> <span class="nn">logging</span>
<span class="kn">from</span> <span class="nn">airflow.models</span> <span class="k">import</span> <span class="n">BaseOperator</span><span class="p">,</span> <span class="n">DagBag</span>
<span class="kn">from</span> <span class="nn">airflow.utils.decorators</span> <span class="k">import</span> <span class="n">apply_defaults</span>
<span class="kn">from</span> <span class="nn">airflow.utils.state</span> <span class="k">import</span> <span class="n">State</span>
<span class="kn">from</span> <span class="nn">airflow</span> <span class="k">import</span> <span class="n">settings</span>
-<span class="kn">from</span> <span class="nn">airflow</span> <span class="k">import</span> <span class="n">configuration</span> <span class="k">as</span> <span class="n">conf</span>
<span class="k">class</span> <span class="nc">DagRunOrder</span><span class="p">(</span><span class="nb">object</span><span class="p">):</span>
@@ -219,12 +222,12 @@
<span class="n">trigger_dag_id</span><span class="p">,</span>
<span class="n">python_callable</span><span class="p">,</span>
<span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">):</span>
- <span class="nb">super</span><span class="p">(</span><span class="n">TriggerDagRunOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="n">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
+ <span class="nb">super</span><span class="p">(</span><span class="n">TriggerDagRunOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">python_callable</span> <span class="o">=</span> <span class="n">python_callable</span>
<span class="bp">self</span><span class="o">.</span><span class="n">trigger_dag_id</span> <span class="o">=</span> <span class="n">trigger_dag_id</span>
<span class="k">def</span> <span class="nf">execute</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">context</span><span class="p">):</span>
- <span class="n">dro</span> <span class="o">=</span> <span class="n">DagRunOrder</span><span class="p">(</span><span class="n">run_id</span><span class="o">=</span><span class="s1">'trig__'</span> <span class="o">+</span> <span class="n">datetime</span><span class="o">.</span><span class="n">now</span><span class="p">()</span><span class="o">.</span><span class="n">isoformat</span><span class="p">())</span>
+ <span class="n">dro</span> <span class="o">=</span> <span class="n">DagRunOrder</span><span class="p">(</span><span class="n">run_id</span><span class="o">=</span><span class="s1">'trig__'</span> <span class="o">+</span> <span class="n">datetime</span><span class="o">.</span><span class="n">utcnow</span><span class="p">()</span><span class="o">.</span><span class="n">isoformat</span><span class="p">())</span>
<span class="n">dro</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">python_callable</span><span class="p">(</span><span class="n">context</span><span class="p">,</span> <span class="n">dro</span><span class="p">)</span>
<span class="k">if</span> <span class="n">dro</span><span class="p">:</span>
<span class="n">session</span> <span class="o">=</span> <span class="n">settings</span><span class="o">.</span><span class="n">Session</span><span class="p">()</span>
@@ -235,12 +238,12 @@
<span class="n">state</span><span class="o">=</span><span class="n">State</span><span class="o">.</span><span class="n">RUNNING</span><span class="p">,</span>
<span class="n">conf</span><span class="o">=</span><span class="n">dro</span><span class="o">.</span><span class="n">payload</span><span class="p">,</span>
<span class="n">external_trigger</span><span class="o">=</span><span class="kc">True</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="s2">"Creating DagRun </span><span class="si">{}</span><span class="s2">"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">dr</span><span class="p">))</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"Creating DagRun </span><span class="si">%s</span><span class="s2">"</span><span class="p">,</span> <span class="n">dr</span><span class="p">)</span>
<span class="n">session</span><span class="o">.</span><span class="n">add</span><span class="p">(</span><span class="n">dr</span><span class="p">)</span>
<span class="n">session</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
<span class="n">session</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
<span class="k">else</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="s2">"Criteria not met, moving on"</span><span class="p">)</span></div>
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"Criteria not met, moving on"</span><span class="p">)</span></div>
</pre></div>
</div>
http://git-wip-us.apache.org/repos/asf/incubator-airflow-site/blob/28a3eb60/_modules/dbapi_hook.html
----------------------------------------------------------------------
diff --git a/_modules/dbapi_hook.html b/_modules/dbapi_hook.html
index ae37fd3..34111f9 100644
--- a/_modules/dbapi_hook.html
+++ b/_modules/dbapi_hook.html
@@ -13,6 +13,8 @@
+
+
@@ -80,7 +82,10 @@
- <ul>
+
+
+
+ <ul>
<li class="toctree-l1"><a class="reference internal" href="../project.html">Project</a></li>
<li class="toctree-l1"><a class="reference internal" href="../license.html">License</a></li>
<li class="toctree-l1"><a class="reference internal" href="../start.html">Quick Start</a></li>
@@ -179,8 +184,7 @@
<span class="kn">from</span> <span class="nn">builtins</span> <span class="k">import</span> <span class="nb">str</span>
<span class="kn">from</span> <span class="nn">past.builtins</span> <span class="k">import</span> <span class="n">basestring</span>
<span class="kn">from</span> <span class="nn">datetime</span> <span class="k">import</span> <span class="n">datetime</span>
-<span class="kn">import</span> <span class="nn">numpy</span>
-<span class="kn">import</span> <span class="nn">logging</span>
+<span class="kn">from</span> <span class="nn">contextlib</span> <span class="k">import</span> <span class="n">closing</span>
<span class="kn">import</span> <span class="nn">sys</span>
<span class="kn">from</span> <span class="nn">sqlalchemy</span> <span class="k">import</span> <span class="n">create_engine</span>
@@ -251,10 +255,9 @@
<span class="k">if</span> <span class="n">sys</span><span class="o">.</span><span class="n">version_info</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o"><</span> <span class="mi">3</span><span class="p">:</span>
<span class="n">sql</span> <span class="o">=</span> <span class="n">sql</span><span class="o">.</span><span class="n">encode</span><span class="p">(</span><span class="s1">'utf-8'</span><span class="p">)</span>
<span class="kn">import</span> <span class="nn">pandas.io.sql</span> <span class="k">as</span> <span class="nn">psql</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">()</span>
- <span class="n">df</span> <span class="o">=</span> <span class="n">psql</span><span class="o">.</span><span class="n">read_sql</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">con</span><span class="o">=</span><span class="n">conn</span><span class="p">,</span> <span class="n">params</span><span class="o">=</span><span class="n">parameters</span><span class="p">)</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="k">return</span> <span class="n">df</span></div>
+
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">())</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
+ <span class="k">return</span> <span class="n">psql</span><span class="o">.</span><span class="n">read_sql</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">con</span><span class="o">=</span><span class="n">conn</span><span class="p">,</span> <span class="n">params</span><span class="o">=</span><span class="n">parameters</span><span class="p">)</span></div>
<div class="viewcode-block" id="DbApiHook.get_records"><a class="viewcode-back" href="../code.html#airflow.hooks.DbApiHook.get_records">[docs]</a> <span class="k">def</span> <span class="nf">get_records</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
<span class="sd">"""</span>
@@ -268,16 +271,14 @@
<span class="sd"> """</span>
<span class="k">if</span> <span class="n">sys</span><span class="o">.</span><span class="n">version_info</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o"><</span> <span class="mi">3</span><span class="p">:</span>
<span class="n">sql</span> <span class="o">=</span> <span class="n">sql</span><span class="o">.</span><span class="n">encode</span><span class="p">(</span><span class="s1">'utf-8'</span><span class="p">)</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">()</span>
- <span class="n">cur</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_cursor</span><span class="p">()</span>
- <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
- <span class="k">else</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">)</span>
- <span class="n">rows</span> <span class="o">=</span> <span class="n">cur</span><span class="o">.</span><span class="n">fetchall</span><span class="p">()</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="k">return</span> <span class="n">rows</span></div>
+
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">())</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">())</span> <span class="k">as</span> <span class="n">cur</span><span class="p">:</span>
+ <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
+ <span class="k">else</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">)</span>
+ <span class="k">return</span> <span class="n">cur</span><span class="o">.</span><span class="n">fetchall</span><span class="p">()</span></div>
<div class="viewcode-block" id="DbApiHook.get_first"><a class="viewcode-back" href="../code.html#airflow.hooks.DbApiHook.get_first">[docs]</a> <span class="k">def</span> <span class="nf">get_first</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
<span class="sd">"""</span>
@@ -291,16 +292,14 @@
<span class="sd"> """</span>
<span class="k">if</span> <span class="n">sys</span><span class="o">.</span><span class="n">version_info</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o"><</span> <span class="mi">3</span><span class="p">:</span>
<span class="n">sql</span> <span class="o">=</span> <span class="n">sql</span><span class="o">.</span><span class="n">encode</span><span class="p">(</span><span class="s1">'utf-8'</span><span class="p">)</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">()</span>
- <span class="n">cur</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">()</span>
- <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
- <span class="k">else</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">)</span>
- <span class="n">rows</span> <span class="o">=</span> <span class="n">cur</span><span class="o">.</span><span class="n">fetchone</span><span class="p">()</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="k">return</span> <span class="n">rows</span></div>
+
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">())</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">())</span> <span class="k">as</span> <span class="n">cur</span><span class="p">:</span>
+ <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
+ <span class="k">else</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">)</span>
+ <span class="k">return</span> <span class="n">cur</span><span class="o">.</span><span class="n">fetchone</span><span class="p">()</span></div>
<div class="viewcode-block" id="DbApiHook.run"><a class="viewcode-back" href="../code.html#airflow.hooks.DbApiHook.run">[docs]</a> <span class="k">def</span> <span class="nf">run</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">sql</span><span class="p">,</span> <span class="n">autocommit</span><span class="o">=</span><span class="kc">False</span><span class="p">,</span> <span class="n">parameters</span><span class="o">=</span><span class="kc">None</span><span class="p">):</span>
<span class="sd">"""</span>
@@ -317,25 +316,24 @@
<span class="sd"> :param parameters: The parameters to render the SQL query with.</span>
<span class="sd"> :type parameters: mapping or iterable</span>
<span class="sd"> """</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">()</span>
<span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">basestring</span><span class="p">):</span>
<span class="n">sql</span> <span class="o">=</span> <span class="p">[</span><span class="n">sql</span><span class="p">]</span>
- <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">supports_autocommit</span><span class="p">:</span>
- <span class="bp">self</span><span class="o">.</span><span class="n">set_autocommit</span><span class="p">(</span><span class="n">conn</span><span class="p">,</span> <span class="n">autocommit</span><span class="p">)</span>
-
- <span class="n">cur</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">()</span>
- <span class="k">for</span> <span class="n">s</span> <span class="ow">in</span> <span class="n">sql</span><span class="p">:</span>
- <span class="k">if</span> <span class="n">sys</span><span class="o">.</span><span class="n">version_info</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o"><</span> <span class="mi">3</span><span class="p">:</span>
- <span class="n">s</span> <span class="o">=</span> <span class="n">s</span><span class="o">.</span><span class="n">encode</span><span class="p">(</span><span class="s1">'utf-8'</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">s</span><span class="p">)</span>
- <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">s</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
- <span class="k">else</span><span class="p">:</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">s</span><span class="p">)</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">close</span><span class="p">()</span></div>
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">())</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
+ <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">supports_autocommit</span><span class="p">:</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">set_autocommit</span><span class="p">(</span><span class="n">conn</span><span class="p">,</span> <span class="n">autocommit</span><span class="p">)</span>
+
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">())</span> <span class="k">as</span> <span class="n">cur</span><span class="p">:</span>
+ <span class="k">for</span> <span class="n">s</span> <span class="ow">in</span> <span class="n">sql</span><span class="p">:</span>
+ <span class="k">if</span> <span class="n">sys</span><span class="o">.</span><span class="n">version_info</span><span class="p">[</span><span class="mi">0</span><span class="p">]</span> <span class="o"><</span> <span class="mi">3</span><span class="p">:</span>
+ <span class="n">s</span> <span class="o">=</span> <span class="n">s</span><span class="o">.</span><span class="n">encode</span><span class="p">(</span><span class="s1">'utf-8'</span><span class="p">)</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="n">s</span><span class="p">)</span>
+ <span class="k">if</span> <span class="n">parameters</span> <span class="ow">is</span> <span class="ow">not</span> <span class="kc">None</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">s</span><span class="p">,</span> <span class="n">parameters</span><span class="p">)</span>
+ <span class="k">else</span><span class="p">:</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">s</span><span class="p">)</span>
+
+ <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span></div>
<span class="k">def</span> <span class="nf">set_autocommit</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">conn</span><span class="p">,</span> <span class="n">autocommit</span><span class="p">):</span>
<span class="n">conn</span><span class="o">.</span><span class="n">autocommit</span> <span class="o">=</span> <span class="n">autocommit</span>
@@ -349,7 +347,7 @@
<div class="viewcode-block" id="DbApiHook.insert_rows"><a class="viewcode-back" href="../code.html#airflow.hooks.DbApiHook.insert_rows">[docs]</a> <span class="k">def</span> <span class="nf">insert_rows</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">table</span><span class="p">,</span> <span class="n">rows</span><span class="p">,</span> <span class="n">target_fields</span><span class="o">=</span><span class="kc">None</span><span class="p">,</span> <span class="n">commit_every</span><span class="o">=</span><span class="mi">1000</span><span class="p">):</span>
<span class="sd">"""</span>
<span class="sd"> A generic way to insert a set of tuples into a table,</span>
-<span class="sd"> the whole set of inserts is treated as one transaction</span>
+<span class="sd"> a new transaction is created every commit_every rows</span>
<span class="sd"> :param table: Name of the target table</span>
<span class="sd"> :type table: str</span>
@@ -366,31 +364,33 @@
<span class="n">target_fields</span> <span class="o">=</span> <span class="s2">"(</span><span class="si">{}</span><span class="s2">)"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">target_fields</span><span class="p">)</span>
<span class="k">else</span><span class="p">:</span>
<span class="n">target_fields</span> <span class="o">=</span> <span class="s1">''</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">()</span>
- <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">supports_autocommit</span><span class="p">:</span>
- <span class="bp">self</span><span class="o">.</span><span class="n">set_autocommit</span><span class="p">(</span><span class="n">conn</span><span class="p">,</span> <span class="kc">False</span><span class="p">)</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
- <span class="n">cur</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">()</span>
- <span class="n">i</span> <span class="o">=</span> <span class="mi">0</span>
- <span class="k">for</span> <span class="n">row</span> <span class="ow">in</span> <span class="n">rows</span><span class="p">:</span>
- <span class="n">i</span> <span class="o">+=</span> <span class="mi">1</span>
- <span class="n">l</span> <span class="o">=</span> <span class="p">[]</span>
- <span class="k">for</span> <span class="n">cell</span> <span class="ow">in</span> <span class="n">row</span><span class="p">:</span>
- <span class="n">l</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_serialize_cell</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">conn</span><span class="p">))</span>
- <span class="n">values</span> <span class="o">=</span> <span class="nb">tuple</span><span class="p">(</span><span class="n">l</span><span class="p">)</span>
- <span class="n">sql</span> <span class="o">=</span> <span class="s2">"INSERT INTO </span><span class="si">{0}</span><span class="s2"> </span><span class="si">{1}</span><span class="s2"> VALUES (</span><span class="si">{2}</span><span class="s2">);"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span>
- <span class="n">table</span><span class="p">,</span>
- <span class="n">target_fields</span><span class="p">,</span>
- <span class="s2">","</span><span class="o">.</span><span class="n">join</span><span class="p">(</span><span class="n">values</span><span class="p">))</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">)</span>
- <span class="k">if</span> <span class="n">commit_every</span> <span class="ow">and</span> <span class="n">i</span> <span class="o">%</span> <span class="n">commit_every</span> <span class="o">==</span> <span class="mi">0</span><span class="p">:</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">commit</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="s2">"Loaded </span><span class="si">{i}</span><span class="s2"> into </span><span class="si">{table}</span><span class="s2"> rows so far"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">()))</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
- <span class="n">cur</span><span class="o">.</span><span class="n">close</span><span class="p">()</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">close</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="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">get_conn</span><span class="p">())</span> <span class="k">as</span> <span class="n">conn</span><span class="p">:</span>
+ <span class="k">if</span> <span class="bp">self</span><span class="o">.</span><span class="n">supports_autocommit</span><span class="p">:</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">set_autocommit</span><span class="p">(</span><span class="n">conn</span><span class="p">,</span> <span class="kc">False</span><span class="p">)</span>
+
+ <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
+
+ <span class="k">with</span> <span class="n">closing</span><span class="p">(</span><span class="n">conn</span><span class="o">.</span><span class="n">cursor</span><span class="p">())</span> <span class="k">as</span> <span class="n">cur</span><span class="p">:</span>
+ <span class="k">for</span> <span class="n">i</span><span class="p">,</span> <span class="n">row</span> <span class="ow">in</span> <span class="nb">enumerate</span><span class="p">(</span><span class="n">rows</span><span class="p">,</span> <span class="mi">1</span><span class="p">):</span>
+ <span class="n">l</span> <span class="o">=</span> <span class="p">[]</span>
+ <span class="k">for</span> <span class="n">cell</span> <span class="ow">in</span> <span class="n">row</span><span class="p">:</span>
+ <span class="n">l</span><span class="o">.</span><span class="n">append</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">_serialize_cell</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">conn</span><span class="p">))</span>
+ <span class="n">values</span> <span class="o">=</span> <span class="nb">tuple</span><span class="p">(</span><span class="n">l</span><span class="p">)</span>
+ <span class="n">placeholders</span> <span class="o">=</span> <span class="p">[</span><span class="s2">"</span><span class="si">%s</span><span class="s2">"</span><span class="p">,]</span><span class="o">*</span><span class="nb">len</span><span class="p">(</span><span class="n">values</span><span class="p">)</span>
+ <span class="n">sql</span> <span class="o">=</span> <span class="s2">"INSERT INTO </span><span class="si">{0}</span><span class="s2"> </span><span class="si">{1}</span><span class="s2"> VALUES (</span><span class="si">{2}</span><span class="s2">);"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span>
+ <span class="n">table</span><span class="p">,</span>
+ <span class="n">target_fields</span><span class="p">,</span>
+ <span class="s2">","</span><span class="o">.</span><span class="n">join</span><span class="p">(</span><span class="n">placeholders</span><span class="p">))</span>
+ <span class="n">cur</span><span class="o">.</span><span class="n">execute</span><span class="p">(</span><span class="n">sql</span><span class="p">,</span> <span class="n">values</span><span class="p">)</span>
+ <span class="k">if</span> <span class="n">commit_every</span> <span class="ow">and</span> <span class="n">i</span> <span class="o">%</span> <span class="n">commit_every</span> <span class="o">==</span> <span class="mi">0</span><span class="p">:</span>
+ <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span>
+ <span class="s2">"Loaded </span><span class="si">{i}</span><span class="s2"> into </span><span class="si">{table}</span><span class="s2"> rows so far"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">())</span>
+ <span class="p">)</span>
+
+ <span class="n">conn</span><span class="o">.</span><span class="n">commit</span><span class="p">()</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span>
<span class="s2">"Done loading. Loaded a total of </span><span class="si">{i}</span><span class="s2"> rows"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">()))</span></div>
<span class="nd">@staticmethod</span>
@@ -406,16 +406,11 @@
<span class="sd"> :rtype: str</span>
<span class="sd"> """</span>
- <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">basestring</span><span class="p">):</span>
- <span class="k">return</span> <span class="s2">"'"</span> <span class="o">+</span> <span class="nb">str</span><span class="p">(</span><span class="n">cell</span><span class="p">)</span><span class="o">.</span><span class="n">replace</span><span class="p">(</span><span class="s2">"'"</span><span class="p">,</span> <span class="s2">"''"</span><span class="p">)</span> <span class="o">+</span> <span class="s2">"'"</span>
- <span class="k">elif</span> <span class="n">cell</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span>
- <span class="k">return</span> <span class="s1">'NULL'</span>
- <span class="k">elif</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">numpy</span><span class="o">.</span><span class="n">datetime64</span><span class="p">):</span>
- <span class="k">return</span> <span class="s2">"'"</span> <span class="o">+</span> <span class="nb">str</span><span class="p">(</span><span class="n">cell</span><span class="p">)</span> <span class="o">+</span> <span class="s2">"'"</span>
- <span class="k">elif</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">datetime</span><span class="p">):</span>
- <span class="k">return</span> <span class="s2">"'"</span> <span class="o">+</span> <span class="n">cell</span><span class="o">.</span><span class="n">isoformat</span><span class="p">()</span> <span class="o">+</span> <span class="s2">"'"</span>
- <span class="k">else</span><span class="p">:</span>
- <span class="k">return</span> <span class="nb">str</span><span class="p">(</span><span class="n">cell</span><span class="p">)</span>
+ <span class="k">if</span> <span class="n">cell</span> <span class="ow">is</span> <span class="kc">None</span><span class="p">:</span>
+ <span class="k">return</span> <span class="kc">None</span>
+ <span class="k">if</span> <span class="nb">isinstance</span><span class="p">(</span><span class="n">cell</span><span class="p">,</span> <span class="n">datetime</span><span class="p">):</span>
+ <span class="k">return</span> <span class="n">cell</span><span class="o">.</span><span class="n">isoformat</span><span class="p">()</span>
+ <span class="k">return</span> <span class="nb">str</span><span class="p">(</span><span class="n">cell</span><span class="p">)</span>
<div class="viewcode-block" id="DbApiHook.bulk_dump"><a class="viewcode-back" href="../code.html#airflow.hooks.DbApiHook.bulk_dump">[docs]</a> <span class="k">def</span> <span class="nf">bulk_dump</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">table</span><span class="p">,</span> <span class="n">tmp_file</span><span class="p">):</span>
<span class="sd">"""</span>
http://git-wip-us.apache.org/repos/asf/incubator-airflow-site/blob/28a3eb60/_modules/druid_hook.html
----------------------------------------------------------------------
diff --git a/_modules/druid_hook.html b/_modules/druid_hook.html
index 20ffb1d..5bb5145 100644
--- a/_modules/druid_hook.html
+++ b/_modules/druid_hook.html
@@ -13,6 +13,8 @@
+
+
@@ -30,6 +32,9 @@
+ <link rel="index" title="Index"
+ href="../genindex.html"/>
+ <link rel="search" title="Search" href="../search.html"/>
<link rel="top" title="Airflow Documentation" href="../index.html"/>
<link rel="up" title="Module code" href="index.html"/>
@@ -40,6 +45,7 @@
<body class="wy-body-for-nav" role="document">
+
<div class="wy-grid-for-nav">
@@ -76,7 +82,10 @@
- <ul>
+
+
+
+ <ul>
<li class="toctree-l1"><a class="reference internal" href="../project.html">Project</a></li>
<li class="toctree-l1"><a class="reference internal" href="../license.html">License</a></li>
<li class="toctree-l1"><a class="reference internal" href="../start.html">Quick Start</a></li>
@@ -90,6 +99,8 @@
<li class="toctree-l1"><a class="reference internal" href="../scheduler.html">Scheduling & Triggers</a></li>
<li class="toctree-l1"><a class="reference internal" href="../plugins.html">Plugins</a></li>
<li class="toctree-l1"><a class="reference internal" href="../security.html">Security</a></li>
+<li class="toctree-l1"><a class="reference internal" href="../api.html">Experimental Rest API</a></li>
+<li class="toctree-l1"><a class="reference internal" href="../integration.html">Integration</a></li>
<li class="toctree-l1"><a class="reference internal" href="../faq.html">FAQ</a></li>
<li class="toctree-l1"><a class="reference internal" href="../code.html">API Reference</a></li>
</ul>
@@ -104,8 +115,10 @@
<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">Airflow</a>
+
+ <i data-toggle="wy-nav-top" class="fa fa-bars"></i>
+ <a href="../index.html">Airflow</a>
+
</nav>
@@ -118,19 +131,36 @@
+
+
+
+
+
+
+
+
+
+
<div role="navigation" aria-label="breadcrumbs navigation">
+
<ul class="wy-breadcrumbs">
- <li><a href="../index.html">Docs</a> »</li>
-
+
+ <li><a href="../index.html">Docs</a> »</li>
+
<li><a href="index.html">Module code</a> »</li>
-
- <li>druid_hook</li>
+
+ <li>druid_hook</li>
+
+
<li class="wy-breadcrumbs-aside">
-
+
</li>
+
</ul>
+
+
<hr/>
</div>
<div role="main" class="document" itemscope="itemscope" itemtype="http://schema.org/Article">
@@ -151,180 +181,90 @@
<span class="c1"># See the License for the specific language governing permissions and</span>
<span class="c1"># limitations under the License.</span>
-<span class="kn">from</span> <span class="nn">__future__</span> <span class="kn">import</span> <span class="n">print_function</span>
-<span class="kn">import</span> <span class="nn">logging</span>
-<span class="kn">import</span> <span class="nn">json</span>
-<span class="kn">import</span> <span class="nn">time</span>
+<span class="kn">from</span> <span class="nn">__future__</span> <span class="k">import</span> <span class="n">print_function</span>
-<span class="kn">from</span> <span class="nn">pydruid.client</span> <span class="kn">import</span> <span class="n">PyDruid</span>
<span class="kn">import</span> <span class="nn">requests</span>
+<span class="kn">import</span> <span class="nn">time</span>
-<span class="kn">from</span> <span class="nn">airflow.hooks.base_hook</span> <span class="kn">import</span> <span class="n">BaseHook</span>
-<span class="kn">from</span> <span class="nn">airflow.exceptions</span> <span class="kn">import</span> <span class="n">AirflowException</span>
-
-<span class="n">LOAD_CHECK_INTERVAL</span> <span class="o">=</span> <span class="mi">5</span>
-<span class="n">DEFAULT_TARGET_PARTITION_SIZE</span> <span class="o">=</span> <span class="mi">5000000</span>
-
-
-<span class="k">class</span> <span class="nc">AirflowDruidLoadException</span><span class="p">(</span><span class="n">AirflowException</span><span class="p">):</span>
- <span class="k">pass</span>
+<span class="kn">from</span> <span class="nn">airflow.exceptions</span> <span class="k">import</span> <span class="n">AirflowException</span>
+<span class="kn">from</span> <span class="nn">airflow.hooks.base_hook</span> <span class="k">import</span> <span class="n">BaseHook</span>
<div class="viewcode-block" id="DruidHook"><a class="viewcode-back" href="../code.html#airflow.hooks.DruidHook">[docs]</a><span class="k">class</span> <span class="nc">DruidHook</span><span class="p">(</span><span class="n">BaseHook</span><span class="p">):</span>
- <span class="sd">'''</span>
-<span class="sd"> Interact with druid.</span>
-<span class="sd"> '''</span>
-
+ <span class="sd">"""</span>
+<span class="sd"> Connection to Druid</span>
+
+<span class="sd"> :param druid_ingest_conn_id: The connection id to the Druid overlord machine which accepts index jobs</span>
+<span class="sd"> :type druid_ingest_conn_id: string</span>
+<span class="sd"> :param timeout: The interval between polling the Druid job for the status of the ingestion job</span>
+<span class="sd"> :type timeout: int</span>
+<span class="sd"> :param max_ingestion_time: The maximum ingestion time before assuming the job failed</span>
+<span class="sd"> :type max_ingestion_time: int</span>
+<span class="sd"> """</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">druid_query_conn_id</span><span class="o">=</span><span class="s1">'druid_query_default'</span><span class="p">,</span>
- <span class="n">druid_ingest_conn_id</span><span class="o">=</span><span class="s1">'druid_ingest_default'</span><span class="p">):</span>
- <span class="bp">self</span><span class="o">.</span><span class="n">druid_query_conn_id</span> <span class="o">=</span> <span class="n">druid_query_conn_id</span>
+ <span class="n">druid_ingest_conn_id</span><span class="o">=</span><span class="s1">'druid_ingest_default'</span><span class="p">,</span>
+ <span class="n">timeout</span><span class="o">=</span><span class="mi">1</span><span class="p">,</span>
+ <span class="n">max_ingestion_time</span><span class="o">=</span><span class="mi">18000</span><span class="p">):</span>
+
<span class="bp">self</span><span class="o">.</span><span class="n">druid_ingest_conn_id</span> <span class="o">=</span> <span class="n">druid_ingest_conn_id</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">timeout</span> <span class="o">=</span> <span class="n">timeout</span>
+ <span class="bp">self</span><span class="o">.</span><span class="n">max_ingestion_time</span> <span class="o">=</span> <span class="n">max_ingestion_time</span>
<span class="bp">self</span><span class="o">.</span><span class="n">header</span> <span class="o">=</span> <span class="p">{</span><span class="s1">'content-type'</span><span class="p">:</span> <span class="s1">'application/json'</span><span class="p">}</span>
-<div class="viewcode-block" id="DruidHook.get_conn"><a class="viewcode-back" href="../code.html#airflow.hooks.DruidHook.get_conn">[docs]</a> <span class="k">def</span> <span class="nf">get_conn</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
- <span class="sd">"""</span>
-<span class="sd"> Returns a druid connection object for query</span>
-<span class="sd"> """</span>
- <span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_connection</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">druid_query_conn_id</span><span class="p">)</span>
- <span class="k">return</span> <span class="n">PyDruid</span><span class="p">(</span>
- <span class="s2">"http://{conn.host}:{conn.port}"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">()),</span>
- <span class="n">conn</span><span class="o">.</span><span class="n">extra_dejson</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'endpoint'</span><span class="p">,</span> <span class="s1">''</span><span class="p">))</span></div>
-
- <span class="nd">@property</span>
- <span class="k">def</span> <span class="nf">ingest_post_url</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
+ <span class="k">def</span> <span class="nf">get_conn_url</span><span class="p">(</span><span class="bp">self</span><span class="p">):</span>
<span class="n">conn</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_connection</span><span class="p">(</span><span class="bp">self</span><span class="o">.</span><span class="n">druid_ingest_conn_id</span><span class="p">)</span>
<span class="n">host</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">host</span>
<span class="n">port</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">port</span>
+ <span class="n">schema</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">extra_dejson</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'schema'</span><span class="p">,</span> <span class="s1">'http'</span><span class="p">)</span>
<span class="n">endpoint</span> <span class="o">=</span> <span class="n">conn</span><span class="o">.</span><span class="n">extra_dejson</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s1">'endpoint'</span><span class="p">,</span> <span class="s1">''</span><span class="p">)</span>
- <span class="k">return</span> <span class="s2">"http://{host}:{port}/{endpoint}"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">())</span>
-
- <span class="k">def</span> <span class="nf">get_ingest_status_url</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">task_id</span><span class="p">):</span>
- <span class="n">post_url</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">ingest_post_url</span>
- <span class="k">return</span> <span class="s2">"{post_url}/{task_id}/status"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">())</span>
-
-<div class="viewcode-block" id="DruidHook.construct_ingest_query"><a class="viewcode-back" href="../code.html#airflow.hooks.DruidHook.construct_ingest_query">[docs]</a> <span class="k">def</span> <span class="nf">construct_ingest_query</span><span class="p">(</span>
- <span class="bp">self</span><span class="p">,</span> <span class="n">datasource</span><span class="p">,</span> <span class="n">static_path</span><span class="p">,</span> <span class="n">ts_dim</span><span class="p">,</span> <span class="n">columns</span><span class="p">,</span> <span class="n">metric_spec</span><span class="p">,</span>
- <span class="n">intervals</span><span class="p">,</span> <span class="n">num_shards</span><span class="p">,</span> <span class="n">target_partition_size</span><span class="p">,</span> <span class="n">hadoop_dependency_coordinates</span><span class="o">=</span><span class="bp">None</span><span class="p">):</span>
- <span class="sd">"""</span>
-<span class="sd"> Builds an ingest query for an HDFS TSV load.</span>
-
-<span class="sd"> :param datasource: target datasource in druid</span>
-<span class="sd"> :param columns: list of all columns in the TSV, in the right order</span>
-<span class="sd"> """</span>
-
- <span class="c1"># backward compatibilty for num_shards, but target_partition_size is the default setting</span>
- <span class="c1"># and overwrites the num_shards</span>
- <span class="k">if</span> <span class="n">target_partition_size</span> <span class="o">==</span> <span class="o">-</span><span class="mi">1</span><span class="p">:</span>
- <span class="k">if</span> <span class="n">num_shards</span> <span class="o">==</span> <span class="o">-</span><span class="mi">1</span><span class="p">:</span>
- <span class="n">target_partition_size</span> <span class="o">=</span> <span class="n">DEFAULT_TARGET_PARTITION_SIZE</span>
- <span class="k">else</span><span class="p">:</span>
- <span class="n">num_shards</span> <span class="o">=</span> <span class="o">-</span><span class="mi">1</span>
-
- <span class="n">metric_names</span> <span class="o">=</span> <span class="p">[</span>
- <span class="n">m</span><span class="p">[</span><span class="s1">'fieldName'</span><span class="p">]</span> <span class="k">for</span> <span class="n">m</span> <span class="ow">in</span> <span class="n">metric_spec</span> <span class="k">if</span> <span class="n">m</span><span class="p">[</span><span class="s1">'type'</span><span class="p">]</span> <span class="o">!=</span> <span class="s1">'count'</span><span class="p">]</span>
- <span class="n">dimensions</span> <span class="o">=</span> <span class="p">[</span><span class="n">c</span> <span class="k">for</span> <span class="n">c</span> <span class="ow">in</span> <span class="n">columns</span> <span class="k">if</span> <span class="n">c</span> <span class="ow">not</span> <span class="ow">in</span> <span class="n">metric_names</span> <span class="ow">and</span> <span class="n">c</span> <span class="o">!=</span> <span class="n">ts_dim</span><span class="p">]</span>
- <span class="n">ingest_query_dict</span> <span class="o">=</span> <span class="p">{</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"index_hadoop"</span><span class="p">,</span>
- <span class="s2">"spec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"dataSchema"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"metricsSpec"</span><span class="p">:</span> <span class="n">metric_spec</span><span class="p">,</span>
- <span class="s2">"granularitySpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"queryGranularity"</span><span class="p">:</span> <span class="s2">"NONE"</span><span class="p">,</span>
- <span class="s2">"intervals"</span><span class="p">:</span> <span class="n">intervals</span><span class="p">,</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"uniform"</span><span class="p">,</span>
- <span class="s2">"segmentGranularity"</span><span class="p">:</span> <span class="s2">"DAY"</span><span class="p">,</span>
- <span class="p">},</span>
- <span class="s2">"parser"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"string"</span><span class="p">,</span>
- <span class="s2">"parseSpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"columns"</span><span class="p">:</span> <span class="n">columns</span><span class="p">,</span>
- <span class="s2">"dimensionsSpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"dimensionExclusions"</span><span class="p">:</span> <span class="p">[],</span>
- <span class="s2">"dimensions"</span><span class="p">:</span> <span class="n">dimensions</span><span class="p">,</span> <span class="c1"># list of names</span>
- <span class="s2">"spatialDimensions"</span><span class="p">:</span> <span class="p">[]</span>
- <span class="p">},</span>
- <span class="s2">"timestampSpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"column"</span><span class="p">:</span> <span class="n">ts_dim</span><span class="p">,</span>
- <span class="s2">"format"</span><span class="p">:</span> <span class="s2">"auto"</span>
- <span class="p">},</span>
- <span class="s2">"format"</span><span class="p">:</span> <span class="s2">"tsv"</span>
- <span class="p">}</span>
- <span class="p">},</span>
- <span class="s2">"dataSource"</span><span class="p">:</span> <span class="n">datasource</span>
- <span class="p">},</span>
- <span class="s2">"tuningConfig"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"hadoop"</span><span class="p">,</span>
- <span class="s2">"jobProperties"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"mapreduce.job.user.classpath.first"</span><span class="p">:</span> <span class="s2">"false"</span><span class="p">,</span>
- <span class="s2">"mapreduce.map.output.compress"</span><span class="p">:</span> <span class="s2">"false"</span><span class="p">,</span>
- <span class="s2">"mapreduce.output.fileoutputformat.compress"</span><span class="p">:</span> <span class="s2">"false"</span><span class="p">,</span>
- <span class="p">},</span>
- <span class="s2">"partitionsSpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"hashed"</span><span class="p">,</span>
- <span class="s2">"targetPartitionSize"</span><span class="p">:</span> <span class="n">target_partition_size</span><span class="p">,</span>
- <span class="s2">"numShards"</span><span class="p">:</span> <span class="n">num_shards</span><span class="p">,</span>
- <span class="p">},</span>
- <span class="p">},</span>
- <span class="s2">"ioConfig"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"inputSpec"</span><span class="p">:</span> <span class="p">{</span>
- <span class="s2">"paths"</span><span class="p">:</span> <span class="n">static_path</span><span class="p">,</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"static"</span>
- <span class="p">},</span>
- <span class="s2">"type"</span><span class="p">:</span> <span class="s2">"hadoop"</span>
- <span class="p">}</span>
- <span class="p">}</span>
- <span class="p">}</span>
- <span class="k">if</span> <span class="n">hadoop_dependency_coordinates</span><span class="p">:</span>
- <span class="n">ingest_query_dict</span><span class="p">[</span>
- <span class="s1">'hadoopDependencyCoordinates'</span><span class="p">]</span> <span class="o">=</span> <span class="n">hadoop_dependency_coordinates</span>
-
- <span class="k">return</span> <span class="n">json</span><span class="o">.</span><span class="n">dumps</span><span class="p">(</span><span class="n">ingest_query_dict</span><span class="p">,</span> <span class="n">indent</span><span class="o">=</span><span class="mi">4</span><span class="p">)</span></div>
-
- <span class="k">def</span> <span class="nf">send_ingest_query</span><span class="p">(</span>
- <span class="bp">self</span><span class="p">,</span> <span class="n">datasource</span><span class="p">,</span> <span class="n">static_path</span><span class="p">,</span> <span class="n">ts_dim</span><span class="p">,</span> <span class="n">columns</span><span class="p">,</span> <span class="n">metric_spec</span><span class="p">,</span>
- <span class="n">intervals</span><span class="p">,</span> <span class="n">num_shards</span><span class="p">,</span> <span class="n">target_partition_size</span><span class="p">,</span> <span class="n">hadoop_dependency_coordinates</span><span class="o">=</span><span class="bp">None</span><span class="p">):</span>
- <span class="n">query</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">construct_ingest_query</span><span class="p">(</span>
- <span class="n">datasource</span><span class="p">,</span> <span class="n">static_path</span><span class="p">,</span> <span class="n">ts_dim</span><span class="p">,</span> <span class="n">columns</span><span class="p">,</span>
- <span class="n">metric_spec</span><span class="p">,</span> <span class="n">intervals</span><span class="p">,</span> <span class="n">num_shards</span><span class="p">,</span> <span class="n">target_partition_size</span><span class="p">,</span> <span class="n">hadoop_dependency_coordinates</span><span class="p">)</span>
- <span class="n">r</span> <span class="o">=</span> <span class="n">requests</span><span class="o">.</span><span class="n">post</span><span class="p">(</span>
- <span class="bp">self</span><span class="o">.</span><span class="n">ingest_post_url</span><span class="p">,</span> <span class="n">headers</span><span class="o">=</span><span class="bp">self</span><span class="o">.</span><span class="n">header</span><span class="p">,</span> <span class="n">data</span><span class="o">=</span><span class="n">query</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="bp">self</span><span class="o">.</span><span class="n">ingest_post_url</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">query</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">r</span><span class="o">.</span><span class="n">text</span><span class="p">)</span>
- <span class="n">d</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">r</span><span class="o">.</span><span class="n">text</span><span class="p">)</span>
- <span class="k">if</span> <span class="s2">"task"</span> <span class="ow">not</span> <span class="ow">in</span> <span class="n">d</span><span class="p">:</span>
- <span class="k">raise</span> <span class="n">AirflowDruidLoadException</span><span class="p">(</span>
- <span class="s2">"[Error]: Ingesting data to druid failed."</span><span class="p">)</span>
- <span class="k">return</span> <span class="n">d</span><span class="p">[</span><span class="s2">"task"</span><span class="p">]</span>
-
-<div class="viewcode-block" id="DruidHook.load_from_hdfs"><a class="viewcode-back" href="../code.html#airflow.hooks.DruidHook.load_from_hdfs">[docs]</a> <span class="k">def</span> <span class="nf">load_from_hdfs</span><span class="p">(</span>
- <span class="bp">self</span><span class="p">,</span> <span class="n">datasource</span><span class="p">,</span> <span class="n">static_path</span><span class="p">,</span> <span class="n">ts_dim</span><span class="p">,</span> <span class="n">columns</span><span class="p">,</span>
- <span class="n">intervals</span><span class="p">,</span> <span class="n">num_shards</span><span class="p">,</span> <span class="n">target_partition_size</span><span class="p">,</span> <span class="n">metric_spec</span><span class="o">=</span><span class="bp">None</span><span class="p">,</span> <span class="n">hadoop_dependency_coordinates</span><span class="o">=</span><span class="bp">None</span><span class="p">):</span>
- <span class="sd">"""</span>
-<span class="sd"> load data to druid from hdfs</span>
-
-<span class="sd"> :param ts_dim: The column name to use as a timestamp</span>
-<span class="sd"> :param metric_spec: A list of dictionaries</span>
-<span class="sd"> """</span>
- <span class="n">task_id</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">send_ingest_query</span><span class="p">(</span>
- <span class="n">datasource</span><span class="p">,</span> <span class="n">static_path</span><span class="p">,</span> <span class="n">ts_dim</span><span class="p">,</span> <span class="n">columns</span><span class="p">,</span> <span class="n">metric_spec</span><span class="p">,</span>
- <span class="n">intervals</span><span class="p">,</span> <span class="n">num_shards</span><span class="p">,</span> <span class="n">target_partition_size</span><span class="p">,</span> <span class="n">hadoop_dependency_coordinates</span><span class="p">)</span>
- <span class="n">status_url</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_ingest_status_url</span><span class="p">(</span><span class="n">task_id</span><span class="p">)</span>
- <span class="k">while</span> <span class="bp">True</span><span class="p">:</span>
- <span class="n">r</span> <span class="o">=</span> <span class="n">requests</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="n">status_url</span><span class="p">)</span>
- <span class="n">d</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">r</span><span class="o">.</span><span class="n">text</span><span class="p">)</span>
- <span class="k">if</span> <span class="n">d</span><span class="p">[</span><span class="s1">'status'</span><span class="p">][</span><span class="s1">'status'</span><span class="p">]</span> <span class="o">==</span> <span class="s1">'FAILED'</span><span class="p">:</span>
- <span class="n">logging</span><span class="o">.</span><span class="n">error</span><span class="p">(</span><span class="n">d</span><span class="p">)</span>
- <span class="k">raise</span> <span class="n">AirflowDruidLoadException</span><span class="p">(</span>
- <span class="s2">"[Error]: Ingesting data to druid failed."</span><span class="p">)</span>
- <span class="k">elif</span> <span class="n">d</span><span class="p">[</span><span class="s1">'status'</span><span class="p">][</span><span class="s1">'status'</span><span class="p">]</span> <span class="o">==</span> <span class="s1">'SUCCESS'</span><span class="p">:</span>
- <span class="k">break</span>
- <span class="n">time</span><span class="o">.</span><span class="n">sleep</span><span class="p">(</span><span class="n">LOAD_CHECK_INTERVAL</span><span class="p">)</span></div></div>
+ <span class="k">return</span> <span class="s2">"http://</span><span class="si">{host}</span><span class="s2">:</span><span class="si">{port}</span><span class="s2">/</span><span class="si">{endpoint}</span><span class="s2">"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="o">**</span><span class="nb">locals</span><span class="p">())</span>
+
+ <span class="k">def</span> <span class="nf">submit_indexing_job</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">json_index_spec</span><span class="p">):</span>
+ <span class="n">url</span> <span class="o">=</span> <span class="bp">self</span><span class="o">.</span><span class="n">get_conn_url</span><span class="p">()</span>
+
+ <span class="n">req_index</span> <span class="o">=</span> <span class="n">requests</span><span class="o">.</span><span class="n">post</span><span class="p">(</span><span class="n">url</span><span class="p">,</span> <span class="n">data</span><span class="o">=</span><span class="n">json_index_spec</span><span class="p">,</span> <span class="n">headers</span><span class="o">=</span><span class="bp">self</span><span class="o">.</span><span class="n">header</span><span class="p">)</span>
+ <span class="k">if</span> <span class="p">(</span><span class="n">req_index</span><span class="o">.</span><span class="n">status_code</span> <span class="o">!=</span> <span class="mi">200</span><span class="p">):</span>
+ <span class="k">raise</span> <span class="n">AirflowException</span><span class="p">(</span><span class="s2">"Did not get 200 when submitting the Druid job to </span><span class="si">{}</span><span class="s2">"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">url</span><span class="p">))</span>
+
+ <span class="n">req_json</span> <span class="o">=</span> <span class="n">req_index</span><span class="o">.</span><span class="n">json</span><span class="p">()</span>
+ <span class="c1"># Wait until the job is completed</span>
+ <span class="n">druid_task_id</span> <span class="o">=</span> <span class="n">req_json</span><span class="p">[</span><span class="s1">'task'</span><span class="p">]</span>
+
+ <span class="n">running</span> <span class="o">=</span> <span class="kc">True</span>
+
+ <span class="n">sec</span> <span class="o">=</span> <span class="mi">0</span>
+ <span class="k">while</span> <span class="n">running</span><span class="p">:</span>
+ <span class="n">req_status</span> <span class="o">=</span> <span class="n">requests</span><span class="o">.</span><span class="n">get</span><span class="p">(</span><span class="s2">"</span><span class="si">{0}</span><span class="s2">/</span><span class="si">{1}</span><span class="s2">/status"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">url</span><span class="p">,</span> <span class="n">druid_task_id</span><span class="p">))</span>
+
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s2">"Job still running for </span><span class="si">%s</span><span class="s2"> seconds..."</span><span class="p">,</span> <span class="n">sec</span><span class="p">)</span>
+
+ <span class="n">sec</span> <span class="o">=</span> <span class="n">sec</span> <span class="o">+</span> <span class="mi">1</span>
+
+ <span class="k">if</span> <span class="n">sec</span> <span class="o">></span> <span class="bp">self</span><span class="o">.</span><span class="n">max_ingestion_time</span><span class="p">:</span>
+ <span class="c1"># ensure that the job gets killed if the max ingestion time is exceeded</span>
+ <span class="n">requests</span><span class="o">.</span><span class="n">post</span><span class="p">(</span><span class="s2">"</span><span class="si">{0}</span><span class="s2">/</span><span class="si">{1}</span><span class="s2">/shutdown"</span><span class="o">.</span><span class="n">format</span><span class="p">(</span><span class="n">url</span><span class="p">,</span> <span class="n">druid_task_id</span><span class="p">))</span>
+ <span class="k">raise</span> <span class="n">AirflowException</span><span class="p">(</span><span class="s1">'Druid ingestion took more than </span><span class="si">%s</span><span class="s1"> seconds'</span><span class="p">,</span> <span class="bp">self</span><span class="o">.</span><span class="n">max_ingestion_time</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">timeout</span><span class="p">)</span>
+
+ <span class="n">status</span> <span class="o">=</span> <span class="n">req_status</span><span class="o">.</span><span class="n">json</span><span class="p">()[</span><span class="s1">'status'</span><span class="p">][</span><span class="s1">'status'</span><span class="p">]</span>
+ <span class="k">if</span> <span class="n">status</span> <span class="o">==</span> <span class="s1">'RUNNING'</span><span class="p">:</span>
+ <span class="n">running</span> <span class="o">=</span> <span class="kc">True</span>
+ <span class="k">elif</span> <span class="n">status</span> <span class="o">==</span> <span class="s1">'SUCCESS'</span><span class="p">:</span>
+ <span class="n">running</span> <span class="o">=</span> <span class="kc">False</span> <span class="c1"># Great success!</span>
+ <span class="k">elif</span> <span class="n">status</span> <span class="o">==</span> <span class="s1">'FAILED'</span><span class="p">:</span>
+ <span class="k">raise</span> <span class="n">AirflowException</span><span class="p">(</span><span class="s1">'Druid indexing job failed, check console for more info'</span><span class="p">)</span>
+ <span class="k">else</span><span class="p">:</span>
+ <span class="k">raise</span> <span class="n">AirflowException</span><span class="p">(</span><span class="s1">'Could not get status of the job, got </span><span class="si">%s</span><span class="s1">'</span><span class="p">,</span> <span class="n">status</span><span class="p">)</span>
+
+ <span class="bp">self</span><span class="o">.</span><span class="n">log</span><span class="o">.</span><span class="n">info</span><span class="p">(</span><span class="s1">'Successful index'</span><span class="p">)</span></div>
</pre></div>
</div>
+ <div class="articleComments">
+
+ </div>
</div>
<footer>
@@ -357,7 +297,8 @@
VERSION:'',
COLLAPSE_INDEX:false,
FILE_SUFFIX:'.html',
- HAS_SOURCE: true
+ HAS_SOURCE: true,
+ SOURCELINK_SUFFIX: '.txt'
};
</script>
<script type="text/javascript" src="../_static/jquery.js"></script>
http://git-wip-us.apache.org/repos/asf/incubator-airflow-site/blob/28a3eb60/_modules/dummy_operator.html
----------------------------------------------------------------------
diff --git a/_modules/dummy_operator.html b/_modules/dummy_operator.html
index a5452a1..e69044b 100644
--- a/_modules/dummy_operator.html
+++ b/_modules/dummy_operator.html
@@ -13,6 +13,8 @@
+
+
@@ -80,7 +82,10 @@
- <ul>
+
+
+
+ <ul>
<li class="toctree-l1"><a class="reference internal" href="../project.html">Project</a></li>
<li class="toctree-l1"><a class="reference internal" href="../license.html">License</a></li>
<li class="toctree-l1"><a class="reference internal" href="../start.html">Quick Start</a></li>
@@ -191,7 +196,7 @@
<span class="nd">@apply_defaults</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="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">):</span>
- <span class="nb">super</span><span class="p">(</span><span class="n">DummyOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="n">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
+ <span class="nb">super</span><span class="p">(</span><span class="n">DummyOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
<span class="k">def</span> <span class="nf">execute</span><span class="p">(</span><span class="bp">self</span><span class="p">,</span> <span class="n">context</span><span class="p">):</span>
<span class="k">pass</span></div>
http://git-wip-us.apache.org/repos/asf/incubator-airflow-site/blob/28a3eb60/_modules/email_operator.html
----------------------------------------------------------------------
diff --git a/_modules/email_operator.html b/_modules/email_operator.html
index ccb6373..45b816a 100644
--- a/_modules/email_operator.html
+++ b/_modules/email_operator.html
@@ -13,6 +13,8 @@
+
+
@@ -80,7 +82,10 @@
- <ul>
+
+
+
+ <ul>
<li class="toctree-l1"><a class="reference internal" href="../project.html">Project</a></li>
<li class="toctree-l1"><a class="reference internal" href="../license.html">License</a></li>
<li class="toctree-l1"><a class="reference internal" href="../start.html">Quick Start</a></li>
@@ -200,7 +205,7 @@
<span class="sd"> :type bcc: list or string (comma or semicolon delimited)</span>
<span class="sd"> """</span>
- <span class="n">template_fields</span> <span class="o">=</span> <span class="p">(</span><span class="s1">'subject'</span><span class="p">,</span> <span class="s1">'html_content'</span><span class="p">)</span>
+ <span class="n">template_fields</span> <span class="o">=</span> <span class="p">(</span><span class="s1">'to'</span><span class="p">,</span> <span class="s1">'subject'</span><span class="p">,</span> <span class="s1">'html_content'</span><span class="p">)</span>
<span class="n">template_ext</span> <span class="o">=</span> <span class="p">(</span><span class="s1">'.html'</span><span class="p">,)</span>
<span class="n">ui_color</span> <span class="o">=</span> <span class="s1">'#e6faf9'</span>
@@ -215,7 +220,7 @@
<span class="n">bcc</span><span class="o">=</span><span class="kc">None</span><span class="p">,</span>
<span class="n">mime_subtype</span><span class="o">=</span><span class="s1">'mixed'</span><span class="p">,</span>
<span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">):</span>
- <span class="nb">super</span><span class="p">(</span><span class="n">EmailOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="n">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
+ <span class="nb">super</span><span class="p">(</span><span class="n">EmailOperator</span><span class="p">,</span> <span class="bp">self</span><span class="p">)</span><span class="o">.</span><span class="fm">__init__</span><span class="p">(</span><span class="o">*</span><span class="n">args</span><span class="p">,</span> <span class="o">**</span><span class="n">kwargs</span><span class="p">)</span>
<span class="bp">self</span><span class="o">.</span><span class="n">to</span> <span class="o">=</span> <span class="n">to</span>
<span class="bp">self</span><span class="o">.</span><span class="n">subject</span> <span class="o">=</span> <span class="n">subject</span>
<span class="bp">self</span><span class="o">.</span><span class="n">html_content</span> <span class="o">=</span> <span class="n">html_content</span>