Apache Drill的Logical Plan的语法

原创，转载请注明出处。

利用业余时间完成了对Drill的两篇文档的翻译，希望方便大家学习drill。

本文是Drill的Logical Plan的翻译稿。drill logical plan是内部信息，虽然我们使用时候不需要关注logical plan，但是对于开发，这个还是需要了解。


前言
Drill的一个目标是在函数层定义清晰的接口以便将来的扩展需求。这就需要理解Drill组件和通用字典。Drill的执行流是由用户或者机器产生的查询请求，被提交给query parser。在query parser中查询请求会被解析成为logical plan. logical plan描述了查询操作的基本数据流。query parser之后的组件是优化器，它读取logical plan并把logical plan转换为physical plan，然后由execution engine来执行physical plan。execution engine依赖于很对组件，比如说各种operator，scanner（读），writers.。另外一个metadata repository 为Drill 查询执行框架提供元数据信息。
Drill第一版中关键组件如下：

Query Parser:通过metadata api请求schema信息，然后产生标准logical plan以供优化器进行优化。
Storage Engine:能够和特定的数据源进行交互，提供读和写的功能，将数据的元数据信息提供给metadata repo，包括schema，file size，data ordering, secondary indices, number of blocks,将native 功能（predicate pushdown，joins，SQL）告知execution engine。
Operator：负责转换数据流
Function：
　　　聚合函数：将一个数据流转换一个单独的data
　　　Scalar Functions：将一个或者多个Scalar data转换为一个单独的data




数据结构
Drill操作的数据对象是树形数据。你可以将之理解成为JSON结构的数据。

{
“key”: “123”,
“value”: {
　　　　“foo”: { “x”: “y”},
　　　　“blue” : [
　　　　 “red”,
　　　　 “blue”
　　　　]
}
       }

或者
{
“key”: “123”,
“value”: {
　　　　“foo”: “bar”,
　　　　“blue” : “orange”
}
}


需要说明的时候，Logical plan和physical plan的抽象数据结构是用JSON来表示的有向无环图。

Data Typing

Plan 内容的关键是一些operator和这些operator的定义，需要注意的时候包括他们作为输入和输出的数据的类型。优化器和执行引擎可以从query plan本身中提取这些数据的类型，除此之外，只有在处理数据的上下文中设置数据type信息，优化器和执行引擎才能动态接受和解析这些type信息。然而目前为止，怎么确定这些信息的类型和具体指定何种type还没有确定下来。
尽管缺乏对信息类型的明确定义，drill尽可能锁定了大多数的operator。

Logical Plan
Purpose
Logical Plan描述了一个语言无关查询请求的抽象数据流。他主要的配合原始查询操作，而不关注优化。这导致了drill查询会比传统的查询要复杂的多。但是这允许了定义高级层次查询语句的灵活性。典型的，logical plan由query parser产生。当schema存在的时候，query parser也同样能够利用这些信息，进一步验证query请求，构造一个logical plan。

这个logical plan，是一个数据流operator的有向无环图（DAG）。DAG的边缘限制了数据的流向。（注意：和传统程序的SSA（静态单赋值形式）相反的是，drill的DAG的SSA的特征允许作为参数的高级有序标量函数表达式，以及直接作为SSA一部分expression，这会便于人类理解，并且更加简洁）。因为logical plan是数据流的抽象代表，每一个operator只能包含一个输出（这不同于physical plan包含多个operator并且包含多个输出）。反观，某个特殊的operator可以订阅其他operator的输出。
因为支持层级数据，Dril提供了很多内部（implosion）外部（explosion）的操作。这些操作允许对于内嵌值的操作也同样能够被一些子DAG来控制。因为每一个操作都是独立的，需要在流中提供一些re-implosion数据。


Logical Plan Vocabulary
Arguments
Logical plan描述了数据集以及logical operator之间的相互关系，每一个operator都拥有若干参数。如下所述：
<name>：name被用来作为记录流中的一个定义，或者是一个特殊输出filed的引用
<string>：引用String
<expr>:值表达式中可能包含若干值函数，值，或者filed的引用。注意，一个值表达式，可以返回一个标量值，或者是负载的数据类型（比如map，或者array）
<aggexpr>：表达式可以包含若干标量表达式（scaler expression）同样能够包含若干aggregate 表达式，比如说（SUM，COUNT，etc）。需要注意的是，一个aggregate 表达式不能内嵌多个aggregate中，但是允许scaler expression引用在两个单独的aggregate之上。
<runaggexpr>:包含aggregate的表达式可以支持运行时操作。
<opref>:整数型定义，或者引用某个特殊的operator
<json>：一种json 类型的string
<operator>：一个operator的定义或者是一个 <opref>

当某些特殊的value是可选的时候，这些value被用星号标记。在sequence operator的Do 参数的上下文中，某些value是可选的，上下文中的可选值，会被使用+表示。

关键概念：
   Value：apache drill 中的value可能是一个标量类型，比如说e (int32, int64, uint32, float32,等等），或者是一个array，或者是一个map。
　　Record：record是多个value的集合，类似于RDBMS中一个行
　　Field：一个record中某个field的位置
　　Stream：是从一个operator输出并且是作为另外一个operator输入的record集。
　　Segments:这是Logical Plan中一个重要的概念。Segment 定义了输入record的一部分子集，这些子集由于某些目的被搜集，在大多数情况下，一个operator能够被提供一个segment key，然后可以分别对segment进行操作，避免了segment跨界操作。

Logical Plan Operators
Operator分成若干类，每一个operator都标示了他的类型，目前operator类包括：
0： 可以产生不依赖其他operator的数据，类似于源数据
1：该operator可以处理一个单独input source
M：可以处理多个input source数据
K：该operator不会产生输出。

属性：
┼:在序列（sequence）外是需要的该标记来表示，而在序列内部则不允许。
注意这里的sequence是一个语法上面的operator，参见最后的sequence部分）
This property is required outside a sequence and not allowed within a sequence.
*: 该属性可选

Scan (0)
输出一个record的流， "storageengine”参数必须要引用一个logical plan中定义了engine 字句的名称，selection参数接收一个json对象，这个json对象可以被数据源自身利用以显示提取的数据数目大小，这个对象的格式和内容仅限于实际正在使用输入源。
Ref参数表示，所有本扫描的数据源中的record都会被放置一个指定的命名空间
{ @id†: <opref>, op: “scan”,
storageengine: <string>,
selection*: <json>,
ref: <name>
}

Constant (0)
Constant返回一个固定结果。1.比如你不确定你的表存不存在，2或者需要评估某个表达式是否属于某个table，这和SQL中VALUES子句的参数是有点类似
{ @id†: <opref>, op: “constant”,
content: <json>
}
例如：
SQL    VALUES (1, 'iamastr') AS t(c1, c2)
而在drill中：
DLP    { @id: 1, op: “constant”, content: { [ { c1: 1, c2: “iamastr” } ] } }
这里的实现目前，是通过使用隐式type，相关讨论信息可以参考JIRA page #57，在将来会加入显示type，drill开发者还没有决定type的具体格式，但是已经进行了大量的讨论，可以参考：
https://docs.google.com/document/d/1nn8sxcuBvpAHm-BoreCWELPIQ8QhhsAUSENX0s5sSys/edit

Join (M)
根据若干join condition，连接两个input。该operator的输出是两个input的集合。这通过将两个input 记录匹配连接条件后，按照连接条件进行组合。如果没有提供join 条件，那么就进行笛卡尔乘积。如果连接的左边记录是{donuts: [data]}，而右边是{purchases: [data]}，combination结果就是{donuts:[data], purchases: [data]}，在连接的时候，同样需要一个type变量来表示join类型。有inner、outer、left类型。

{ @id†: <opref>, op: “join”,
left: <input>,
right: <input>,
type: <inner|outer|left>,
conditions*: [
{relationship: <reltype>, left: <expr>, right: <expr>}, ...
]
}
其中reltype，可以是>, >=, <=, <, !=, ==，这类似于SQL的JOINON的条件语句

Union (M)
将多个输入数据合并成为一个单独steam，这里不需要顺序。如果distinct=true，那么重复的数据就会移除。这类似于Union和Union All。
{ @id†: <opref>, op: “union”,
distinct: <true|false>,
inputs: [
<input>, ... <input>
]
}


Store (1 K)
将stream的输出保存存到一个storage engine中，在logical plan中 storageengine 参数引用了一个已经定义好的storage engine。Target 参数描述了storage engine的特殊参数（比如，存储类型，filename等等）。Partition 的值描述了如何切分output data，切分可以是随机的，hash，或者是基于顺序的。另外一个可选参数start，描述了目标partition集。如果start没有定义，那么系统会根据query语句以及执行计划自定义一个合适的partition集合。因为logical plan并不是一个对用户直接开放的接口，drill通过一个特殊的sink 类型输出到终端，展示个用户（该sink 被称为rpc sink）。Logical plan 接口的消费者负责设置一个inbound endpoint ，通过该endpoint，执行引擎层能够获取到结果。
{ @id†: <opref>, op: “store”,
input†: <input>,
storageengine: <string>,
target: <json>,
partition*: {
type: <RANDOM|HASH|ORDERED>,
exprs: [<expr>, … <expr>],
starts*: [<expr>, … <expr>]
}
}

Project (1)
返回和对应expression相对应的输入数据流的一个子集。Projections参数定义了在输出record中要保留的字段，对于每一个projection，输出record的value的名称是通过ref来指定，而expression被应用在输入record上，输出相应的计算结果。为了方便区分，可以将projection的那么设置为以output开头，比如说：{ref: “output”，expr: “sum(donuts.sales)}”。若有多个project重叠，那么后面的projection会覆盖掉前面的projection。
{ @id†: <opref>, op: “project”,
input†: <input>,
projections: [
{ref: <name>, expr: <expr>}, ...
]
}

Order (1)
该operator可以根据若干个order expression来对输入数据流进行排序，排序可以指定应用在某些特殊的segment中，通过指定within参数。Segment可以通过其他的operator来定义，比如说group operator。一旦segment需要被排序，那么排序只能在每隔segment内部进行运行。
nullCollation是可选择参数，该参数定义null值放置在order后的前端还是末尾。
默认是放置在order的前端。
{ @id†: <opref>, op: “order”,
input†: <input>,
within*: <name>,
orderings: [
{order: <desc|asc>, expr: <expr>, nullCollation: <first|last> }, ...
]
}


Filter (1)
该operator可以根据expression，移除数据流中某些数值。对于每一个输入record，当expression的结果为true的时候，那么该record就会被输出，反之，那么该条记录就不会被输出。
{ @id†: <opref>, op: “filter”,
input†: <input>,
expr: <expr>
}


Transform (1)
转换一个数据，就是为了该数据允许被引用，或者是数据流更加清晰。转换数据的expression可以是匿名的，也可以是明确定义的，这些expression是不能在数据的output 流中被使用。
例如：当使用segment operator，如果group expression不是简单的字段引用，那么group expression 就会被引用到segment，然后被丢弃。如果该group后的结果需要被保留，那么，需要在segment 操作之前或者之后应用transform参数。
Transform参数中，expr是一个已经存在的field 引用，而ref是一个新的field 引用。

Limit (1)
限制输出是输入的一个前缀部分。输入record隐式的以0开始。First，last参数分别定义了输出第一个到最后一个之间的所有记录，类似于mysql的limit的语言。
比如Last=0，返回0条记录，而First = 0, Last = 1返回第一个对象。Limit是针对整个记录集。
{ @id†: <opref>, op: “limit”,
input†: <input>,
first: <number>,
last: <number>
}


Segment (1)
该operator负责搜集所有在应用了expression的基础上面具有相同值的记录，并且将这些记录作为一个单独segment（可以称为group）。每一个输出的record对于某个expression都具有相同value。Segment operator 也能够输出一个用来表示expression值的key 。
Segment operator是稳定，这就意味了在segment中记录的顺序是和segment的输入顺序是一致的。然后，输出的时候不能保证这种顺序。
{ @id†: <opref>, op: “segment”,
input†: <input>,
ref: <name>,
exprs: [<expr>,..., <expr>]
}



Window Frame (1)
对于输入stream中每一条记录（我们称为target record），window operator 都保持了一个可滑动的窗口。该窗口在某条target record之前或者之后，包含一定数量的records。
window frame operator 可以操作整个input steam，或者只操作每一个输入的segment。
一个简单的例子：若window的start=-2，end=0，以一个包含了5条记录的segment作为输入，那么输出就包含12行：[0] [0,1] [0,1,2] [1,2,3] [2,3,4]，每一条输出record，都包含两个另外的字段：ref.segment和ref.position。其中ref.segment表示当前window segment，ref.position的取值可以为正，可以为负，可以为0。对这个简单的例子，ref.segment: [0,1,1,2,2,2,3,3,3,4,4,4],ref.position: [0,-1,0,-1,0,1,-1,0,1,-1,0,1]。
（注：我是不明白这个对应关系是什么）。
window operator可以使用一个segment的key作为输入，该key是为了限制window operator仅作用于某个segment，而不会跨界操作。
Start和end参数，限定了window的范围。Start和end可以同时指定，也可以指定一个，但是start必须要小于end。
{ @id†: <opref>, op: “windowframe”,
input†: <input>,
within*: <name>,
start*: <number>,
end*: <number>
ref: {
segment: <name>,
position: <name>
},
}


CollapsingAggregate (1)
（这个词 真心不知道怎么翻译，姑且按照字面意思来确定崩溃汇总，看到他的作用可以称为单行汇总）
collapse aggregate会将一个segment的所有记录汇总为一条record。当没有定义segment（即没有设定within），那么collapse aggregate会将
collapse aggregate的输出只限于已经提供了aggregation语句的值，并且value的值被定义为carryovers。collapse aggregate也可以利用一个可选的target field，引用了一个record来携带carryover values。
在target没有定义的时候，collapse aggregate会自由的选择提取carryover value的record（因为每一个record的值都相等）。如果target设定了，会从一条target field 为true的record中提取carryover 变量。如果多个record都包含一个target field 的值为true，那么carryover value可以从其中一条record中提取。若target segment中没有一条记录record的target filed字段为true，将不会有record从target segment中被提取出来。每一个segment不会有多余一个record被选择去carryover value。
{ @id†: <opref>, op: “collapsingaggregate”,
input†: <input>,
within*: <name>,
target*: <name>,
carryovers: [<name>, … , <name>],
aggregations: [
{ref: <name>, expr: <aggexpr> },...
]
}


RunningAggregate (1)
running aggregate operator以一条record作为输入，可以将aggregate 的value添加到input record上面，并且将之作为输出。Aggregation是按照定义的顺序引用于每条record中。Within参数限制了这些Aggregation在每一个segment内部进行操作
{ @id†: <opref>, op: “runningaggregate”,
input†: <input>,
within*: <name>,
aggregations: [
{ref: <name>, expr: <aggexpr> },...
]
}


Flatten (1)
可以根据某条单独的输入record产生一条或者多条输出record，输出record内容是在输入record基础上添加了某些字段的多条record。这种特殊的情况依赖于flatten表达式。Drop参数，定义了再output record中是否保留target expression中定义的fields。

对于标量数据或者Map类型数据：当expression返回一个scalar或者一个map的时候，flatten 会为每一个输入record都返回一个包含了一个附加的field的single record，该field就是在ref参数中定义名称，他的value是和 expression值的一样。
对于Array数据：如果expression返回一个数组，那么flatten会返回原输入record 多个拷贝记录，这些记录包含一个额外的字段，该字段值对应array中一个元素。
{ @id†: <opref>, op: “flatten”,
input†: <input>,
ref: <name>,
expr: <expr>,
drop: <boolean>
}


Sequence (1)
Sequence 是一个语言上的operator用来简化flow的定义。因为大多数的operator都是单输入operator，Sequence 允许定义一个1:1:1：。。。：1的flow，而不需要定义input和定义引用。在第一个operator的后面的每一个operator都必须一个single input输入operator。第一个operator可以是一个source operator，或者是一个single input operator。 Sequence中的最后一个元素可以是一个sink。
Sequence只能在最后一个元素为sink的时候是一个sink。

{ @id: <opref>, op: “sequence”,
input: <input>, do: [
<operator>, <operator>, ... <operator>
]
}



Physical Plan
物理plan（也可以称之为执行计划）可以被execution engin 理解，并能获得预期结果。他是query planner的 输出，是logical plan的转化，一般情况下，physical and execution plans都可以使用json格式来描述的。

Physical Plan Operators

Storage Operators
可以针对不同数据源。
例如：scan-json, scan-hbase, scan-trevni, scan-cassandra, scan-rcfile, scan-pb, scan-pbcol, scanmongo,
scan-text, scan-sequencefile, scan-odbc, scan-jdbc (and corresponding store-* for
each).


Normal Operators
limit, sort, hash-join, merge-join, streaming-aggregate, partial-aggregate, hash-aggregate,
partition, merge-join, union,

评论

2 楼 扬州老鬼 2013-11-06  

Apache Drill provides low latency ad-hoc queries to many different data sources, including nested data. Inspired by Google's Dremel, Drill is designed to scale to 10,000 servers and query petabytes of data in seconds.

1 楼 redstarofsleep 2013-11-06  

我是来点赞的!
话说这是什么高级东东?