Status

Current state: ACCEPTED

Discussion thread: N/A

JIRA:  Unable to render Jira issues macro, execution error.

Released: Samza 1.0

Purpose

This document outlines a proposal for extending Samza’s Execution Planner to verify agreement in partition count among the stream(s) behind Tables and other streams participating in Stream-Table Joins in applications written using Samza High-Level APIs.

Background

Motivating Example: Stream-Stream Join

In a typical application using Samza high-level API, the user defines a sequence of operations to be performed on a stream of messages using a set of methods defined on the org.apache.samza.operators.MessageStream interface.

For instance, to perform the operations illustrated in Fig. 1 on a stream of messages, a user can write the Samza app in listing 1 using Samza high-level API:

public class StreamStreamJoinApp implements StreamApplication {  
   @Override
    public void init(StreamGraph graph, Config config) {
      MessageStream s1 = graph
          .getInputStream("S1")
          .filter(/* Omitted for brevity */);

      MessageStream s2 = graph
          .getInputStream("S2");

      OutputStream s3 = graph.getOutputStream("S3");

      s1.join(s2, /* Omitted for brevity */)
        .sendTo(s3);
    }
}

This code gets translated by Samza’s core libraries into an internal representation where every operation/transformation is represented by an OperatorSpec object, a logical abstraction that describes the operation specified by the app author.

Some of these OperatorSpecs represent message streams which can be thought of as channels propagating data into and out of the application, e.g.

• InputOperatorSpecs represent input data streams, e.g. S1 and S2, from which input messages are read.

• OutputOperatorSpecs represent output data streams, e.g. S3, to which processed data is produced.

A common example of such message streams is Kafka topics which can be used to carry data consumed or produced by Samza applications. But the fact that Kafka topics are typically sharded into different partitions presents an interesting challenge to some stream operations. In the application above, for instance, it is specified that a Join operation is to be performed between a filtered version of data stream S1 and data stream S2. But in the case where these data streams are Kafka topics, Samza has to ensure that these Kafka topics are — at the very least — sharded into the same number of partitions, or else the Join operation would be invalid.

Responsibilities of Samza’s Execution Planner

Verifying Compatibility of Joined Input Streams

The Execution Planner is the core Samza module responsible for verifying that all streams participating in any given Join operation agree in partition count. To achieve this, it traverses the graph of OperatorSpecs produced by Samza High-Level API to verify compliance to this requirement among all such sets of streams.

Inferring Partition Counts of Intermediate Streams

Another closely-related responsibility of Samza’s Execution Planner is inferring partition counts of all intermediate streams present in the OperatorSpec graph. Such streams are introduced into the OperatorSpec graph whenever the Partition-By operation is used, and are represented by the same type of OperatorSpecs used to represent input streams, i.e. InputOperatorSpec. Unlike input streams however, intermediate streams have no defined partition counts by default. As we said, it is the Execution Planner that decides the partition count of every intermediate stream after traversing the OperatorSpec graph, according to the following rules:

1. Any intermediate stream joined with an input stream gets assigned the same partition count as that input stream.

2. Any intermediate stream not covered by the first rule is assigned the partition count value specified by the Samza config property job.intermediate.stream.partitions.

3. If no value is specified for job.intermediate.stream.partitions, the Execution Planner falls back to using the maximum partition count among all input and output streams, capped at a maximum hard-coded value of 256.

It is important to realize there are situations where it is not possible to enforce agreement between an intermediate stream and the input streams it is joined with, a scenario that would cause the Execution Planner to signal an error and reject the whole application. Fig. 5 illustrates one such case.

Tables and Stream-Table Joins

A recent addition to Samza is the introduction of Table, a key-value abstraction that facilitates accessing remotely stored data. And with this addition, it was also made possible to perform Join operations between Tables and streams.

The code sample below demonstrates how this can be achieved using Samza High-Level API.

public class StreamTableJoinApp implements StreamApplication {  
   @Override
    public void init(StreamGraph graph, Config config) {

      MessageStream s1 = graph.getInputStream("S1");
      MessageStream s1Prime = s1.partitionBy(/* Omitted for brevity */);

      // Assume local table
      Table t = graph.getTable(/* Omitted for brevity */);

      s1Prime.sendTo(t);

      MessageStream s2 = graph.getInputStream("S2");
      OutputStream s3 = graph.getOutputStream("S3");

      s2.join(t, /* Omitted for brevity */)
        .sendTo(s3);
    }
}

But since Tables can be populated with data flowing from input streams (aka local tables), it is still important to ensure that the stream used to populate the table has the same number of partitions as the stream the table is joined with. Failing to do so exposes Stream-Table Joins to the same class of problems Stream-Stream Joins could run into if Samza were to allow joining 2 streams with different partition counts, i.e. invalid Joins.

Side-Input Streams

Another recent addition to Samza that is related to Tables is Side-Input Streams. Simply put, this feature allows Samza application authors to specify that a table should be populated — and constantly updated — with data from one or more input streams. Such streams have been given the name Side-Input Streams or Side-Inputs for short.

As far as Samza’s Execution Planner is concerned, Tables with side-input Streams are not much different from other tables. They can be treated like other tables populated with data from a non-side-input stream. However, there is one key difference that requires treating tables with side-inputs differently. This difference is that, unlike all other streams, side-input streams are not represented in the OperatorSpec graph. At the time of writing, no InputOperatorSpecs — or any OperatorSpec for that matter — are created to represent side-input streams in the OperatorSpec graph. Only the names of side-input streams are specified in the configuration of the tables they populate. And since Samza’s Execution Planner currently relies on traversing the OperatorSpec graph to identify groups of input streams related through Joins, accounting for tables with side-inputs will require some special handling.

Problem Definition

At the time of writing, Samza’s Execution Planner does not account for Tables, which means it is simply incapable of fulfilling its responsibilities with Samza High-Level applications performing Stream-Table Joins.

In particular, assuming that we have:

• A table T

• An input stream S1, with partition count P1, used to populate table T

• An input stream S2, with partition count P2, joined with table T

The table below enumerates a number of cases in which Samza’s current Execution Planner does not enforce the necessary constraints on P1 and P2 to ensure the validity of the Stream-Table Join between table T and stream S2.

Cases #1 and #2 apply equally well if S1 is a side-input stream.

In all these cases, Samza application authors have no defense against essentially invalid Stream-Table Joins.

Problem Analysis

As explained in the Responsibilities of Samza’s Execution Planner section, Samza’s Execution Planner employs traversal of the OperatorSpec graph to identify groups of input and/or intermediate streams involved in Join operations and verify agreement among their partition counts. To understand the reason why this does not work in the case of Stream-Table Joins, we examine the OperatorSpec graph generated for the Samza application in the code sample below.

public class StreamTableJoinApp implements StreamApplication {  
   @Override
    public void init(StreamGraph graph, Config config) {
      Table<KV<Integer, String>> t = graph
        .getTable(/* Omitted for brevity */);

      MessageStream s1 = graph
          .getInputStream("S1")
          .filter(/* Omitted for brevity*/);
      
      s1.sendTo(t);  
      
      MessageStream s2 = graph.getInputStream("S2");

      OutputStream s3 = graph.getOutputStream("S3");

      s2.join(t, /* Omitted for brevity*/)
        .sendTo(s3);
    }
}

It is important to observe the following key differences between this graph and the graph representing the Stream-Stream Join application in Fig. 2:

1. A different OperatorSpec, StreamTableJoinOperator, is used to represent Stream-Table Join operations.

2. A new terminal SendToTableOperatorSpec is used to represent the operation of producing data to a table.

3. The StreamTableJoinOperatorSpec is not connected to the SendToTableOperatorSpec. It only has a reference to the table (TableSpec) participating in the Stream-Table Join operation.

The disconnect between the SendToTable and StreamTableJoin OperatorSpecs is attributed to the way the OperatorSpec graph gets constructed, whereby it is modeled exactly after the StreamApplication defined by the user, i.e.

• Chaining StreamGraph’s getInputStream(), filter(), and SendTo(Table) results in an InputOperatorSpec connected to a FilterOperatorSpec which is in turn connected to a SendToTableOperatorSpec.

• Chaining join(Table) and sendTo(Stream) results in a StreamTableJoinOperatorSpec connected to an OutputStreamSpec.

And since no further operations can be chained to sendTo(Table) according to the current Samza High-Level APIs, a SendToTableOperatorSpec is always a terminal vertex in the OperatorSpec graph, and all subsequent operations on a table will have their corresponding OperatorSpecs in disconnected components of the graph.

Contrasting the OperatorSpec graph generated in this scenario with that of the Stream-Stream Join application in listing 1 confirms this causation.

Proposed Changes

To extend Samza’s ExecutionPlanner to support Tables, we need to address the disconnect between a SendToTableOperatorSpec and all relevant StreamTableJoinOperatorSpecs. One possibility that does not require changing Samza’s High-Level APIs is to modify the OperatorSpec graph traversal such that virtual connections are assumed between every SendToTableOperatorSpec and all the StreamTableJoinOperatorSpecs that reference the same table (TableSpec) in the entire OperatorSpec graph.

As it stands at the time of writing, the implementation of Samza’s Execution Planner requires significant restructuring so it can be easily extended to support Tables. A separate document that explains the current state of the Execution Planner as well as the proposed changes is Modularizing Join Validation in Samza's Execution Planner.