| ---++ Contents |
| * <a href="#Onboarding Steps">Onboarding Steps</a> |
| * <a href="#Sample Pipeline">Sample Pipeline</a> |
| * [[HiveIntegration][Hive Examples]] |
| |
| ---+++ Onboarding Steps |
| * Create cluster definition for the cluster, specifying name node, job tracker, workflow engine endpoint, messaging endpoint. Refer to [[EntitySpecification][cluster definition]] for details. |
| * Create Feed definitions for each of the input and output specifying frequency, data path, ownership. Refer to [[EntitySpecification][feed definition]] for details. |
| * Create Process definition for your job. Process defines configuration for the workflow job. Important attributes are frequency, inputs/outputs and workflow path. Refer to [[EntitySpecification][process definition]] for process details. |
| * Define workflow for your job using the workflow engine(only oozie is supported as of now). Refer [[http://oozie.apache.org/docs/3.1.3-incubating/WorkflowFunctionalSpec.html][Oozie Workflow Specification]]. The libraries required for the workflow should be available in lib folder in workflow path. |
| * Set-up workflow definition, libraries and referenced scripts on hadoop. |
| * Submit cluster definition |
| * Submit and schedule feed and process definitions |
| |
| |
| ---+++ Sample Pipeline |
| ---++++ Cluster |
| Cluster definition that contains end points for name node, job tracker, oozie and jms server: |
| The cluster locations MUST be created prior to submitting a cluster entity to Falcon. |
| *staging* must have 777 permissions and the parent dirs must have execute permissions |
| *working* must have 755 permissions and the parent dirs must have execute permissions |
| |
| <verbatim> |
| <?xml version="1.0"?> |
| <!-- |
| Cluster configuration |
| --> |
| <cluster colo="ua2" description="" name="corp" xmlns="uri:falcon:cluster:0.1" |
| xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> |
| <interfaces> |
| <interface type="readonly" endpoint="hftp://name-node.com:50070" version="2.5.0" /> |
| |
| <interface type="write" endpoint="hdfs://name-node.com:54310" version="2.5.0" /> |
| |
| <interface type="execute" endpoint="job-tracker:54311" version="2.5.0" /> |
| |
| <interface type="workflow" endpoint="http://oozie.com:11000/oozie/" version="4.0.1" /> |
| |
| <interface type="messaging" endpoint="tcp://jms-server.com:61616?daemon=true" version="5.1.6" /> |
| </interfaces> |
| |
| <locations> |
| <location name="staging" path="/projects/falcon/staging" /> |
| <location name="temp" path="/tmp" /> |
| <location name="working" path="/projects/falcon/working" /> |
| </locations> |
| </cluster> |
| </verbatim> |
| |
| ---++++ Input Feed |
| Hourly feed that defines feed path, frequency, ownership and validity: |
| <verbatim> |
| <?xml version="1.0" encoding="UTF-8"?> |
| <!-- |
| Hourly sample input data |
| --> |
| |
| <feed description="sample input data" name="SampleInput" xmlns="uri:falcon:feed:0.1" |
| xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> |
| <groups>group</groups> |
| |
| <frequency>hours(1)</frequency> |
| |
| <late-arrival cut-off="hours(6)" /> |
| |
| <clusters> |
| <cluster name="corp" type="source"> |
| <validity start="2009-01-01T00:00Z" end="2099-12-31T00:00Z" timezone="UTC" /> |
| <retention limit="months(24)" action="delete" /> |
| </cluster> |
| </clusters> |
| |
| <locations> |
| <location type="data" path="/projects/bootcamp/data/${YEAR}-${MONTH}-${DAY}-${HOUR}/SampleInput" /> |
| <location type="stats" path="/projects/bootcamp/stats/SampleInput" /> |
| <location type="meta" path="/projects/bootcamp/meta/SampleInput" /> |
| </locations> |
| |
| <ACL owner="suser" group="users" permission="0755" /> |
| |
| <schema location="/none" provider="none" /> |
| </feed> |
| </verbatim> |
| |
| ---++++ Output Feed |
| Daily feed that defines feed path, frequency, ownership and validity: |
| <verbatim> |
| <?xml version="1.0" encoding="UTF-8"?> |
| <!-- |
| Daily sample output data |
| --> |
| |
| <feed description="sample output data" name="SampleOutput" xmlns="uri:falcon:feed:0.1" |
| xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"> |
| <groups>group</groups> |
| |
| <frequency>days(1)</frequency> |
| |
| <late-arrival cut-off="hours(6)" /> |
| |
| <clusters> |
| <cluster name="corp" type="source"> |
| <validity start="2009-01-01T00:00Z" end="2099-12-31T00:00Z" timezone="UTC" /> |
| <retention limit="months(24)" action="delete" /> |
| </cluster> |
| </clusters> |
| |
| <locations> |
| <location type="data" path="/projects/bootcamp/output/${YEAR}-${MONTH}-${DAY}/SampleOutput" /> |
| <location type="stats" path="/projects/bootcamp/stats/SampleOutput" /> |
| <location type="meta" path="/projects/bootcamp/meta/SampleOutput" /> |
| </locations> |
| |
| <ACL owner="suser" group="users" permission="0755" /> |
| |
| <schema location="/none" provider="none" /> |
| </feed> |
| </verbatim> |
| |
| ---++++ Process |
| Sample process which runs daily at 6th hour on corp cluster. It takes one input - !SampleInput for the previous day(24 instances). It generates one output - !SampleOutput for previous day. The workflow is defined at /projects/bootcamp/workflow/workflow.xml. Any libraries available for the workflow should be at /projects/bootcamp/workflow/lib. The process also defines properties queueName, ssh.host, and fileTimestamp which are passed to the workflow. In addition, Falcon exposes the following properties to the workflow: nameNode, jobTracker(hadoop properties), input and output(Input/Output properties). |
| |
| <verbatim> |
| <?xml version="1.0" encoding="UTF-8"?> |
| <!-- |
| Daily sample process. Runs at 6th hour every day. Input - last day's hourly data. Generates output for yesterday |
| --> |
| <process name="SampleProcess"> |
| <cluster name="corp" /> |
| |
| <frequency>days(1)</frequency> |
| |
| <validity start="2012-04-03T06:00Z" end="2022-12-30T00:00Z" timezone="UTC" /> |
| |
| <inputs> |
| <input name="input" feed="SampleInput" start="yesterday(0,0)" end="today(-1,0)" /> |
| </inputs> |
| |
| <outputs> |
| <output name="output" feed="SampleOutput" instance="yesterday(0,0)" /> |
| </outputs> |
| |
| <properties> |
| <property name="queueName" value="reports" /> |
| <property name="ssh.host" value="host.com" /> |
| <property name="fileTimestamp" value="${coord:formatTime(coord:nominalTime(), 'yyyy-MM-dd')}" /> |
| </properties> |
| |
| <workflow engine="oozie" path="/projects/bootcamp/workflow" /> |
| |
| <retry policy="periodic" delay="minutes(5)" attempts="3" /> |
| |
| <late-process policy="exp-backoff" delay="hours(1)"> |
| <late-input input="input" workflow-path="/projects/bootcamp/workflow/lateinput" /> |
| </late-process> |
| </process> |
| </verbatim> |
| |
| ---++++ Oozie Workflow |
| The sample user workflow contains 3 actions: |
| * Pig action - Executes pig script /projects/bootcamp/workflow/script.pig |
| * concatenator - Java action that concatenates part files and generates a single file |
| * file upload - ssh action that gets the concatenated file from hadoop and sends the file to a remote host |
| |
| <verbatim> |
| <workflow-app xmlns="uri:oozie:workflow:0.2" name="sample-wf"> |
| <start to="pig" /> |
| |
| <action name="pig"> |
| <pig> |
| <job-tracker>${jobTracker}</job-tracker> |
| <name-node>${nameNode}</name-node> |
| <prepare> |
| <delete path="${output}"/> |
| </prepare> |
| <configuration> |
| <property> |
| <name>mapred.job.queue.name</name> |
| <value>${queueName}</value> |
| </property> |
| <property> |
| <name>mapreduce.fileoutputcommitter.marksuccessfuljobs</name> |
| <value>true</value> |
| </property> |
| </configuration> |
| <script>${nameNode}/projects/bootcamp/workflow/script.pig</script> |
| <param>input=${input}</param> |
| <param>output=${output}</param> |
| <file>lib/dependent.jar</file> |
| </pig> |
| <ok to="concatenator" /> |
| <error to="fail" /> |
| </action> |
| |
| <action name="concatenator"> |
| <java> |
| <job-tracker>${jobTracker}</job-tracker> |
| <name-node>${nameNode}</name-node> |
| <prepare> |
| <delete path="${nameNode}/projects/bootcamp/concat/data-${fileTimestamp}.csv"/> |
| </prepare> |
| <configuration> |
| <property> |
| <name>mapred.job.queue.name</name> |
| <value>${queueName}</value> |
| </property> |
| </configuration> |
| <main-class>com.wf.Concatenator</main-class> |
| <arg>${output}</arg> |
| <arg>${nameNode}/projects/bootcamp/concat/data-${fileTimestamp}.csv</arg> |
| </java> |
| <ok to="fileupload" /> |
| <error to="fail"/> |
| </action> |
| |
| <action name="fileupload"> |
| <ssh> |
| <host>localhost</host> |
| <command>/tmp/fileupload.sh</command> |
| <args>${nameNode}/projects/bootcamp/concat/data-${fileTimestamp}.csv</args> |
| <args>${wf:conf("ssh.host")}</args> |
| <capture-output/> |
| </ssh> |
| <ok to="fileUploadDecision" /> |
| <error to="fail"/> |
| </action> |
| |
| <decision name="fileUploadDecision"> |
| <switch> |
| <case to="end"> |
| ${wf:actionData('fileupload')['output'] == '0'} |
| </case> |
| <default to="fail"/> |
| </switch> |
| </decision> |
| |
| <kill name="fail"> |
| <message>Workflow failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message> |
| </kill> |
| |
| <end name="end" /> |
| </workflow-app> |
| </verbatim> |
| |
| ---++++ File Upload Script |
| The script gets the file from hadoop, rsyncs the file to /tmp on remote host and deletes the file from hadoop |
| <verbatim> |
| #!/bin/bash |
| |
| trap 'echo "output=$?"; exit $?' ERR INT TERM |
| |
| echo "Arguments: $@" |
| SRCFILE=$1 |
| DESTHOST=$3 |
| |
| FILENAME=`basename $SRCFILE` |
| rm -f /tmp/$FILENAME |
| hadoop fs -copyToLocal $SRCFILE /tmp/ |
| echo "Copied $SRCFILE to /tmp" |
| |
| rsync -ztv --rsh=ssh --stats /tmp/$FILENAME $DESTHOST:/tmp |
| echo "rsynced $FILENAME to $DESTUSER@$DESTHOST:$DESTFILE" |
| |
| hadoop fs -rmr $SRCFILE |
| echo "Deleted $SRCFILE" |
| |
| rm -f /tmp/$FILENAME |
| echo "output=0" |
| </verbatim> |
