About Cascading?

발번역 주의 요망, 내용 중 오류 발견시 연락 요망 = =

# What is Cascading?

Cascading은 분산컴퓨팅 클러스터 혹은 단일컴퓨터 노드 환경에서 질의를 정의&공유 하는 등의 작업을 하고

data processing workflow를 실행하는 data processing API이다.

단일컴퓨터 노드 환경(local mode)는 test코드와 workflow를 cluster에 배포 전에 효율적으로 테스트 할 수 있다. 

분산컴퓨팅 환경에서는 Apache Hadoop plaform에서 이용 할 수 있다. 

이를 이용해 쉽게 Hadoop Application을 개발 하고 Job을 생성 하고 스케쥴링 할 수 있다. 

쉽게 설명 하자면 

HDFS 등에 분산되어있는 데이터를 Cascading을 이용하여 쉽게 추출/정제를 하고 work flow를 생성 할 수 있다. 

User는 Map/Reduce를 직접 구현하지 않아도 Casading의 Each, GroupBy, Aggregator, Filter 등을 이용하여 Hadoop Job에서 Mapper와 Reducer를 작동 시킬 수 있다.  

또한 기 구현 된 Operation 중에 원하는 Operation이 없다면 Interface를 상속받아 구현 할 수 있다.

# Example - word count 

source = Input 데이터의 위치와 원하는 형태의 스키마로 Tap 생성 

sink =  저장하려는 스키마와, OuputPath, 싱크모드를 지정한 Tap 생성 

assembly= 파이프를 만들어서 wordcount로 명명하고,

RegularExpression Generator로 각각의 라인을 단어로 쪼개서 word라고 명명된 필드에 넣음 

word로 groupBy하여 count함

Scheme sinkScheme = new TextLine( new Fields( "word", "count" ) ); Tap sink = new Hfs( sinkScheme, outputPath, SinkMode.REPLACE ); // the 'head' of the pipe assembly Pipe assembly = new Pipe( "wordcount" ); // For each input Tuple // parse out each word into a new Tuple with the field name "word" // regular expressions are optional in Cascading String regex = "(?<!\\pL)(?=\\pL)[^ ]*(?<=\\pL)(?!\\pL)"; Function function = new RegexGenerator( new Fields( "word" ), regex ); assembly = new Each( assembly, new Fields( "line" ), function ); // group the Tuple stream by the "word" value assembly = new GroupBy( assembly, new Fields( "word" ) ); // For every Tuple group // count the number of occurrences of "word" and store result in // a field named "count" Aggregator count = new Count( new Fields( "count" ) ); assembly = new Every( assembly, count ); // initialize app properties, tell Hadoop which jar file to use Properties properties = new Properties(); AppProps.setApplicationJarClass( properties, Main.class ); // plan a new Flow from the assembly using the source and sink Taps // with the above properties FlowConnector flowConnector = new HadoopFlowConnector( properties ); Flow flow = flowConnector.connect( "word-count", source, sink, assembly ); // execute the flow, block until complete flow.complete();

# Comparison of pipe types

Pipe type	Purpose	Input	Output
Pipe	instantiate a pipe; create or name a branch	name	a (named) pipe
SubAssembly	create nested subassemblies
Each	apply a filter or function, or branch a stream	tuple stream (grouped or not)	a tuple stream, optionally filtered or transformed
Merge	merge two or more streams with identical fields	two or more tuple streams	a tuple stream, unsorted
GroupBy	sort/group on field values; optionally merge two or more streams with identical fields	one or more tuple streams with identical fields	a single tuple stream, grouped on key field(s) with optional secondary sort
Every	apply aggregator or buffer operation	grouped tuple stream	a tuple stream plus new fields with operation results
CoGroup	join 1 or more streams on matching field values	one or more tuple streams	a single tuple stream, joined on key field(s)
HashJoin	join 1 or more streams on matching field values	one or more tuple streams	a tuple stream in arbitrary order

# 결론

M/R 코드는 딱 한번 짜봤는데 간단한 M/R조차 개발하는게 쉽지 않았다. 

Cascading은 쉽게 work flow를 생성하여 ETL 업무를 하는데 적격이다. 

또한 Cascading API는 Java뿐아니라 Scala, Clojure, Groove, JRuby, Jython 등에서도 사용 할 수 있다.

그러나 work flow가 필요 없는 간단한 질의의 경우 Impala나 Shark, Hive, Pig 등이 더 유리 할 수 있다. 

개발 전엔 항상 데이터 크기, 모양, 해상도, 작업 복잡도, 실시간이냐 아니냐 등

 여러가지를 잘 따져보고 자신의 상황에 맞는 것을 사용해야됨!!