Hadoop Cascading简介

Hadoop Cascading 是一个开源框架，该框架搭建在Hadoop之上。Cascading提供了一系列易用的数据操作API，如count，group by，join等等共开发人员直接使用，且该框架自动将这些数据操作翻译为底层Hadoop可运行的Map和reduce操作，从而大大加快程序开发速度。

Cascading中使用了一系列特有的概念如flow(数据流)，tap，pipe(数据管道)等。flow可以理解为一个数据流，数据流过时被一一处理。flow的起点是source tap，也就是数据源；终点是sink tap也就是数据输出。tap之间由pipe相连接。下列代码可以容易的理解它们的相互关系：

用例一

String inPath = args[ 0 ];
Tap inTap = new Hfs( new TextDelimited( true, "\t" ), inPath );
String outPath = args[ 1 ];
Tap outTap = new Hfs( new TextDelimited( true, "\t" ), outPath );

Pipe copyPipe = new Pipe( "copy" );
FlowDef flowDef = FlowDef.flowDef()
.setName( "copy" )
.addSource( copyPipe, inTap )
.addTailSink( copyPipe, outTap );

Flow wcFlow = flowConnector.connect( flowDef );
wcFlow.complete();

上面就是一个完整的Cascading小程序，它简单的将文件从源目录拷贝到目标目录。首先我们定义了in tap和out tap，它们都是HDFS中的目录。之后我们定义了一个名为“copy”的pipe。然后定义了一个flow，由代码可知我们给这个flow添加了source tap和sink tap，由“copy”pipe相连。最后启动这个flow。

我们再来回顾一下这个过程。Cascading会形成由一个或多个step构成的物理执行计划，通常我们称之为有向无环图(DAG，它在大数据领域得到了广泛应用，例如在Hive中执行HQL时，实际也是自动产生了DAG并在Hadoop上运行）。当Cascading运行时，数据以键值对的形式流过DAG图。一个step在Hadoop中可能对应多个map或reduce阶段。例如我们知道在Hadoop中做join操作需要编写reducer来完成，那么在cascading中虽然我们简单的使用其提供的join API，但实际上该API会被转换为reducer并在Hadoop上执行。

接下来再看一个复杂一些的例子，如何使用cascading计算word count。word count在大数据领域就像打印“hello world”一样，是经典的入门案例。如果不清楚这个案例，请先参考文章《MapReduce概念》：

/article/9408019.html

用例二

word count的第一步是将输入的字符串按空格分隔，得到一个一个的单词：

Fields token = new Fields( "token" );
Fields text = new Fields( "text" );
RegexSplitGenerator splitter
= new RegexSplitGenerator( token, "[ \\[\\]\\(\\),.]" );

Pipe docPipe = new Each( "token", text, splitter, Fields.RESULTS );

定义一个splitter，并按照splitter中定义的正则表达式将输入字符串分隔成一个一个的单词，也称为token。从docPipe流出的即为一个一个的单词。注意new each()指对每一个输入的键值对都应用该逻辑。

Pipe wcPipe = new Pipe( "wc", docPipe );
wcPipe = new GroupBy( wcPipe, token );
wcPipe = new Every( wcPipe, Fields.ALL, new Count(), Fields.ALL );

定义名为wcPipe的管道，与docPipe相连。在wcPipe中根据token做groupby操作，也就是汇总操作。注意new every()指在一批键值对上做操作，即聚合操作。

FlowDef flowDef = FlowDef.flowDef()
.setName( "wc" )
.addSource( docPipe, docTap )
.addTailSink( wcPipe, wcTap );

最后定义flow，执行flow并将结果写入wc.dot文件：

Flow wcFlow = flowConnector.connect( flowDef );

wcFlow.writeDOT( “dot/wc.dot” );

wcFlow.complete();

总得来说代码量还是比较少的，但个人感觉语义设计的不是很清晰。

用例三

实例的Pipe会更复杂一些，我们来看下面的例子。对于一个输入cvs文件，将每一行数据中的无用字段抛弃掉，一些字段做转换，并新增两个字段。

private static Pipe createWarrantyPipe(String fields) {
Pipe docPipe;
Fields scrubArguments = new Fields("#ORDER_BUSINESS_UNIT_ID");

//输入文件是一个cvs，包含很多字段。cvs包含字段的具体含义在配置文件中。下的代码实际是读出配置文件中定义的cvs字段，作为Pipe的输入字段。
List<String> items = new ArrayList<String>(Arrays.asList(fields.split(",")));
for (String s : items) {
Fields scrubArgument = new Fields(s);
scrubArguments = scrubArguments.append(scrubArgument);
}

// 这里定义了Pipe的输出字段，且新增了两个原cvs中没有的新字段。
Fields outputScrubArguments = scrubArguments.append(new Fields("PROCESSED_FLG")).append(new Fields("STAGE_DT")).append(new Fields("FileName"));

//下面的Pipe定义了输入字段，输出字段，且转换关系由ScrubWarFunction指定。
docPipe = new Pipe("copy");
docPipe = new Each(docPipe, scrubArguments, new ScrubWarFunction(outputScrubArguments), Fields.RESULTS);
return docPipe;
}

完整的ScrubWarFunction代码如下：

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Calendar;

import org.apache.hadoop.mapred.FileSplit;

import cascading.flow.FlowProcess;
import cascading.flow.hadoop.HadoopFlowProcess;
import cascading.operation.BaseOperation;
import cascading.operation.Function;
import cascading.operation.FunctionCall;
import cascading.tap.hadoop.io.MultiInputSplit;
import cascading.tuple.Fields;
import cascading.tuple.Tuple;
import cascading.tuple.TupleEntry;

public class ScrubWarFunction extends BaseOperation implements Function {

public ScrubWarFunction(Fields fieldDeclaration) {
super(2, fieldDeclaration);
}

public void operate(FlowProcess flowProcess, FunctionCall functionCall) {
TupleEntry inputargument = functionCall.getArguments();

HadoopFlowProcess hfp = (HadoopFlowProcess) flowProcess;
MultiInputSplit mis = (MultiInputSplit) hfp.getReporter().getInputSplit();
FileSplit fs = (FileSplit) mis.getWrappedInputSplit();
String FileName = fs.getPath().getName();

DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd HH:mm:ss");
Calendar cal = Calendar.getInstance();

String STAGE_DT = dateFormat.format(cal.getTime());
String PROCESSED_FLAG = "0";

Tuple result = new Tuple();

// All the fields be transformed and other fields to be added to the result.
for (int i = 0; i < inputargument.size(); i++) {
result.add(inputargument.getString(i));
}

// Add additional fields.
result.add(STAGE_DT);
result.add(PROCESSED_FLAG);
result.add(FileName);

functionCall.getOutputCollector().add(result);
}

}