{"action":"opened","issue":{"url":"https://api.github.com/repos/guevara/read-it-later/issues/1134","repository_url":"https://api.github.com/repos/guevara/read-it-later","labels_url":"https://api.github.com/repos/guevara/read-it-later/issues/1134/labels{/name}","comments_url":"https://api.github.com/repos/guevara/read-it-later/issues/1134/comments","events_url":"https://api.github.com/repos/guevara/read-it-later/issues/1134/events","html_url":"https://github.com/guevara/read-it-later/issues/1134","id":372306373,"node_id":"MDU6SXNzdWUzNzIzMDYzNzM=","number":1134,"title":"Spark SQL / Catalyst 内部原理 与 RBO","user":{"login":"guevara","id":228309,"node_id":"MDQ6VXNlcjIyODMwOQ==","avatar_url":"https://avatars1.githubusercontent.com/u/228309?v=4","gravatar_id":"","url":"https://api.github.com/users/guevara","html_url":"https://github.com/guevara","followers_url":"https://api.github.com/users/guevara/followers","following_url":"https://api.github.com/users/guevara/following{/other_user}","gists_url":"https://api.github.com/users/guevara/gists{/gist_id}","starred_url":"https://api.github.com/users/guevara/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/guevara/subscriptions","organizations_url":"https://api.github.com/users/guevara/orgs","repos_url":"https://api.github.com/users/guevara/repos","events_url":"https://api.github.com/users/guevara/events{/privacy}","received_events_url":"https://api.github.com/users/guevara/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-10-21T11:47:11Z","updated_at":"2018-10-21T11:47:11Z","closed_at":null,"author_association":"OWNER","body":"Spark SQL / Catalyst 内部原理 与 RBO<br>\n<br>\nhttps://ift.tt/2PLNFyf<br>\n<br>\n郭俊 Jason<br>\n<br>\n<div><div>\n<div>\n<blockquote>\n<p>原创文章，转载请务必将下面这段话置于文章开头处。<br>本文转发自<a href=\"http://www.jasongj.com\"><strong>技术世界</strong></a>，<a href=\"http://www.jasongj.com/spark/rbo/\">原文链接</a>　<a href=\"http://www.jasongj.com/spark/rbo/\">http://www.jasongj.com/spark/rbo/</a></p>\n</blockquote>\n<blockquote>\n<p>本文所述内容均基于 2018年9月10日 Spark 最新 Release 2.3.1 版本。后续将持续更新</p>\n</blockquote>\n<h1>\n<a title=\"Spark SQL 架构\" href=\"http://www.jasongj.com/spark/rbo/#Spark-SQL-%E6%9E%B6%E6%9E%84\"></a>Spark SQL 架构</h1>\n<p>Spark SQL 的整体架构如下图所示<br><img alt=\"Spark SQL Catalyst\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql.png\"></p>\n<p>从上图可见，无论是直接使用 SQL 语句还是使用 DataFrame，都会经过如下步骤转换成 DAG 对 RDD 的操作</p>\n<ul>\n<li>Parser 解析 SQL，生成 Unresolved Logical Plan</li>\n<li>由 Analyzer 结合 Catalog 信息生成 Resolved Logical Plan</li>\n<li>Optimizer根据预先定义好的规则对 Resolved Logical Plan 进行优化并生成 Optimized Logical Plan</li>\n<li>Query Planner 将 Optimized Logical Plan 转换成多个 Physical Plan</li>\n<li>CBO 根据 Cost Model 算出每个 Physical Plan 的代价并选取代价最小的 Physical Plan 作为最终的 Physical Plan</li>\n<li>Spark 以 DAG 的方法执行上述 Physical Plan</li>\n<li>在执行 DAG 的过程中，Adaptive Execution 根据运行时信息动态调整执行计划从而提高执行效率</li>\n</ul>\n<h1>\n<a title=\"Parser\" href=\"http://www.jasongj.com/spark/rbo/#Parser\"></a>Parser</h1>\n<p>Spark SQL 使用 Antlr 进行记法和语法解析，并生成 UnresolvedPlan。</p>\n<p>当用户使用 SparkSession.sql(sqlText : String) 提交 SQL 时，SparkSession 最终会调用 SparkSqlParser 的 parsePlan 方法。该方法分两步</p>\n<ul>\n<li>使用 Antlr 生成的 SqlBaseLexer 对 SQL 进行词法分析，生成 CommonTokenStream</li>\n<li>使用 Antlr 生成的 SqlBaseParser 进行语法分析，得到 LogicalPlan</li>\n</ul>\n<p>现在两张表，分别定义如下<br><figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br></pre></td>\n<td><pre><span><span>CREATE</span> <span>TABLE</span> score (</span><br><span> <span>id</span> <span>INT</span>,</span><br><span> math_score <span>INT</span>,</span><br><span> english_score <span>INT</span></span><br><span>)</span><br></pre></td>\n</tr></table></figure></p>\n<figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br></pre></td>\n<td><pre><span><span>CREATE</span> <span>TABLE</span> people (</span><br><span> <span>id</span> <span>INT</span>,</span><br><span> age <span>INT</span>,</span><br><span> <span>name</span> <span>INT</span></span><br><span>)</span><br></pre></td>\n</tr></table></figure>\n<p>对其进行关联查询如下<br><figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br><span>6</span><br><span>7</span><br><span>8</span><br><span>9</span><br></pre></td>\n<td><pre><span><span>SELECT</span> <span>sum</span>(v)</span><br><span><span>FROM</span> (</span><br><span> <span>SELECT</span> score.id,</span><br><span> <span>100</span> + <span>80</span> + score.math_score + score.english_score <span>AS</span> v</span><br><span> <span>FROM</span> people</span><br><span> <span>JOIN</span> score</span><br><span> <span>ON</span> people.id = score.id</span><br><span> <span>AND</span> people.age &gt; <span>10</span></span><br><span>) tmp</span><br></pre></td>\n</tr></table></figure></p>\n<p>生成的 UnresolvedPlan 如下图所示。</p>\n<p><img alt=\"Spark SQL Parser\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_parser.png\"></p>\n<p>从上图可见</p>\n<ul>\n<li>查询涉及的两张表，被解析成了两个 UnresolvedRelation，也即只知道这们是两张表，却并不知道它们是 EXTERNAL TABLE 还是 MANAGED TABLE，也不知道它们的数据存在哪儿，更不知道它们的表结构如何</li>\n<li>sum(v) 的结果未命名</li>\n<li>Project 部分只知道是选择出了属性，却并不知道这些属性属于哪张表，更不知道其数据类型</li>\n<li>Filter 部分也不知道数据类型</li>\n</ul>\n<p>Spark SQL 解析出的 UnresolvedPlan 如下所示<br><figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br><span>6</span><br><span>7</span><br><span>8</span><br></pre></td>\n<td><pre><span>== Parsed Logical Plan ==</span><br><span>'Project [unresolvedalias('sum('v), None)]</span><br><span>+- 'SubqueryAlias tmp</span><br><span> +- 'Project ['score.id, (((100 + 80) + 'score.math_score) + 'score.english_score) AS v#493]</span><br><span> +- 'Filter (('people.id = 'score.id) &amp;&amp; ('people.age &gt; 10))</span><br><span> +- 'Join Inner</span><br><span> :- 'UnresolvedRelation `people`</span><br><span> +- 'UnresolvedRelation `score`</span><br></pre></td>\n</tr></table></figure></p>\n<h1>\n<a title=\"Analyzer\" href=\"http://www.jasongj.com/spark/rbo/#Analyzer\"></a>Analyzer</h1>\n<p>从 Analyzer 的构造方法可见</p>\n<ul>\n<li>Analyzer 持有一个 SessionCatalog 对象的引用</li>\n<li>Analyzer 继承自 RuleExecutor[LogicalPlan]，因此可对 LogicalPlan 进行转换</li>\n</ul>\n<figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br></pre></td>\n<td><pre><span><span><span>class</span> <span>Analyzer</span>(<span></span></span></span><br><span><span><span> catalog: <span>SessionCatalog</span>,</span></span></span><br><span><span><span> conf: <span>SQLConf</span>,</span></span></span><br><span><span><span> maxIterations: <span>Int</span></span>)</span></span><br><span><span> <span>extends</span> <span>RuleExecutor</span>[<span>LogicalPlan</span>] <span>with</span> <span>CheckAnalysis</span> </span>{</span><br></pre></td>\n</tr></table></figure>\n<p>Analyzer 包含了如下的转换规则<br><figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br><span>6</span><br><span>7</span><br><span>8</span><br><span>9</span><br><span>10</span><br><span>11</span><br><span>12</span><br><span>13</span><br><span>14</span><br><span>15</span><br><span>16</span><br><span>17</span><br><span>18</span><br><span>19</span><br><span>20</span><br><span>21</span><br><span>22</span><br><span>23</span><br><span>24</span><br><span>25</span><br><span>26</span><br><span>27</span><br><span>28</span><br><span>29</span><br><span>30</span><br><span>31</span><br><span>32</span><br><span>33</span><br><span>34</span><br><span>35</span><br><span>36</span><br><span>37</span><br><span>38</span><br><span>39</span><br><span>40</span><br><span>41</span><br><span>42</span><br><span>43</span><br><span>44</span><br><span>45</span><br><span>46</span><br><span>47</span><br><span>48</span><br><span>49</span><br><span>50</span><br><span>51</span><br><span>52</span><br><span>53</span><br><span>54</span><br><span>55</span><br><span>56</span><br></pre></td>\n<td><pre><span><span>lazy</span> <span>val</span> batches: <span>Seq</span>[<span>Batch</span>] = <span>Seq</span>(</span><br><span> <span>Batch</span>(<span>\"Hints\"</span>, fixedPoint,</span><br><span> <span>new</span> <span>ResolveHints</span>.<span>ResolveBroadcastHints</span>(conf),</span><br><span> <span>ResolveHints</span>.<span>RemoveAllHints</span>),</span><br><span> <span>Batch</span>(<span>\"Simple Sanity Check\"</span>, <span>Once</span>,</span><br><span> <span>LookupFunctions</span>),</span><br><span> <span>Batch</span>(<span>\"Substitution\"</span>, fixedPoint,</span><br><span> <span>CTESubstitution</span>,</span><br><span> <span>WindowsSubstitution</span>,</span><br><span> <span>EliminateUnions</span>,</span><br><span> <span>new</span> <span>SubstituteUnresolvedOrdinals</span>(conf)),</span><br><span> <span>Batch</span>(<span>\"Resolution\"</span>, fixedPoint,</span><br><span> <span>ResolveTableValuedFunctions</span> ::</span><br><span> <span>ResolveRelations</span> ::</span><br><span> <span>ResolveReferences</span> ::</span><br><span> <span>ResolveCreateNamedStruct</span> ::</span><br><span> <span>ResolveDeserializer</span> ::</span><br><span> <span>ResolveNewInstance</span> ::</span><br><span> <span>ResolveUpCast</span> ::</span><br><span> <span>ResolveGroupingAnalytics</span> ::</span><br><span> <span>ResolvePivot</span> ::</span><br><span> <span>ResolveOrdinalInOrderByAndGroupBy</span> ::</span><br><span> <span>ResolveAggAliasInGroupBy</span> ::</span><br><span> <span>ResolveMissingReferences</span> ::</span><br><span> <span>ExtractGenerator</span> ::</span><br><span> <span>ResolveGenerate</span> ::</span><br><span> <span>ResolveFunctions</span> ::</span><br><span> <span>ResolveAliases</span> ::</span><br><span> <span>ResolveSubquery</span> ::</span><br><span> <span>ResolveSubqueryColumnAliases</span> ::</span><br><span> <span>ResolveWindowOrder</span> ::</span><br><span> <span>ResolveWindowFrame</span> ::</span><br><span> <span>ResolveNaturalAndUsingJoin</span> ::</span><br><span> <span>ExtractWindowExpressions</span> ::</span><br><span> <span>GlobalAggregates</span> ::</span><br><span> <span>ResolveAggregateFunctions</span> ::</span><br><span> <span>TimeWindowing</span> ::</span><br><span> <span>ResolveInlineTables</span>(conf) ::</span><br><span> <span>ResolveTimeZone</span>(conf) ::</span><br><span> <span>ResolvedUuidExpressions</span> ::</span><br><span> <span>TypeCoercion</span>.typeCoercionRules(conf) ++</span><br><span> extendedResolutionRules : _*),</span><br><span> <span>Batch</span>(<span>\"Post-Hoc Resolution\"</span>, <span>Once</span>, postHocResolutionRules: _*),</span><br><span> <span>Batch</span>(<span>\"View\"</span>, <span>Once</span>,</span><br><span> <span>AliasViewChild</span>(conf)),</span><br><span> <span>Batch</span>(<span>\"Nondeterministic\"</span>, <span>Once</span>,</span><br><span> <span>PullOutNondeterministic</span>),</span><br><span> <span>Batch</span>(<span>\"UDF\"</span>, <span>Once</span>,</span><br><span> <span>HandleNullInputsForUDF</span>),</span><br><span> <span>Batch</span>(<span>\"FixNullability\"</span>, <span>Once</span>,</span><br><span> <span>FixNullability</span>),</span><br><span> <span>Batch</span>(<span>\"Subquery\"</span>, <span>Once</span>,</span><br><span> <span>UpdateOuterReferences</span>),</span><br><span> <span>Batch</span>(<span>\"Cleanup\"</span>, fixedPoint,</span><br><span> <span>CleanupAliases</span>)</span><br><span> )</span><br></pre></td>\n</tr></table></figure></p>\n<p>例如， ResolveRelations 用于分析查询用到的 Table 或 View。本例中 UnresolvedRelation (people) 与 UnresolvedRelation (score) 被解析为 HiveTableRelation (json.people) 与 HiveTableRelation (json.score)，并列出其各自包含的字段名。</p>\n<p>经 Analyzer 分析后得到的 Resolved Logical Plan 如下所示</p>\n<figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br><span>6</span><br><span>7</span><br><span>8</span><br><span>9</span><br><span>10</span><br><span>11</span><br></pre></td>\n<td><pre><span>== Analyzed Logical Plan ==</span><br><span>sum(v): bigint</span><br><span>Aggregate [sum(cast(v#493 as bigint)) AS sum(v)#504L]</span><br><span>+- SubqueryAlias tmp</span><br><span> +- Project [id#500, (((100 + 80) + math_score#501) + english_score#502) AS v#493]</span><br><span> +- Filter ((id#496 = id#500) &amp;&amp; (age#497 &gt; 10))</span><br><span> +- Join Inner</span><br><span> :- SubqueryAlias people</span><br><span> : +- HiveTableRelation `jason`.`people`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#496, age#497, name#498]</span><br><span> +- SubqueryAlias score</span><br><span> +- HiveTableRelation `jason`.`score`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#500, math_score#501, english_score#502]</span><br></pre></td>\n</tr></table></figure>\n<p>Analyzer 分析前后的 LogicalPlan 对比如下</p>\n<p><img alt=\"Spark SQL Analyzer\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_analyzer.png\"></p>\n<p>由上图可见，分析后，每张表对应的字段集，字段类型，数据存储位置都已确定。Project 与 Filter 操作的字段类型以及在表中的位置也已确定。</p>\n<p>有了这些信息，已经可以直接将该 LogicalPlan 转换为 Physical Plan 进行执行。</p>\n<p>但是由于不同用户提交的 SQL 质量不同，直接执行会造成不同用户提交的语义相同的不同 SQL 执行效率差距甚远。换句话说，如果要保证较高的执行效率，用户需要做大量的 SQL 优化，使用体验大大降低。</p>\n<p>为了尽可能保证无论用户是否熟悉 SQL 优化，提交的 SQL 质量如何， Spark SQL 都能以较高效率执行，还需在执行前进行 LogicalPlan 优化。</p>\n<h1>\n<a title=\"Optimizer\" href=\"http://www.jasongj.com/spark/rbo/#Optimizer\"></a>Optimizer</h1>\n<p>Spark SQL 目前的优化主要是基于规则的优化，即 RBO （Rule-based optimization）</p>\n<ul>\n<li>每个优化以 Rule 的形式存在，每条 Rule 都是对 Analyzed Plan 的等价转换</li>\n<li>RBO 设计良好，易于扩展，新的规则可以非常方便地嵌入进 Optimizer</li>\n<li>RBO 目前已经足够好，但仍然需要更多规则来 cover 更多的场景</li>\n<li>优化思路主要是减少参与计算的数据量以及计算本身的代价</li>\n</ul>\n<p><strong><em>PushdownPredicate</em></strong><br>PushdownPredicate 是最常见的用于减少参与计算的数据量的方法。</p>\n<p>前文中直接对两表进行 Join 操作，然后再 进行 Filter 操作。引入 PushdownPredicate 后，可先对两表进行 Filter 再进行 Join，如下图所示。</p>\n<p><img alt=\"Spark SQL RBO Predicate Pushdown\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_predicate_pushdown.png\"></p>\n<p>当 Filter 可过滤掉大部分数据时，参与 Join 的数据量大大减少，从而使得 Join 操作速度大大提高。</p>\n<p>这里需要说明的是，此处的优化是 LogicalPlan 的优化，从逻辑上保证了将 Filter 下推后由于参与 Join 的数据量变少而提高了性能。另一方面，在物理层面，Filter 下推后，对于支持 Filter 下推的 Storage，并不需要将表的全量数据扫描出来再过滤，而是直接只扫描符合 Filter 条件的数据，从而在物理层面极大减少了扫描表的开销，提高了执行速度。</p>\n<p><strong><em>ConstantFolding</em></strong><br>本文的 SQL 查询中，Project 部分包含了 100 + 800 + match_score + english_score 。如果不进行优化，那如果有一亿条记录，就会计算一亿次 100 + 80，非常浪费资源。因此可通过 ConstantFolding 将这些常量合并，从而减少不必要的计算，提高执行速度。</p>\n<p><img alt=\"Spark SQL RBO Constant Folding\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_constant_folding.png\"></p>\n<p><strong><em>ColumnPruning</em></strong><br>在上图中，Filter 与 Join 操作会保留两边所有字段，然后在 Project 操作中筛选出需要的特定列。如果能将 Project 下推，在扫描表时就只筛选出满足后续操作的最小字段集，则能大大减少 Filter 与 Project 操作的中间结果集数据量，从而极大提高执行速度。</p>\n<p><img alt=\"Spark SQL RBO Column Pruning\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_column_pruning.png\"></p>\n<p>这里需要说明的是，此处的优化是逻辑上的优化。在物理上，Project 下推后，对于列式存储，如 Parquet 和 ORC，可在扫描表时就只扫描需要的列而跳过不需要的列，进一步减少了扫描开销，提高了执行速度。</p>\n<p>经过如上优化后的 LogicalPlan 如下</p>\n<figure><table><tr>\n<td><pre><span>1</span><br><span>2</span><br><span>3</span><br><span>4</span><br><span>5</span><br><span>6</span><br><span>7</span><br><span>8</span><br><span>9</span><br></pre></td>\n<td><pre><span>== Optimized Logical Plan ==</span><br><span>Aggregate [sum(cast(v#493 as bigint)) AS sum(v)#504L]</span><br><span>+- Project [((180 + math_score#501) + english_score#502) AS v#493]</span><br><span> +- Join Inner, (id#496 = id#500)</span><br><span> :- Project [id#496]</span><br><span> : +- Filter ((isnotnull(age#497) &amp;&amp; (age#497 &gt; 10)) &amp;&amp; isnotnull(id#496))</span><br><span> : +- HiveTableRelation `jason`.`people`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#496, age#497, name#498]</span><br><span> +- Filter isnotnull(id#500)</span><br><span> +- HiveTableRelation `jason`.`score`, org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [id#500, math_score#501, english_score#502]</span><br></pre></td>\n</tr></table></figure>\n<h1>\n<a title=\"SparkPlanner\" href=\"http://www.jasongj.com/spark/rbo/#SparkPlanner\"></a>SparkPlanner</h1>\n<p>得到优化后的 LogicalPlan 后，SparkPlanner 将其转化为 SparkPlan 即物理计划。</p>\n<p>本例中由于 score 表数据量较小，Spark 使用了 BroadcastJoin。因此 score 表经过 Filter 后直接使用 BroadcastExchangeExec 将数据广播出去，然后结合广播数据对 people 表使用 BroadcastHashJoinExec 进行 Join。再经过 Project 后使用 HashAggregateExec 进行分组聚合。</p>\n<p><img alt=\"Spark SQL RBO Column Pruning\" src=\"http://www.jasongj.com/img/spark/spark2_rbo/spark_sql_physical_plan.png\"></p>\n<p>至此，一条 SQL 从提交到解析、分析、优化以及执行的完整过程就介绍完毕。</p>\n<p>本文介绍的 Optimizer 属于 RBO，实现简单有效。它属于 LogicalPlan 的优化，所有优化均基于 LogicalPlan 本身的特点，未考虑数据本身的特点，也未考虑算子本身的代价。下文将介绍 CBO，它充分考虑了数据本身的特点（如大小、分布）以及操作算子的特点（中间结果集的分布及大小）及代价，从而更好的选择执行代价最小的物理执行计划，即 SparkPlan。</p>\n<h1>\n<a title=\"Spark 系列文章\" href=\"http://www.jasongj.com/spark/rbo/#Spark-%E7%B3%BB%E5%88%97%E6%96%87%E7%AB%A0\"></a>Spark 系列文章</h1>\n</div>\n</div></div><br>\n<br>\n<br>\n<br>\n<br>\n<br>\nvia 技术世界 <br>\n<br>\nOctober 21, 2018 at 07:46PM"}}
{"actor":{"display_login":"guevara"}}