001 /*
002 * Copyright (c) 2007-2015 Concurrent, Inc. All Rights Reserved.
003 *
004 * Project and contact information: http://www.cascading.org/
005 *
006 * This file is part of the Cascading project.
007 *
008 * Licensed under the Apache License, Version 2.0 (the "License");
009 * you may not use this file except in compliance with the License.
010 * You may obtain a copy of the License at
011 *
012 * http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing, software
015 * distributed under the License is distributed on an "AS IS" BASIS,
016 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017 * See the License for the specific language governing permissions and
018 * limitations under the License.
019 */
020
021 package cascading.stats.hadoop;
022
023 import java.util.ArrayList;
024 import java.util.Arrays;
025 import java.util.Collection;
026 import java.util.Collections;
027 import java.util.List;
028
029 import cascading.CascadingException;
030 import cascading.flow.Flow;
031 import cascading.management.state.ClientState;
032 import cascading.stats.FlowStats;
033 import cascading.stats.FlowStepStats;
034 import riffle.process.scheduler.ProcessException;
035 import riffle.process.scheduler.ProcessWrapper;
036
037 /**
038 * ProcessFlowStats is a sub-class of FlowStats which can fetch counters from a ProcessWrapper and hook them into the
039 * stats mechanism of Cascading.
040 *
041 * @deprecated ProcessFlowStats will be moved to a different package in Cascading 3.0.
042 */
043 @Deprecated
044 public class ProcessFlowStats extends FlowStats
045 {
046 /** The ProcessWrapper having the actual counters. */
047 private ProcessWrapper processWrapper;
048
049 /**
050 * Constructs a new ProcessFlowStats instance.
051 *
052 * @param flow a Flow instance.
053 * @param clientState The current client state.
054 * @param processWrapper a ProcessWrapper instance.
055 */
056 public ProcessFlowStats( Flow flow, ClientState clientState, ProcessWrapper processWrapper )
057 {
058 super( flow, clientState );
059 this.processWrapper = processWrapper;
060 }
061
062 @Override
063 public List<FlowStepStats> getFlowStepStats()
064 {
065 return getChildrenInternal();
066 }
067
068 @Override
069 public Collection getChildren()
070 {
071 return getChildrenInternal();
072 }
073
074 /**
075 * Internal method to retrieve the child stats objects from the ProcessWrapper.
076 */
077 private List<FlowStepStats> getChildrenInternal()
078 {
079 try
080 {
081 if( !processWrapper.hasChildren() )
082 {
083 if( processWrapper.hasCounters() )
084 return Arrays.<FlowStepStats>asList(
085 new ProcessStepStats( clientState, processWrapper.getCounters(), new ProcessFlowStep( processWrapper, 1 ) ) );
086 else
087 return Collections.emptyList();
088 }
089
090 List<FlowStepStats> childStepStats = new ArrayList<FlowStepStats>();
091 int counter = 0;
092 for( Object process : processWrapper.getChildren() )
093 {
094 ProcessWrapper childWrapper = new ProcessWrapper( process );
095 if( childWrapper.hasCounters() )
096 {
097 ProcessStepStats processStepStats = new ProcessStepStats( clientState, childWrapper.getCounters(),
098 new ProcessFlowStep( processWrapper, counter ) );
099
100 counter++;
101
102 childStepStats.add( processStepStats );
103 }
104 }
105 return childStepStats;
106 }
107 catch( ProcessException exception )
108 {
109 throw new CascadingException( exception );
110 }
111 }
112
113 @Override
114 public int getStepsCount()
115 {
116 try
117 {
118 if( !processWrapper.hasChildren() )
119 return 1; // there is always a step, even if it is opaque to us
120
121 return processWrapper.getChildren().size();
122 }
123 catch( ProcessException exception )
124 {
125 throw new CascadingException( exception );
126 }
127 }
128 }