Package org.apache.storm.bolt

# Class JoinBolt

java.lang.Object
org.apache.storm.topology.base.BaseWindowedBolt
org.apache.storm.bolt.JoinBolt

All Implemented Interfaces:
  `Serializable`, `IComponent`, `IWindowedBolt`

public class JoinBolt extends BaseWindowedBolt

See Also:
  - Serialized Form

- ## Nested Class Summary

  Nested Classes
  Modifier and Type
  Class
  Description
  `protected static class`
  `JoinBolt.FieldSelector`
  `protected class`
  `JoinBolt.JoinAccumulator`
  `protected static class`
  `JoinBolt.JoinInfo`
  Describes how to join the other stream with the current stream.
  `protected static enum`
  `JoinBolt.JoinType`
  `protected class`
  `JoinBolt.ResultRecord`
  `static enum`
  `JoinBolt.Selector`

  ## Nested classes/interfaces inherited from class org.apache.storm.topology.base.BaseWindowedBolt

  `BaseWindowedBolt.Count, BaseWindowedBolt.Duration`

- ## Field Summary

  Fields
  Modifier and Type
  Field
  Description
  `protected LinkedHashMap<String,JoinBolt.JoinInfo>`
  `joinCriteria`
  `protected JoinBolt.FieldSelector[]`
  `outputFields`
  `protected String`
  `outputStreamName`
  `protected final JoinBolt.Selector`
  `selectorType`

  ### Fields inherited from class org.apache.storm.topology.base.BaseWindowedBolt

  `timestampExtractor, windowConfiguration`

- ## Constructor Summary

  Constructors
  Constructor
  Description
  `JoinBolt(String sourceId, String fieldName)`
  Calls JoinBolt(Selector.SOURCE, sourceId, fieldName)
  `JoinBolt(JoinBolt.Selector type, String srcOrStreamId, String fieldName)`
  Introduces the first stream to start the join with.

- ## Method Summary

  Modifier and Type
  Method
  Description
  `void`
  `declareOutputFields(OutputFieldsDeclarer declarer)`
  Declare the output schema for all the streams of this topology.
  `protected JoinBolt.JoinAccumulator`
  `doInnerJoin(JoinBolt.JoinAccumulator probe, Map<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)`
  `protected JoinBolt.JoinAccumulator`
  `doJoin(JoinBolt.JoinAccumulator probe, HashMap<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)`
  `protected JoinBolt.JoinAccumulator`
  `doLeftJoin(JoinBolt.JoinAccumulator probe, Map<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)`
  `protected ArrayList<Object>`
  `doProjection(ArrayList<Tuple> tuples, JoinBolt.FieldSelector[] projectionFields)`
  `void`
  `execute(TupleWindow inputWindow)`
  Process the tuple window and optionally emit new tuples based on the tuples in the input window.
  `protected JoinBolt.JoinAccumulator`
  `hashJoin(List<Tuple> tuples)`
  `JoinBolt`
  `join(String newStream, String field, String priorStream)`
  Performs inner Join with the newStream.
  `JoinBolt`
  `leftJoin(String newStream, String field, String priorStream)`
  Performs left Join with the newStream.
  `protected Object`
  `lookupField(JoinBolt.FieldSelector fieldSelector, Tuple tuple)`
  `void`
  `prepare(Map<String,Object> topoConf, TopologyContext context, OutputCollector collector)`
  This is similar to the `IBolt.prepare(Map, TopologyContext, OutputCollector)` except that while emitting, the tuples are automatically anchored to the tuples in the inputWindow.
  `JoinBolt`
  `select(String commaSeparatedKeys)`
  Specify projection fields.
  `BaseWindowedBolt`
  `withLag(BaseWindowedBolt.Duration duration)`
  Specify the maximum time lag of the tuple timestamp in milliseconds.
  `JoinBolt`
  `withLateTupleStream(String streamId)`
  Specify a stream id on which late tuples are going to be emitted.
  `JoinBolt`
  `withOutputStream(String streamName)`
  Optional.
  `JoinBolt`
  `withTimestampExtractor(TimestampExtractor timestampExtractor)`
  Specify the timestamp extractor implementation.
  `JoinBolt`
  `withTimestampField(String fieldName)`
  Specify a field in the tuple that represents the timestamp as a long value.
  `JoinBolt`
  `withTumblingWindow(BaseWindowedBolt.Count count)`
  A count based tumbling window.
  `JoinBolt`
  `withTumblingWindow(BaseWindowedBolt.Duration duration)`
  A time duration based tumbling window.
  `BaseWindowedBolt`
  `withWatermarkInterval(BaseWindowedBolt.Duration interval)`
  Specify the watermark event generation interval.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Count windowLength)`
  A tuple count based window that slides with every incoming tuple.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Count windowLength, BaseWindowedBolt.Count slidingInterval)`
  Tuple count based sliding window configuration.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Count windowLength, BaseWindowedBolt.Duration slidingInterval)`
  Tuple count and time duration based sliding window configuration.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Duration windowLength)`
  A time duration based window that slides with every incoming tuple.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Duration windowLength, BaseWindowedBolt.Count slidingInterval)`
  Time duration and count based sliding window configuration.
  `JoinBolt`
  `withWindow(BaseWindowedBolt.Duration windowLength, BaseWindowedBolt.Duration slidingInterval)`
  Time duration based sliding window configuration.

  ### Methods inherited from class org.apache.storm.topology.base.BaseWindowedBolt

  `cleanup, getComponentConfiguration, getTimestampExtractor`

  ### Methods inherited from class java.lang.Object

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

- ## Field Details

  - ### selectorType

    protected final JoinBolt.Selector selectorType

  - ### joinCriteria

    protected LinkedHashMap<String,JoinBolt.JoinInfo> joinCriteria

  - ### outputFields

    protected JoinBolt.FieldSelector[] outputFields

  - ### outputStreamName

    protected String outputStreamName

- ## Constructor Details

  - ### JoinBolt

    public JoinBolt(String sourceId, String fieldName)
    Calls JoinBolt(Selector.SOURCE, sourceId, fieldName)
    Parameters:
      `sourceId` - Id of source component (spout/bolt) from which this bolt is receiving data
      `fieldName` - the field to use for joining the stream (x.y.z format)

  - ### JoinBolt

    public JoinBolt(JoinBolt.Selector type, String srcOrStreamId, String fieldName)
    Introduces the first stream to start the join with. Equivalent SQL ... select .... from srcOrStreamId ...
    Parameters:
      `type` - Specifies whether 'srcOrStreamId' refers to stream name/source component
      `srcOrStreamId` - name of stream OR source component
      `fieldName` - the field to use for joining the stream (x.y.z format)

- ## Method Details

  - ### withOutputStream

    public JoinBolt withOutputStream(String streamName)
    Optional. Allows naming the output stream of this bolt. If not specified, the emits will happen on 'default' stream.

  - ### join

    public JoinBolt join(String newStream, String field, String priorStream)
    Performs inner Join with the newStream. SQL: `from priorStream inner join newStream on newStream.field = priorStream.field1` same as: `new WindowedQueryBolt(priorStream,field1). join(newStream, field, priorStream);` Note: priorStream must be previously joined. Valid ex: `new WindowedQueryBolt(s1,k1). join(s2,k2, s1). join(s3,k3, s2);` Invalid ex: `new WindowedQueryBolt(s1,k1). join(s3,k3, s2). join(s2,k2, s1);`
    Parameters:
      `newStream` - Either stream name or name of upstream component
      `field` - the field on which to perform the join

  - ### leftJoin

    public JoinBolt leftJoin(String newStream, String field, String priorStream)
    Performs left Join with the newStream. SQL : from stream1 left join stream2 on stream2.field = stream1.field1 same as: new WindowedQueryBolt(stream1, field1). leftJoin(stream2, field, stream1);

    Note: priorStream must be previously joined Valid ex: new WindowedQueryBolt(s1,k1). leftJoin(s2,k2, s1). leftJoin(s3,k3, s2); Invalid ex: new WindowedQueryBolt(s1,k1). leftJoin(s3,k3, s2). leftJoin(s2,k2, s1);

    Parameters:
      `newStream` - Either a name of a stream or an upstream component
      `field` - the field on which to perform the join

  - ### select

    public JoinBolt select(String commaSeparatedKeys)
    Specify projection fields. i.e. Specifies the fields to include in the output. e.g: .select("field1, stream2:field2, field3") Nested Key names are supported for nested types: e.g: .select("outerKey1.innerKey1, outerKey1.innerKey2, stream3:outerKey2.innerKey3)" Inner types (non leaf) must be Map<> in order to support nested lookup using this dot notation This selected fields implicitly declare the output fieldNames for the bolt based.

  - ### declareOutputFields

    public void declareOutputFields(OutputFieldsDeclarer declarer)
    Description copied from interface: `IComponent`
    Declare the output schema for all the streams of this topology.
    Specified by:
      `declareOutputFields` in interface `IComponent`
    Overrides:
      `declareOutputFields` in class `BaseWindowedBolt`
    Parameters:
      `declarer` - this is used to declare output stream ids, output fields, and whether or not each output stream is a direct stream

  - ### prepare

    public void prepare(Map<String,Object> topoConf, TopologyContext context, OutputCollector collector)
    Description copied from interface: `IWindowedBolt`
    This is similar to the `IBolt.prepare(Map, TopologyContext, OutputCollector)` except that while emitting, the tuples are automatically anchored to the tuples in the inputWindow.
    Specified by:
      `prepare` in interface `IWindowedBolt`
    Overrides:
      `prepare` in class `BaseWindowedBolt`

  - ### execute

    public void execute(TupleWindow inputWindow)
    Description copied from interface: `IWindowedBolt`
    Process the tuple window and optionally emit new tuples based on the tuples in the input window.

  - ### hashJoin

    protected JoinBolt.JoinAccumulator hashJoin(List<Tuple> tuples)

  - ### doJoin

    protected JoinBolt.JoinAccumulator doJoin(JoinBolt.JoinAccumulator probe, HashMap<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)

  - ### doInnerJoin

    protected JoinBolt.JoinAccumulator doInnerJoin(JoinBolt.JoinAccumulator probe, Map<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)

  - ### doLeftJoin

    protected JoinBolt.JoinAccumulator doLeftJoin(JoinBolt.JoinAccumulator probe, Map<Object,ArrayList<Tuple>> buildInput, JoinBolt.JoinInfo joinInfo, boolean finalJoin)

  - ### doProjection

    protected ArrayList<Object> doProjection(ArrayList<Tuple> tuples, JoinBolt.FieldSelector[] projectionFields)

  - ### lookupField

    protected Object lookupField(JoinBolt.FieldSelector fieldSelector, Tuple tuple)

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Count windowLength, BaseWindowedBolt.Count slidingInterval)
    Description copied from class: `BaseWindowedBolt`
    Tuple count based sliding window configuration.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the number of tuples in the window
      `slidingInterval` - the number of tuples after which the window slides

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Count windowLength, BaseWindowedBolt.Duration slidingInterval)
    Description copied from class: `BaseWindowedBolt`
    Tuple count and time duration based sliding window configuration.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the number of tuples in the window
      `slidingInterval` - the time duration after which the window slides

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Duration windowLength, BaseWindowedBolt.Count slidingInterval)
    Description copied from class: `BaseWindowedBolt`
    Time duration and count based sliding window configuration.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the time duration of the window
      `slidingInterval` - the number of tuples after which the window slides

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Duration windowLength, BaseWindowedBolt.Duration slidingInterval)
    Description copied from class: `BaseWindowedBolt`
    Time duration based sliding window configuration.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the time duration of the window
      `slidingInterval` - the time duration after which the window slides

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Count windowLength)
    Description copied from class: `BaseWindowedBolt`
    A tuple count based window that slides with every incoming tuple.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the number of tuples in the window

  - ### withWindow

    public JoinBolt withWindow(BaseWindowedBolt.Duration windowLength)
    Description copied from class: `BaseWindowedBolt`
    A time duration based window that slides with every incoming tuple.
    Overrides:
      `withWindow` in class `BaseWindowedBolt`
    Parameters:
      `windowLength` - the time duration of the window

  - ### withTumblingWindow

    public JoinBolt withTumblingWindow(BaseWindowedBolt.Count count)
    Description copied from class: `BaseWindowedBolt`
    A count based tumbling window.
    Overrides:
      `withTumblingWindow` in class `BaseWindowedBolt`
    Parameters:
      `count` - the number of tuples after which the window tumbles

  - ### withTumblingWindow

    public JoinBolt withTumblingWindow(BaseWindowedBolt.Duration duration)
    Description copied from class: `BaseWindowedBolt`
    A time duration based tumbling window.
    Overrides:
      `withTumblingWindow` in class `BaseWindowedBolt`
    Parameters:
      `duration` - the time duration after which the window tumbles

  - ### withTimestampField

    public JoinBolt withTimestampField(String fieldName)
    Description copied from class: `BaseWindowedBolt`
    Specify a field in the tuple that represents the timestamp as a long value. If this field is not present in the incoming tuple, an `IllegalArgumentException` will be thrown. The field MUST contain a timestamp in milliseconds
    Overrides:
      `withTimestampField` in class `BaseWindowedBolt`
    Parameters:
      `fieldName` - the name of the field that contains the timestamp

  - ### withTimestampExtractor

    public JoinBolt withTimestampExtractor(TimestampExtractor timestampExtractor)
    Description copied from class: `BaseWindowedBolt`
    Specify the timestamp extractor implementation.
    Overrides:
      `withTimestampExtractor` in class `BaseWindowedBolt`
    Parameters:
      `timestampExtractor` - the `TimestampExtractor` implementation

  - ### withLateTupleStream

    public JoinBolt withLateTupleStream(String streamId)
    Description copied from class: `BaseWindowedBolt`
    Specify a stream id on which late tuples are going to be emitted. They are going to be accessible via the `WindowedBoltExecutor.LATE_TUPLE_FIELD` field. It must be defined on a per-component basis, and in conjunction with the `BaseWindowedBolt.withTimestampField(java.lang.String)`, otherwise `IllegalArgumentException` will be thrown.
    Overrides:
      `withLateTupleStream` in class `BaseWindowedBolt`
    Parameters:
      `streamId` - the name of the stream used to emit late tuples on

  - ### withLag

    public BaseWindowedBolt withLag(BaseWindowedBolt.Duration duration)
    Description copied from class: `BaseWindowedBolt`
    Specify the maximum time lag of the tuple timestamp in milliseconds. It means that the tuple timestamps cannot be out of order by more than this amount.
    Overrides:
      `withLag` in class `BaseWindowedBolt`
    Parameters:
      `duration` - the max lag duration

  - ### withWatermarkInterval

    public BaseWindowedBolt withWatermarkInterval(BaseWindowedBolt.Duration interval)
    Description copied from class: `BaseWindowedBolt`
    Specify the watermark event generation interval. For tuple based timestamps, watermark events are used to track the progress of time
    Overrides:
      `withWatermarkInterval` in class `BaseWindowedBolt`
    Parameters:
      `interval` - the interval at which watermark events are generated