Spork: Pig on Spark实现分析

介绍

Spork是Pig on Spark的highly experimental版本,依赖的版本也比较久,如之前文章里所说,目前我把Spork维护在自己的github上:flare-spork
本文分析的是Spork的实现方式和具体内容。

Spark Launcher

在hadoop executionengine包路径下,写了一个Spark启动器,同MapReduceLauncher类似,会在launchPig的时候,把传入的物理执行计划进行翻译。
MR启动器翻译的是MR的操作,以及进一步的MR JobControl。而Spark启动器将物理执行计划的部分物理操作直接翻译成了RDD的操作。
有一个缺点是翻译成RDD算子之后,缺少优化过程,也就是直接物理操作的映射翻译,具体执行逻辑会完全交给Spark DAGScheduler去切分,由TaskScheduler去调度任务。
比如对Pig来说,直到见到Dump/Store,才会触发整个翻译和launch。那么在这一次物理执行计划中,对应到Spark可能是多次任务。

在目前的实现方式下,翻译物理操作交给多个Convertor的实现类来完成,
public interface POConverter<IN, OUT, T extends PhysicalOperator> {
	
    RDD<OUT> convert(List<RDD<IN>> rdd, T physicalOperator) throws IOException;
    
}
抽象类POConvertor提供了convert方法,输入参数中的List<RDD>是本次物理操作的前驱们产生的RDDs,可以认为是会依赖的父RDDs。
这样一次转化结果就是产生nextRDD,而nextRDD是否在spark上真正触发计算,目前来看是不去控制的,也就是上面提到的,一次Pig物理执行计划可能会有Spark执行多次任务。

在使用的时候,以-x spark的方式就可以启动以Spark为backend engine的Pig环境。

下面具体看目前做了哪些PO操作的转化工作,具体怎么转化的。

Load/Store

走的都是NewHadoopRDD路线。

 

Load方面是通过POLoad获得文件路径,pigContext获得必要配置信息,然后交由SparkContext调用newAPIHadoopFile来获得NewHadoopRDD,最后把Tuple2<Text, Tuple>的RDD map成只剩value的RDD<Tuple>。

 

Store方面是先把最近的前驱rdd转会成Key为空Text的Tuple2<Text, Tuple>,然后映射为PairRDDFunctions,借助pigContext生成POStore操作,最后调用RDD的saveAsNewAPIHadoopFile存到HDFS上。


Foreach、Filter、Limit

ForEach里实现一个Iterator[T] => Iterator[T]的方法,把foreach转化为rdd.mapPartitions()方法。

Iterator[T]=> Iterator[T]方法的实现,会依赖原本的POForEach来获得nextTuple和进行一些别的操作,来实现一个新的Iterator。

 

对于hadoop backend的executionengine里的抽象类PhysicalOperator来说,

setInput()和attachInput()方法是放入带处理的tuple数据,

getNextTuple()的时候触发processTuple(),处理对象就是内部的Input Tuple。

 

所以ForEach操作实现Iterator的时候,在readNext()方法里掺入了以上设置Input数据的操作,在返回前调用getNextTuple()返回处理后的结果。

 

POFilter也是通过setInput()和attachInput()以及getNextTuple()来返回处理结果。

所以在实现为RDD操作的时候,把以上步骤包装成一个FilterFunction,传入rdd.filter(Function)处理。

 

POLimit同POFilter是完全一样的。


Distinct

现在RDD已经直接具备distinct(numPartitions: Int)方法了。


这里的distinct实现同rdd里的distinct逻辑是完全一样的。

第一步:把类型为Tuple的rdd映射成为Tuple2<Tuple, Object>,其中value部分是null的;

第二步:进行rdd.reduceByKey(merge_function, parallelism)操作,merge_function对两个value部分的Object不做任何处理,也就是按key reduce且不对value部分处理;

第三步:对第二步的结果进行rdd.map(function, ClassTag)处理,function为得到Tuple2<Tuple, Object>里的._1,即key值:Tuple。



Union

Union是一次求并过程,直接new UnionRDD<Tuple>返回。

由于UnionRDD处理的是Seq<RDD>,所以使用JavaConversions.asScalaBuffer(List<RDD<Tuple>>)进行一下转换再传入。



Sort

Sort过程:

第一步:把Tuple类型的RDD转成Tuple2<Tuple, Object>类型,Object为空

第二步:根据第一步结果,new OrderedRDDFunctions<Tuple, Object,Tuple2<Tuple, Object>>

,其sortByKey方法产出一个排过序的RDD<Tuple2<Tuple, Object>>。OrderedRDDFunctions里的Key类型必须是可排序的,比较器复用的是POSort的mComparator。sortByKey结果返回的是ShuffleRDD,其Partitioner是RangePartitioner,排序之后,每个Partition里存放的都是一个范围内的排过序的值。

第三步:调用rdd.mapPartition(function, xx, xx),function作用为把Iterator<Tuple2<Tuple,Object>>吐成Iterator<Tuple>,即再次取回Key值,此时已有序。


Split

POSplit的处理是直接返回第一个祖先RDD。


LocalRearrange

LocalRearrange -> Global Rearrange -> Package是一同出现的。

Spork: Pig on Spark实现分析

Local rearrange直接依赖

  physicalOperator.setInputs(null);
  physicalOperator.attachInput(t);
  result = physicalOperator.getNextTuple();

三步得到result。返回的Tuple格式为(index, key, value)。

依赖POLocalRearrange本身内部对input tuple的处理。


GlobalRearrange

待处理的Tuple格式是(index, key, value)。最后结果为(key, { values })

 

如果父RDD只有一个:

先进行按key进行一次groupBy,得到结果是Tuple2<Object, Seq<Tuple>>

然后做一次map操作,得到(key, { values })形态的RDD,即Tuple<Object, Iterator>

 

如果父RDD有多个:

让通过rdd的map操作先将Tuple从(index, key, value)转成(key, value)形态,然后把这个rdd集合new成CoGroupRDD,包含一次(Seq) JavaConversions.asScalaBuffer(rddPairs)转化。最后调用CoGroupRDD的map方法,把Tuple2<Object,Seq<Seq<Tuple>>>转化成Tuple<Object, Iterator>,即(key, { values })形态。实际上,CoGroupRDD的map方法内部做的事情,是针对每个Key里的Iterator集合,进行了Iterator之间的合并操作。

 

Package

Package需要把global rearrange处理后的key, Seq<Tuple>进行group。具体的待处理Tuple结构是这样的:(key, Seq<Tuple>:{(index,key, value without key)})

tuple.get(0)是keyTuple,tuple.get(1)是Iterator<Tuple>,最后返回(key, {values}),即Tuple<Object, Iterator>

 


全文完 :)
上一篇:elastic-job失效转移异常


下一篇:Java源码剖析之server2008上拉不出验证码