001/* A Spark Reduce stub that runs a Kepler workflow.
002 * 
003 * Copyright (c) 2014 The Regents of the University of California.
004 * All rights reserved.
005 *
006 * '$Author: crawl $'
007 * '$Date: 2016-10-17 19:14:15 +0000 (Mon, 17 Oct 2016) $' 
008 * '$Revision: 34532 $'
009 * 
010 * Permission is hereby granted, without written agreement and without
011 * license or royalty fees, to use, copy, modify, and distribute this
012 * software and its documentation for any purpose, provided that the above
013 * copyright notice and the following two paragraphs appear in all copies
014 * of this software.
015 *
016 * IN NO EVENT SHALL THE UNIVERSITY OF CALIFORNIA BE LIABLE TO ANY PARTY
017 * FOR DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES
018 * ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION, EVEN IF
019 * THE UNIVERSITY OF CALIFORNIA HAS BEEN ADVISED OF THE POSSIBILITY OF
020 * SUCH DAMAGE.
021 *
022 * THE UNIVERSITY OF CALIFORNIA SPECIFICALLY DISCLAIMS ANY WARRANTIES,
023 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
024 * MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE. THE SOFTWARE
025 * PROVIDED HEREUNDER IS ON AN "AS IS" BASIS, AND THE UNIVERSITY OF
026 * CALIFORNIA HAS NO OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES,
027 * ENHANCEMENTS, OR MODIFICATIONS.
028 *
029 */
030package org.kepler.spark.stub;
031
032import java.util.ArrayList;
033import java.util.Iterator;
034import java.util.LinkedList;
035import java.util.List;
036
037import org.kepler.ddp.actor.pattern.stub.ReduceInput;
038
039import ptolemy.data.ArrayToken;
040import ptolemy.data.Token;
041import ptolemy.kernel.util.IllegalActionException;
042import scala.Tuple2;
043
044/** A Spark stub to run a Kepler workflow in a map after the groupByKey operator.
045 * 
046 *  @author Daniel Crawl
047 *  @version $Id: KeplerReduceStub.java 34532 2016-10-17 19:14:15Z crawl $
048 *  
049 */
050public class KeplerReduceStub 
051    extends KeplerPairFlatMapFunction<Iterator<Tuple2<Object, Iterable<Object>>>, Object, Object> {
052
053    public KeplerReduceStub() {
054        super("ReduceInput", "ReduceOutput");
055    }
056
057    /** Transfer the data from Spark to Kepler, execute the workflow,
058     *  and then transfer the data from Kepler to Spark.
059     */
060    @Override
061    public Iterator<Tuple2<Object, Object>> call(Iterator<Tuple2<Object, Iterable<Object>>> iterator) throws Exception {
062
063        final List<Tuple2<Object, Object>> retval = new LinkedList<Tuple2<Object, Object>>();
064
065        // this method can be called with no input, so only proceed if
066        // input is present.
067        if(!iterator.hasNext()) {
068            return retval.iterator();
069        }
070
071        _initialize();
072        
073        ReduceInput reduceInputActor = (ReduceInput) _sourceActor;
074
075        while(iterator.hasNext()) {
076            
077            Tuple2<Object, Iterable<Object>> input = iterator.next();
078        
079            // place the input in the input actor
080            Token keyToken = StubUtilities.convertToToken(input._1);
081            ArrayToken valuesToken = null;
082            
083            final List<Token> tokenList = new ArrayList<Token>();
084            Iterator<?> valuesIterator = input._2.iterator();
085            while(valuesIterator.hasNext()) {
086                tokenList.add(StubUtilities.convertToToken(valuesIterator.next()));
087            }
088            try {
089                valuesToken = new ArrayToken(tokenList.toArray(new Token[tokenList.size()]));
090            } catch (IllegalActionException e) {
091                throw new RuntimeException("Error creating array token.", e);
092            }
093    
094            //System.out.println(name + " setInput start");
095            reduceInputActor.setInput(keyToken, valuesToken);
096            //System.out.println(name + " setInput end");
097            
098            if(_runWorkflowLifecyclePerInput) {
099                _manager.execute();
100            }
101    
102            // read the output from the output actor
103            try {
104                //System.out.println(name + " getOutput start");
105                final List<Token> outputTokenList = _sinkActor.getOutput();
106                //System.out.println(name + " getOutput end");
107                if(outputTokenList != null) {
108                    StubUtilities.convertTokenToTupleList(outputTokenList, retval);
109                } else if(_keplerManagerException != null) {
110                    throw _keplerManagerException;
111                }
112            } catch (IllegalActionException e) {
113                throw new RuntimeException("Error getting output for " + _sinkActor.getName() + ".", e);
114            }
115            
116        }
117        
118        _cleanup();
119        
120        return retval.iterator();
121        
122    }
123}