mapreduce的典型案例（hadoop的mapreduce常见算法案例有几种）

本文目录

• hadoop的mapreduce常见算法案例有几种
• 大数据时代的案例分析
• hive 什么样的operator会导致生成一个mapreduce任务

hadoop的mapreduce常见算法案例有几种

• wordcount是最常见的

• 基本MapReduce模式
  
  计数与求和
  问题陈述:
  有许多文档，每个文档都有一些字段组成。需要计算出每个字段在所有文档中的出现次数或者这些字段的其他什么统计值。例如，给定一个log文件，其中的每条记录都包含一个响应时间，需要计算出平均响应时间。
  解决方案:
  让我们先从简单的例子入手。在下面的代码片段里，Mapper每遇到指定词就把频次记1，Reducer一个个遍历这些词的集合然后把他们的频次加和。
  1 class Mapper
  2 method Map(docid id, doc d)
  3 for all term t in doc d do
  4 Emit(term t, count 1)
  5
  6 class Reducer
  7 method Reduce(term t, counts )
  8 sum = 0
  9 for all count c in do
  10 sum = sum + c
  11 Emit(term t, count sum)
  这种方法的缺点显而易见，Mapper提交了太多无意义的计数。它完全可以通过先对每个文档中的词进行计数从而减少传递给Reducer的数据量:
  1 class Mapper
  2 method Map(docid id, doc d)
  3 H = new AssociativeArray
  4 for all term t in doc d do
  5 H{t} = H{t} + 1
  6 for all term t in H do
  7 Emit(term t, count H{t})
  如果要累计计数的的不只是单个文档中的内容，还包括了一个Mapper节点处理的所有文档，那就要用到Combiner了:
  1 class Mapper
  2 method Map(docid id, doc d)
  3 for all term t in doc d do
  4 Emit(term t, count 1)
  5
  6 class Combiner
  7 method Combine(term t, )
  8 sum = 0
  9 for all count c in do
  10 sum = sum + c
  11 Emit(term t, count sum)
  12
  13 class Reducer
  14 method Reduce(term t, counts )
  15 sum = 0
  16 for all count c in do
  17 sum = sum + c
  18 Emit(term t, count sum)
  应用：Log 分析, 数据查询
  
  整理归类
  
  问题陈述:
  有一系列条目，每个条目都有几个属性，要把具有同一属性值的条目都保存在一个文件里，或者把条目按照属性值分组。 最典型的应用是倒排索引。
  解决方案：
  解决方案很简单。 在 Mapper 中以每个条目的所需属性值作为 key，其本身作为值传递给 Reducer。 Reducer 取得按照属性值分组的条目，然后可以处理或者保存。如果是在构建倒排索引，那么 每个条目相当于一个词而属性值就是词所在的文档ID。
  应用：倒排索引， ETL
  过滤 (文本查找)，解析和校验
  问题陈述:
  假设有很多条记录，需要从其中找出满足某个条件的所有记录，或者将每条记录传换成另外一种形式（转换操作相对于各条记录独立，即对一条记录的操作与其他记录无关）。像文本解析、特定值抽取、格式转换等都属于后一种用例。
  解决方案:
  非常简单，在Mapper 里逐条进行操作，输出需要的值或转换后的形式。
  应用：日志分析，数据查询，ETL，数据校验
  
  分布式任务执行
  
  问题陈述:
  大型计算可以分解为多个部分分别进行然后合并各个计算的结果以获得最终结果。
  解决方案: 将数据切分成多份作为每个 Mapper 的输入，每个Mapper处理一份数据，执行同样的运算，产生结果，Reducer把多个Mapper的结果组合成一个。
  案例研究： 数字通信系统模拟
  像 WiMAX 这样的数字通信模拟软件通过系统模型来传输大量的随机数据，然后计算传输中的错误几率。 每个 Mapper 处理样本 1/N 的数据，计算出这部分数据的错误率，然后在 Reducer 里计算平均错误率。
  应用：工程模拟，数字分析，性能测试
  排序
  问题陈述:
  有许多条记录，需要按照某种规则将所有记录排序或是按照顺序来处理记录。
  解决方案: 简单排序很好办 – Mappers 将待排序的属性值为键，整条记录为值输出。 不过实际应用中的排序要更加巧妙一点， 这就是它之所以被称为MapReduce 核心的原因（“核心”是说排序？因为证明Hadoop计算能力的实验是大数据排序？还是说Hadoop的处理过程中对key排序的环节？）。在实践中，常用组合键来实现二次排序和分组。
  MapReduce 最初只能够对键排序， 但是也有技术利用可以利用Hadoop 的特性来实现按值排序。想了解的话可以看这篇博客。
  按照BigTable的概念，使用 MapReduce来对最初数据而非中间数据排序，也即保持数据的有序状态更有好处，必须注意这一点。换句话说，在数据插入时排序一次要比在每次查询数据的时候排序更高效。
  应用：ETL，数据分析
  
  非基本 MapReduce 模式
  
  迭代消息传递 (图处理)
  
  问题陈述：
  假设一个实体网络，实体之间存在着关系。 需要按照与它比邻的其他实体的属性计算出一个状态。这个状态可以表现为它和其它节点之间的距离， 存在特定属性的邻接点的迹象， 邻域密度特征等等。
  解决方案：
  网络存储为系列节点的结合，每个节点包含有其所有邻接点ID的列表。按照这个概念，MapReduce 迭代进行，每次迭代中每个节点都发消息给它的邻接点。邻接点根据接收到的信息更新自己的状态。当满足了某些条件的时候迭代停止，如达到了最大迭代次数（网络半径）或两次连续的迭代几乎没有状态改变。从技术上来看，Mapper 以每个邻接点的ID为键发出信息，所有的信息都会按照接受节点分组，reducer 就能够重算各节点的状态然后更新那些状态改变了的节点。下面展示了这个算法：
  1 class Mapper
  2 method Map(id n, object N)
  3 Emit(id n, object N)
  4 for all id m in N.OutgoingRelations do
  5 Emit(id m, message getMessage(N))
  6
  7 class Reducer
  8 method Reduce(id m, )
  9 M = null
  10 messages =
  11 for all s in do
  12 if IsObject(s) then
  13 M = s
  14 else // s is a message
  15 messages.add(s)
  16 M.State = calculateState(messages)
  17 Emit(id m, item M)
  一个节点的状态可以迅速的沿着网络传全网，那些被感染了的节点又去感染它们的邻居，整个过程就像下面的图示一样：
  案例研究： 沿分类树的有效性传递
  问题陈述：
  这个问题来自于真实的电子商务应用。将各种货物分类，这些类别可以组成一个树形结构，比较大的分类（像男人、女人、儿童）可以再分出小分类（像男裤或女装），直到不能再分为止（像男式蓝色牛仔裤）。这些不能再分的基层类别可以是有效（这个类别包含有货品）或者已无效的（没有属于这个分类的货品）。如果一个分类至少含有一个有效的子分类那么认为这个分类也是有效的。我们需要在已知一些基层分类有效的情况下找出分类树上所有有效的分类。
  解决方案：
  这个问题可以用上一节提到的框架来解决。我们咋下面定义了名为 getMessage和 calculateState 的方法：
  1 class N
  2 State in {True = 2, False = 1, null = 0},
  3 initialized 1 or 2 for end-of-line categories, 0 otherwise
  4 method getMessage(object N)
  5 return N.State
  6 method calculateState(state s, data )
  7 return max( )
  案例研究：广度优先搜索
  问题陈述：需要计算出一个图结构中某一个节点到其它所有节点的距离。
  解决方案： Source源节点给所有邻接点发出值为0的信号，邻接点把收到的信号再转发给自己的邻接点，每转发一次就对信号值加1：
  1 class N
  2 State is distance,
  3 initialized 0 for source node, INFINITY for all other nodes
  4 method getMessage(N)
  5 return N.State + 1
  6 method calculateState(state s, data )
  7 min( )
  案例研究：网页排名和 Mapper 端数据聚合
  这个算法由Google提出，使用权威的PageRank算法，通过连接到一个网页的其他网页来计算网页的相关性。真实算法是相当复杂的，但是核心思想是权重可以传播，也即通过一个节点的各联接节点的权重的均值来计算节点自身的权重。
  1 class N
  2 State is PageRank
  3 method getMessage(object N)
  4 return N.State / N.OutgoingRelations.size()
  5 method calculateState(state s, data )
  6 return ( sum() )
  要指出的是上面用一个数值来作为评分实际上是一种简化，在实际情况下，我们需要在Mapper端来进行聚合计算得出这个值。下面的代码片段展示了这个改变后的逻辑 （针对于 PageRank 算法）：
  1 class Mapper
  2 method Initialize
  3 H = new AssociativeArray
  4 method Map(id n, object N)
  5 p = N.PageRank / N.OutgoingRelations.size()
  6 Emit(id n, object N)
  7 for all id m in N.OutgoingRelations do
  8 H{m} = H{m} + p
  9 method Close
  10 for all id n in H do
  11 Emit(id n, value H{n})
  12
  13 class Reducer
  14 method Reduce(id m, )
  15 M = null
  16 p = 0
  17 for all s in do
  18 if IsObject(s) then
  19 M = s
  20 else
  21 p = p + s
  22 M.PageRank = p
  23 Emit(id m, item M)
  应用：图分析，网页索引
  
  值去重 （对唯一项计数）
  问题陈述: 记录包含值域F和值域 G，要分别统计相同G值的记录中不同的F值的数目 (相当于按照 G分组).
  这个问题可以推而广之应用于分面搜索（某些电子商务网站称之为Narrow Search）
  Record 1: F=1, G={a, b}
  Record 2: F=2, G={a, d, e}
  Record 3: F=1, G={b}
  Record 4: F=3, G={a, b}
  Result:
  a -》 3 // F=1, F=2, F=3
  b -》 2 // F=1, F=3
  d -》 1 // F=2
  e -》 1 // F=2
  解决方案 I:
  第一种方法是分两个阶段来解决这个问题。第一阶段在Mapper中使用F和G组成一个复合值对，然后在Reducer中输出每个值对，目的是为了保证F值的唯一性。在第二阶段，再将值对按照G值来分组计算每组中的条目数。
  第一阶段：
  1 class Mapper
  2 method Map(null, record )
  3 for all category g in
  4 Emit(record , count 1)
  5
  6 class Reducer
  7 method Reduce(record )
  8 Emit(record , null )
  第二阶段：
  1 class Mapper
  2 method Map(record , null)
  3 Emit(value g, count 1)
  4
  5 class Reducer
  6 method Reduce(value g, counts )
  7 Emit(value g, sum( ) )
  解决方案 II:
  第二种方法只需要一次MapReduce 即可实现，但扩展性不强。算法很简单-Mapper 输出值和分类，在Reducer里为每个值对应的分类去重然后给每个所属的分类计数加1，最后再在Reducer结束后将所有计数加和。这种方法适用于只有有限个分类，而且拥有相同F值的记录不是很多的情况。例如网络日志处理和用户分类，用户的总数很多，但是每个用户的事件是有限的，以此分类得到的类别也是有限的。值得一提的是在这种模式下可以在数据传输到Reducer之前使用Combiner来去除分类的重复值。
  1 class Mapper
  2 method Map(null, record )
  3 for all category g in
  4 Emit(value f, category g)
  5
  6 class Reducer
  7 method Initialize
  8 H = new AssociativeArray : category -》 count
  9 method Reduce(value f, categories )
  10 )
  11 for all category g in
  12 H{g} = H{g} + 1
  13 method Close
  14 for all category g in H do
  15 Emit(category g, count H{g})
  应用：日志分析，用户计数
  互相关
  问题陈述：有多个各由若干项构成的组，计算项两两共同出现于一个组中的次数。假如项数是N，那么应该计算N*N。
  这种情况常见于文本分析（条目是单词而元组是句子），市场分析（购买了此物的客户还可能购买什么）。如果N*N小到可以容纳于一台机器的内存，实现起来就比较简单了。
  配对法
  第一种方法是在Mapper中给所有条目配对，然后在Reducer中将同一条目对的计数加和。但这种做法也有缺点：
  使用 combiners 带来的的好处有限，因为很可能所有项对都是唯一的
  不能有效利用内存
  1 class Mapper
  2 method Map(null, items )
  3 for all item i in
  4 for all item j in
  5 Emit(pair , count 1)
  6
  7 class Reducer
  8 method Reduce(pair )
  9 s = sum()
  10 Emit(pair, count s)
  Stripes Approach（条方法？不知道这个名字怎么理解）
  第二种方法是将数据按照pair中的第一项来分组，并维护一个关联数组，数组中存储的是所有关联项的计数。The second approach is to group data by the first item in pair and maintain an associative array (“stripe”) where counters for all adjacent items are accumulated. Reducer receives all stripes for leading item i, merges them, and emits the same result as in the Pairs approach.
  中间结果的键数量相对较少，因此减少了排序消耗。
  可以有效利用 combiners。
  可在内存中执行，不过如果没有正确执行的话也会带来问题。
  实现起来比较复杂。
  一般来说， “stripes” 比 “pairs” 更快
  1 class Mapper
  2 method Map(null, items )
  3 for all item i in
  4 H = new AssociativeArray : item -》 counter
  5 for all item j in
  6 H{j} = H{j} + 1
  7 Emit(item i, stripe H)
  8
  9 class Reducer
  10 method Reduce(item i, stripes )
  11 H = new AssociativeArray : item -》 counter
  12 H = merge-sum( )
  13 for all item j in H.keys()
  14 Emit(pair , H{j})
  应用：文本分析，市场分析
  参考资料：Lin J. Dyer C. Hirst G. Data Intensive Processing MapReduce
  用MapReduce 表达关系模式
  在这部分我们会讨论一下怎么使用MapReduce来进行主要的关系操作。
  筛选（Selection）
  1 class Mapper
  2 method Map(rowkey key, tuple t)
  3 if t satisfies the predicate
  4 Emit(tuple t, null)
  投影（Projection）
  投影只比筛选稍微复杂一点，在这种情况下我们可以用Reducer来消除可能的重复值。
  1 class Mapper
  2 method Map(rowkey key, tuple t)
  3 tuple g = project(t) // extract required fields to tuple g
  4 Emit(tuple g, null)
  5
  6 class Reducer

大数据时代的案例分析

个案一
你开心他就买你焦虑他就抛
华尔街“德温特资本市场”公司首席执行官保罗·霍廷每天的工作之一，就是利用电脑程序分析全球3.4亿微博账户的留言，进而判断民众情绪，再以“1”到“50”进行打分。根据打分结果，霍廷再决定如何处理手中数以百万美元计的股票。
霍廷的判断原则很简单：如果所有人似乎都高兴，那就买入；如果大家的焦虑情绪上升，那就抛售。
这一招收效显著——当年第一季度，霍廷的公司获得了7%的收益率。
个案二
国际商用机器公司（IBM）估测，这些“数据”值钱的地方主要在于时效。对于片刻便能定输赢的华尔街，这一时效至关重要。曾经，华尔街2%的企业搜集微博等平台的“非正式”数据；如今，接近半数企业采用了这种手段。
●“社会流动”创业公司在“大数据”行业生机勃勃，和微博推特是合作伙伴。它分析数据，告诉广告商什么是正确的时间，谁是正确的用户，什么是应该发表的正确内容，备受广告商热爱。
●通过乔希·詹姆斯的Omniture（著名的网页流量分析工具）公司，你可以知道有多少人访问你的网站，以及他们呆了多长时间——这些数据对于任何企业来说都至关重要。詹姆斯把公司卖掉，进账18亿美元。
●微软专家吉拉德喜欢把这些“大数据”结果可视化：他把客户请到办公室，将包含这些公司的数据图谱展现出来——有些是普通的时间轴，有些像蒲公英，有些则是铺满整个画面的泡泡，泡泡中显示这些客户的粉丝正在谈论什么话题。
●“脸谱”数据分析师杰弗逊的工作就是搭建数据分析模型，弄清楚用户点击广告的动机和方式。
处理和分析工具
用于分析大数据的工具主要有开源与商用两个生态圈。
开源大数据生态圈：
1、Hadoop HDFS、HadoopMapReduce, HBase、Hive 渐次诞生，早期Hadoop生态圈逐步形成。
2、. Hypertable是另类。它存在于Hadoop生态圈之外，但也曾经有一些用户。
3、NoSQL，membase、MongoDb
商用大数据生态圈：
1、一体机数据库/数据仓库：IBM PureData(Netezza), OracleExadata, SAP Hana等等。
2、数据仓库：TeradataAsterData, EMC GreenPlum, HPVertica 等等。
3、数据集市：QlikView、 Tableau 、 以及国内的Yonghong Data Mart 。

hive 什么样的operator会导致生成一个mapreduce任务

Hive中在做多表关联时,由于Hive的SQL优化引擎还不够强大,表的关联顺序不同往往导致产生不同数量的MapReduce作业数。这时就需要通过分析执行计划对SQL进行调整，以获得最少的MapReduce作业数。举一个例子（案例基于Hive 0.6.0）：

create table ljn1(
k1 bigint,
k2 String,
v1 int
);

create table ljn2(
k1 bigint,
v2 int
);

create table ljn3(
k1 bigint,
v3 int
);

create table ljn4(
k1 bigint,
v4 int
);

create table ljn5(
k1 bigint,
v5 int
);

create table ljn6(
k2 string,
v6 int
);

然后看一下下面这个SQL的执行计划:
explain
select a.v1
from
ljn1 a
left outer join ljn2 b on (a.k1 = b.k1)
left outer join ljn3 c on (a.k1 = c.k1)
left outer join ljn4 d on (a.k1 = d.k1)
left outer join ljn6 e on (a.k2 = e.k2)
left outer join ljn5 f on (a.k1 = f.k1);
STAGE DEPENDENCIES:
Stage-5 is a root stage
Stage-1 depends on stages: Stage-5
Stage-2 depends on stages: Stage-1
Stage-0 is a root stage

STAGE PLANS:
Stage: Stage-5
Map Reduce
Alias -》 Map Operator Tree:
a
TableScan
alias: a
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 0
value expressions:
expr: k1
type: bigint
expr: k2
type: string
expr: v1
type: int
b
TableScan
alias: b
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 1
Reduce Operator Tree:
Join Operator
condition map:
Left Outer Join0 to 1
condition expressions:
0 {VALUE._col0} {VALUE._col1} {VALUE._col2}
1
handleSkewJoin: false
outputColumnNames: _col0, _col1, _col2
File Output Operator
compressed: true
GlobalTableId: 0
table:
input format: org.apache.hadoop.mapred.SequenceFileInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat

Stage: Stage-1
Map Reduce
Alias -》 Map Operator Tree:
$INTNAME
Reduce Output Operator
key expressions:
expr: _col0
type: bigint
sort order: +
Map-reduce partition columns:
expr: _col0
type: bigint
tag: 0
value expressions:
expr: _col1
type: string
expr: _col2
type: int
c
TableScan
alias: c
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 1
d
TableScan
alias: d
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 2
f
TableScan
alias: f
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 3
Reduce Operator Tree:
Join Operator
condition map:
Left Outer Join0 to 1
Left Outer Join0 to 2
Left Outer Join0 to 3
condition expressions:
0 {VALUE._col3} {VALUE._col4}
1
2
3
handleSkewJoin: false
outputColumnNames: _col3, _col4
File Output Operator
compressed: true
GlobalTableId: 0
table:
input format: org.apache.hadoop.mapred.SequenceFileInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat

Stage: Stage-2
Map Reduce
Alias -》 Map Operator Tree:
$INTNAME
Reduce Output Operator
key expressions:
expr: _col3
type: string
sort order: +
Map-reduce partition columns:
expr: _col3
type: string
tag: 0
value expressions:
expr: _col4
type: int
e
TableScan
alias: e
Reduce Output Operator
key expressions:
expr: k2
type: string
sort order: +
Map-reduce partition columns:
expr: k2
type: string
tag: 1
Reduce Operator Tree:
Join Operator
condition map:
Left Outer Join0 to 1
condition expressions:
0 {VALUE._col10}
1
handleSkewJoin: false
outputColumnNames: _col10
Select Operator
expressions:
expr: _col10
type: int
outputColumnNames: _col0
File Output Operator
compressed: true
GlobalTableId: 0
table:
input format: org.apache.hadoop.mapred.TextInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat

Stage: Stage-0
Fetch Operator
limit: -1

常规来讲,这个SQL非常简单,a表是主表，与其他表左外关联用到了k1和k2两个关联键，使用两个MapReduce作业完全可以搞定。但是这个SQL的执行计划却给出了3个作业：（Stage-0用做数据的最终展示，该作业可以忽略不计）第1个作业（Stage-5）是a表与b表关联；第2个作业（Stage-1）是第1个作业的中间结果再与c、d、f三表关联；第3个作业（Stage-2）是第2个作业的中间结果再与e表关联。
有点搞不懂了吧，第1和第2个作业明明可以合并在一起来完成的呀！其实我也搞不懂，从执行计划中看不出原由。而且如果这个SQL去掉c或者e其中的一个关联表，第1和第2个作业就可以合并在一起！很奇妙，我没有深入探究，应该是Hive的规则优化器还不够完美。
总之，遇到这种多表关联的情况一定要记得看一下执行计划，看看Hive是不是生成了多余的作业。如果Hive真的犯傻生成了多余的作业，就要尝试改变一下SQL的写法。通常是将关联键相同的表放在一起，如果还不行就再引入子查询。例如上面这个例子改为如下SQL就可以只生成2个作业了：
explain
select t.v1
from
(
select a.k2,a.v1
from
ljn1 a
left outer join ljn2 b on (a.k1 = b.k1)
left outer join ljn3 c on (a.k1 = c.k1)
left outer join ljn4 d on (a.k1 = d.k1)
left outer join ljn5 f on (a.k1 = f.k1)
) t
left outer join ljn6 e on (t.k2 = e.k2)
;
STAGE DEPENDENCIES:
Stage-1 is a root stage
Stage-2 depends on stages: Stage-1
Stage-0 is a root stage

STAGE PLANS:
Stage: Stage-1
Map Reduce
Alias -》 Map Operator Tree:
t:a
TableScan
alias: a
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 0
value expressions:
expr: k2
type: string
expr: v1
type: int
t:b
TableScan
alias: b
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 1
t:c
TableScan
alias: c
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 2
t:d
TableScan
alias: d
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 3
t:f
TableScan
alias: f
Reduce Output Operator
key expressions:
expr: k1
type: bigint
sort order: +
Map-reduce partition columns:
expr: k1
type: bigint
tag: 4
Reduce Operator Tree:
Join Operator
condition map:
Left Outer Join0 to 1
Left Outer Join0 to 2
Left Outer Join0 to 3
Left Outer Join0 to 4
condition expressions:
0 {VALUE._col1} {VALUE._col2}
1
2
3
4
handleSkewJoin: false
outputColumnNames: _col1, _col2
Select Operator
expressions:
expr: _col1
type: string
expr: _col2
type: int
outputColumnNames: _col0, _col1
File Output Operator
compressed: true
GlobalTableId: 0
table:
input format: org.apache.hadoop.mapred.SequenceFileInputFormat
output format: org.apache.hadoop.hive.ql.io.HiveSequenceFileOutputFormat

Stage: Stage-2
Map Reduce
Alias -》 Map Operator Tree:
$INTNAME
Reduce Output Operator
key expressions:
expr: _col0
type: string
sort order: +
Map-reduce partition columns:
expr: _col0
type: string
tag: 0
value expressions:
expr: _col1
type: int
e
TableScan
alias: e
Reduce Output Operator
key expressions:
expr: k2