org.apache.flink.storm.wrappers

Class BoltWrapper<IN,OUT>

java.lang.Object

org.apache.flink.streaming.api.operators.AbstractStreamOperator<OUT>

org.apache.flink.storm.wrappers.BoltWrapper<IN,OUT>

All Implemented Interfaces:

Serializable, CheckpointListener, KeyContext, OneInputStreamOperator<IN,OUT>, StreamOperator<OUT>, Disposable

Direct Known Subclasses:

MergedInputsBoltWrapper

public class BoltWrapper<IN,OUT>
extends AbstractStreamOperator<OUT>
implements OneInputStreamOperator<IN,OUT>

A BoltWrapper wraps an IRichBolt in order to execute the Storm bolt within a Flink Streaming program. It takes the Flink input tuples of type IN and transforms them into StormTuples that the bolt can process. Furthermore, it takes the bolt's output tuples and transforms them into Flink tuples of type OUT (see AbstractStormCollector for supported types).

Works for single input streams only! See MergedInputsBoltWrapper for multi-input stream Bolts.

See Also:

Serialized Form

Nested Class Summary

Nested classes/interfaces inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

AbstractStreamOperator.CountingOutput<OUT>

Field Summary

Fields

Modifier and Type	Field and Description
protected org.apache.storm.topology.IRichBolt	bolt The wrapped Storm bolt.
static String	DEFAULT_ID The default input component ID.
static String	DEFUALT_BOLT_NAME The default bolt ID.
protected TimestampedCollector<OUT>	flinkCollector We have to use this because Operators must output StreamRecord.

Fields inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

chainingStrategy, config, latencyStats, LOG, metrics, output, timeServiceManager

Constructor Summary

Constructors

Constructor and Description
BoltWrapper(org.apache.storm.topology.IRichBolt bolt) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, Collection<String> rawOutputs) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, org.apache.storm.tuple.Fields inputSchema) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, org.apache.storm.tuple.Fields inputSchema, Collection<String> rawOutputs) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, org.apache.storm.tuple.Fields inputSchema, String[] rawOutputs) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, String[] rawOutputs) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.
BoltWrapper(org.apache.storm.topology.IRichBolt bolt, String name, String inputStreamId, String inputComponentId, org.apache.storm.tuple.Fields inputSchema, Collection<String> rawOutputs) Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program.

Method Summary

Modifier and Type	Method and Description
void	dispose() This method is called at the very end of the operator's life, both in the case of a successful completion of the operation, and in the case of a failure and canceling.
void	open() This method is called immediately before any elements are processed, it should contain the operator's initialization logic, e.g.
void	processElement(StreamRecord<IN> element) Processes one element that arrived at this operator.
void	setStormTopology(org.apache.storm.generated.StormTopology stormTopology) Sets the original Storm topology.

Methods inherited from class org.apache.flink.streaming.api.operators.AbstractStreamOperator

close, getChainingStrategy, getContainingTask, getCurrentKey, getExecutionConfig, getInternalTimerService, getKeyedStateBackend, getKeyedStateStore, getMetricGroup, getOperatorConfig, getOperatorID, getOperatorName, getOperatorStateBackend, getOrCreateKeyedState, getPartitionedState, getPartitionedState, getProcessingTimeService, getRuntimeContext, getUserCodeClassloader, initializeState, initializeState, notifyCheckpointComplete, numEventTimeTimers, numProcessingTimeTimers, prepareSnapshotPreBarrier, processLatencyMarker, processLatencyMarker1, processLatencyMarker2, processWatermark, processWatermark1, processWatermark2, reportOrForwardLatencyMarker, setChainingStrategy, setCurrentKey, setKeyContextElement1, setKeyContextElement2, setup, snapshotState, snapshotState

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.apache.flink.streaming.api.operators.OneInputStreamOperator

processLatencyMarker, processWatermark

Methods inherited from interface org.apache.flink.streaming.api.operators.StreamOperator

close, getChainingStrategy, getMetricGroup, getOperatorID, initializeState, prepareSnapshotPreBarrier, setChainingStrategy, setKeyContextElement1, setKeyContextElement2, setup, snapshotState

Methods inherited from interface org.apache.flink.runtime.state.CheckpointListener

notifyCheckpointComplete

Methods inherited from interface org.apache.flink.streaming.api.operators.KeyContext

getCurrentKey, setCurrentKey

Field Detail

DEFAULT_ID

public static final String DEFAULT_ID

The default input component ID.

See Also:

Constant Field Values

DEFUALT_BOLT_NAME

public static final String DEFUALT_BOLT_NAME

The default bolt ID.

See Also:

Constant Field Values

bolt

protected final org.apache.storm.topology.IRichBolt bolt

The wrapped Storm bolt.

flinkCollector

protected transient TimestampedCollector<OUT> flinkCollector

We have to use this because Operators must output StreamRecord.

Constructor Detail

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible for POJO input types. The output type will be one of Tuple0 to Tuple25 depending on the bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt to be used.

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   org.apache.storm.tuple.Fields inputSchema)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. The given input schema enable attribute-by-name access for input types Tuple0 to Tuple25. The output type will be one of Tuple0 to Tuple25 depending on the bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt to be used.

inputSchema - The schema (ie, ordered field names) of the input stream. @throws IllegalArgumentException

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   String[] rawOutputs)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible for POJO input types. The output type can be any type if parameter rawOutput is true and the bolt'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 bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt 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.

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   Collection<String> rawOutputs)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. As no input schema is defined, attribute-by-name access in only possible for POJO input types. The output type can be any type if parameter rawOutput is true and the bolt'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 bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt 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.

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   org.apache.storm.tuple.Fields inputSchema,
                   String[] rawOutputs)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. The given input schema enable attribute-by-name access for input types Tuple0 to Tuple25. The output type can be any type if parameter rawOutput is true and the bolt'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 bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt to be used.

inputSchema - The schema (ie, ordered field names) of the input stream.

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

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   org.apache.storm.tuple.Fields inputSchema,
                   Collection<String> rawOutputs)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. The given input schema enable attribute-by-name access for input types Tuple0 to Tuple25. The output type can be any type if parameter rawOutput is true and the bolt'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 bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt to be used.

inputSchema - The schema (ie, ordered field names) of the input stream. @throws IllegalArgumentException If rawOutput is true and the number of declared output attributes is not 1 or if rawOutput is false and the number of declared output attributes is not with range

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

Throws:

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

BoltWrapper

public BoltWrapper(org.apache.storm.topology.IRichBolt bolt,
                   String name,
                   String inputStreamId,
                   String inputComponentId,
                   org.apache.storm.tuple.Fields inputSchema,
                   Collection<String> rawOutputs)
            throws IllegalArgumentException

Instantiates a new BoltWrapper that wraps the given Storm bolt such that it can be used within a Flink streaming program. The given input schema enable attribute-by-name access for input types Tuple0 to Tuple25. The output type can be any type if parameter rawOutput is true and the bolt'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 bolt's declared number of attributes.

Parameters:

bolt - The Storm bolt to be used.

name - The name of the bolt.

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

Throws:

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

Method Detail

setStormTopology

public void setStormTopology(org.apache.storm.generated.StormTopology stormTopology)

Sets the original Storm topology.

Parameters:

stormTopology - The original Storm topology.

open

public void open()
          throws Exception

Description copied from class: AbstractStreamOperator

This method is called immediately before any elements are processed, it should contain the operator's initialization logic, e.g. state initialization.

The default implementation does nothing.

Specified by:

open in interface StreamOperator<OUT>

Overrides:

open in class AbstractStreamOperator<OUT>

Throws:

Exception - An exception in this method causes the operator to fail.

dispose

public void dispose()
             throws Exception

Description copied from class: AbstractStreamOperator

This method is called at the very end of the operator's life, both in the case of a successful completion of the operation, and in the case of a failure and canceling.

This method is expected to make a thorough effort to release all resources that the operator has acquired.

Specified by:

dispose in interface StreamOperator<OUT>

Specified by:

dispose in interface Disposable

Overrides:

dispose in class AbstractStreamOperator<OUT>

Throws:

Exception - if something goes wrong during disposal.

processElement

public void processElement(StreamRecord<IN> element)
                    throws Exception

Description copied from interface: OneInputStreamOperator

Processes one element that arrived at this operator. This method is guaranteed to not be called concurrently with other methods of the operator.

Specified by:

processElement in interface OneInputStreamOperator<IN,OUT>

Throws:

Exception