org.apache.flink.storm.wrappers

Class SpoutWrapper<OUT>

java.lang.Object

org.apache.flink.api.common.functions.AbstractRichFunction

org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction<OUT>

org.apache.flink.storm.wrappers.SpoutWrapper<OUT>

All Implemented Interfaces:

Serializable, Function, RichFunction, StoppableFunction, ParallelSourceFunction<OUT>, SourceFunction<OUT>

public final class SpoutWrapper<OUT>
extends RichParallelSourceFunction<OUT>
implements StoppableFunction

A SpoutWrapper wraps an IRichSpout in order to execute it within a Flink Streaming program. It takes the spout's output tuples and transforms them into Flink tuples of type OUT (see SpoutCollector for supported types).

Per default, SpoutWrapper calls the wrapped spout's nextTuple() method in an infinite loop.
Alternatively, SpoutWrapper can call nextTuple() for a finite number of times and terminate automatically afterwards (for finite input streams). The number of nextTuple() calls can be specified as a certain number of invocations or can be undefined. In the undefined case, SpoutWrapper terminates if no record was emitted to the output collector for the first time during a call to nextTuple().
If the given spout implements FiniteSpout interface and numberOfInvocations is not provided or is null, SpoutWrapper calls nextTuple() method until FiniteSpout.reachedEnd() returns true.

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from interface org.apache.flink.streaming.api.functions.source.SourceFunction

SourceFunction.SourceContext<T>

Constructor Summary

Constructors

Constructor and Description
SpoutWrapper(backtype.storm.topology.IRichSpout spout) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, Collection<String> rawOutputs) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, Collection<String> rawOutputs, Integer numberOfInvocations) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, Integer numberOfInvocations) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, String[] rawOutputs) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, String[] rawOutputs, Integer numberOfInvocations) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times.
SpoutWrapper(backtype.storm.topology.IRichSpout spout, String name, Collection<String> rawOutputs, Integer numberOfInvocations) Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times.

Method Summary

Modifier and Type	Method and Description
void	cancel() Cancels the source.
void	close() Tear-down method for the user code.
void	run(SourceFunction.SourceContext<OUT> ctx) Starts the source.
void	setStormTopology(backtype.storm.generated.StormTopology stormTopology) Sets the original Storm topology.
void	stop() Stops the source.

Methods inherited from class org.apache.flink.api.common.functions.AbstractRichFunction

getIterationRuntimeContext, getRuntimeContext, open, setRuntimeContext

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop. The output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

Throws:

IllegalArgumentException - If the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    Integer numberOfInvocations)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times. The output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

numberOfInvocations - The number of calls to ISpout.nextTuple(). If value is negative, SpoutWrapper terminates if no tuple was emitted for the first time. If value is null, finite invocation is disabled.

Throws:

IllegalArgumentException - If the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    String[] rawOutputs)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop. The output type can be any type if parameter rawOutput is true and the spout's number of declared output tuples is 1. If rawOutput is false the output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

rawOutputs - Contains stream names if a single attribute output stream, should not be of type Tuple1 but be of a raw type. (Can be null.)

Throws:

IllegalArgumentException - If rawOuput is true and the number of declared output attributes is not 1 or if rawOuput is false and the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    String[] rawOutputs,
                    Integer numberOfInvocations)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times. The output type can be any type if parameter rawOutput is true and the spout's number of declared output tuples is 1. If rawOutput is false the output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

rawOutputs - Contains stream names if a single attribute output stream, should not be of type Tuple1 but be of a raw type. (Can be null.)

numberOfInvocations - The number of calls to ISpout.nextTuple(). If value is negative, SpoutWrapper terminates if no tuple was emitted for the first time. If value is null, finite invocation is disabled.

Throws:

IllegalArgumentException - If rawOuput is true and the number of declared output attributes is not 1 or if rawOuput is false and the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    Collection<String> rawOutputs)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout in an infinite loop. The output type can be any type if parameter rawOutput is true and the spout's number of declared output tuples is 1. If rawOutput is false the output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

rawOutputs - Contains stream names if a single attribute output stream, should not be of type Tuple1 but be of a raw type. (Can be null.)

Throws:

IllegalArgumentException - If rawOuput is true and the number of declared output attributes is not 1 or if rawOuput is false and the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    Collection<String> rawOutputs,
                    Integer numberOfInvocations)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times. The output type can be any type if parameter rawOutput is true and the spout's number of declared output tuples is 1. If rawOutput is false the output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

rawOutputs - Contains stream names if a single attribute output stream, should not be of type Tuple1 but be of a raw type. (Can be null.)

numberOfInvocations - The number of calls to ISpout.nextTuple(). If value is negative, SpoutWrapper terminates if no tuple was emitted for the first time. If value is null, finite invocation is disabled.

Throws:

IllegalArgumentException - If rawOuput is true and the number of declared output attributes is not 1 or if rawOuput is false and the number of declared output attributes is not with range [0;25].

SpoutWrapper

public SpoutWrapper(backtype.storm.topology.IRichSpout spout,
                    String name,
                    Collection<String> rawOutputs,
                    Integer numberOfInvocations)
             throws IllegalArgumentException

Instantiates a new SpoutWrapper that calls the nextTuple() method of the given spout a finite number of times. The output type can be any type if parameter rawOutput is true and the spout's number of declared output tuples is 1. If rawOutput is false the output type will be one of Tuple0 to Tuple25 depending on the spout's declared number of attributes.

Parameters:

spout - The spout to be used.

name - The name of the spout.

rawOutputs - Contains stream names if a single attribute output stream, should not be of type Tuple1 but be of a raw type. (Can be null.)

numberOfInvocations - The number of calls to ISpout.nextTuple(). If value is negative, SpoutWrapper terminates if no tuple was emitted for the first time. If value is null, finite invocation is disabled.

Throws:

IllegalArgumentException - If rawOuput is true and the number of declared output attributes is not 1 or if rawOuput is false and the number of declared output attributes is not with range [0;25].

Method Detail

setStormTopology

public void setStormTopology(backtype.storm.generated.StormTopology stormTopology)

Sets the original Storm topology.

Parameters:

stormTopology - The original Storm topology.

run

public final void run(SourceFunction.SourceContext<OUT> ctx)
               throws Exception

Description copied from interface: SourceFunction

Starts the source. Implementations can use the SourceFunction.SourceContext emit elements.

Sources that implement Checkpointed must lock on the checkpoint lock (using a synchronized block) before updating internal state and emitting elements, to make both an atomic operation:

  public class ExampleSource<T> implements SourceFunction<T>, Checkpointed<Long> {
      private long count = 0L;
      private volatile boolean isRunning = true;

      public void run(SourceContext<T> ctx) {
          while (isRunning && count < 1000) {
              synchronized (ctx.getCheckpointLock()) {
                  ctx.collect(count);
                  count++;
              }
          }
      }

      public void cancel() {
          isRunning = false;
      }

      public Long snapshotState(long checkpointId, long checkpointTimestamp) { return count; }

      public void restoreState(Long state) { this.count = state; }
 }
 

Specified by:

run in interface SourceFunction<OUT>

Parameters:

ctx - The context to emit elements to and for accessing locks.

Throws:

Exception

cancel

public void cancel()

Cancels the source. Most sources will have a while loop inside the SourceFunction.run(SourceContext) method. The implementation needs to ensure that the source will break out of that loop after this method is called.

A typical pattern is to have an "volatile boolean isRunning" flag that is set to false in this method. That flag is checked in the loop condition.

When a source is canceled, the executing thread will also be interrupted (via Thread.interrupt()). The interruption happens strictly after this method has been called, so any interruption handler can rely on the fact that this method has completed. It is good practice to make any flags altered by this method "volatile", in order to guarantee the visibility of the effects of this method to any interruption handler.

Sets the isRunning flag to false.

Specified by:

cancel in interface SourceFunction<OUT>

stop

public void stop()

Stops the source. In contrast to cancel() this is a request to the source function to shut down gracefully. Pending data can still be emitted and it is not required to stop immediately -- however, in the near future. The job will keep running until all emitted data is processed completely.

Most streaming sources will have a while loop inside the run() method. You need to ensure that the source will break out of this loop. This can be achieved by having a volatile field "isRunning" that is checked in the loop and that is set to false in this method.

The call to stop() should not block and not throw any exception.

Sets the isRunning flag to false.

Specified by:

stop in interface StoppableFunction

close

public void close()
           throws Exception

Description copied from interface: RichFunction

Tear-down method for the user code. It is called after the last call to the main working methods (e.g. map or join). For functions that are part of an iteration, this method will be invoked after each iteration superstep.

This method can be used for clean up work.

Specified by:

close in interface RichFunction

Overrides:

close in class AbstractRichFunction

Throws:

Exception - Implementations may forward exceptions, which are caught by the runtime. When the runtime catches an exception, it aborts the task and lets the fail-over logic decide whether to retry the task execution.