Module datarush.library
Package com.pervasive.datarush.operators.join

Class CrossJoin

  • All Implemented Interfaces:
    LogicalOperator
    public final class CrossJoin
extends ExecutableOperator
    Produce the cartesian product of two sets of records. The output is typed using the merge of the two input types; if the right type contains a field already named in the left type, it will be renamed to avoid collision. See TypeUtil.merge(RecordTokenType...) for details on the renaming process.

    To generate pairs, the right-hand data is temporarily stored to disk for multiple passes. The left-hand data is read into memory in chunks, which are then used to generate a full set of pairs with the iterable right-hand data. This process is repeated until the left-hand data is exhausted.

    No guarantee is made in respect to the ordering of the output data.

    • Constructor Detail

      • CrossJoin

        public CrossJoin()
        Performs a cross join between two inputs

    • Method Detail

      • getLeftInput

        public RecordPort getLeftInput()
        Returns the left input port
        Returns:
        the left input port

      • getRightInput

        public RecordPort getRightInput()
        Returns the right input port
        Returns:
        the right input port

      • getOutput

        public RecordPort getOutput()
        Returns the input port
        Returns:
        the output port

      • setBufferRows

        public void setBufferRows​(int rows)
        Sets the size (in rows) of the memory buffer used to generate output record. Larger values can increase performance due to decreased intermediate file buffering.
        Parameters:
        rows - number of rows to store in memory buffer

      • getBufferRows

        public int getBufferRows()
        Gets the size (in rows) of the memory buffer used to generate output records.
        Returns:
        the number of rows allowed in the memory buffer

      • computeMetadata

        protected void computeMetadata​(StreamingMetadataContext ctx)
        Description copied from class: StreamingOperator
        Implementations must adhere to the following contracts

        General

        Regardless of input ports/output port types, all implementations must do the following:

        1. Validation. Validation of configuration should always be performed first.
        2. Declare parallelizability.. Implementations must declare parallelizability by calling StreamingMetadataContext.parallelize(ParallelismStrategy).

        Input record ports

        Implementations with input record ports must declare the following:
        1. Required data ordering:
          2. Implementations that have data ordering requirements must declare them by calling RecordPort#setRequiredDataOrdering, otherwise data may arrive in any order.
        2. Required data distribution (only applies to parallelizable operators):
          3. Implementations that have data distribution requirements must declare them by calling RecordPort#setRequiredDataDistribution, otherwise data will arrive in an unspecified partial distribution.
        Note that if the upstream operator's output distribution/ordering is compatible with those required, we avoid a re-sort/re-distribution which is generally a very large savings from a performance standpoint. In addition, some operators may chose to query the upstream output distribution/ordering by calling RecordPort#getSourceDataDistribution and RecordPort#getSourceDataOrdering. These should be viewed as a hints to help chose a more efficient algorithm. In such cases, though, operators must still declare data ordering and data distribution requirements; otherwise there is no guarantee that data will arrive sorted/distributed as required.

        Output record ports

        Implementations with output record ports must declare the following:
        1. Type: Implementations must declare their output type by calling RecordPort#setType.
        Implementations with output record ports may declare the following:
        1. Output data ordering: Implementations that can make guarantees as to their output ordering may do so by calling RecordPort#setOutputDataOrdering
        2. Output data distribution (only applies to parallelizable operators): Implementations that can make guarantees as to their output distribution may do so by calling RecordPort#setOutputDataDistribution
        Note that both of these properties are optional; if unspecified, performance may suffer since the framework may unnecessarily re-sort/re-distributed the data.

        Input model ports

        In general, there is nothing special to declare for input model ports. Models are implicitly duplicated to all partitions when going from non-parallel to parallel operators. The case of a model going from a parallel to a non-parallel node is a special case of a "model reducer" operator. In the case of a model reducer, the downstream operator, must declare the following:
        1. Merge handler: Model reducers must declare a merge handler by calling AbstractModelPort#setMergeHandler.
        Note that MergeModel is a convenient, re-usable model reducer, parameterized with a merge-handler.

        Output model ports

        SimpleModelPort's have no associated metadata and therefore there is never any output metadata to declare. PMMLPort's, on the other hand, do have associated metadata. For all PMMLPorts, implementations must declare the following:
        1. pmmlModelSpec: Implementations must declare the PMML model spec by calling PMMLPort.setPMMLModelSpec.
        Specified by:
        computeMetadata in class StreamingOperator
        Parameters:
        ctx - the context

      • execute

        protected void execute​(ExecutionContext ctx)
        The controller handles chunking the LHS data into reasonable chunks that can be loaded into memory. The LHS chunks are then joined (cartesian product) against an iteration of the RHS data. Having to use an application graph and export many ports from that graph, the partitioning has to be handled directly. Subgraphs are run until the LHS data is exhausted.
        Specified by:
        execute in class ExecutableOperator
        Parameters:
        ctx - context in which to lookup physical ports bound to logical ports