Status
Current state: UNDER DISCUSSIONACCEPTED
Discussion thread:
JIRA: Jira server ASF JIRA serverId 5aa69414-a9e9-3523-82ec-879b028fb15b key SAMZA-1804
Motivation
Samza jobs using the High Level API currently specify their inputs and outputs partly in code (i.e., the streamIds), and partly in configuration (e.g., system name, physical name and everything else). Complexity of configuring these inputs and outputs has been a long standing issue for Samza users. We want to allow users to describe their system, inputs, outputs and their properties entirely in code. To do this, we propose new System and Stream "descriptors" APIs. Once the new Now that StreamApplication/TaskApplication APIs for Samza are available (SEP-13), these will be usable may be used for describing inputs and outputs in the Low Level API as well.
...
- Create a new SystemDescriptor instance and optionally configure it.
- Create Obtain a new InputDescriptor<I> and OutputDescriptor<O> using the SystemDescriptor above and optionally configure them.
- Call StreamGraph#getInputStreamCall StreamApplicationDescriptor#getInputStream(InputDescriptor<I>) to get the MessageStream<I> and StreamGraph#getOutputStreamthe MessageStream<I> and StreamApplicationDescriptor#getOutputStream(OutputDescriptor<O>) to get the OutputStream<O>.
- Apply transforms on the input MessageStream and produce the results to OutputStream.
...
Users will not be required to describe everything in code - they'll have an option to do so. They may describe as much of their their system/stream properties in code as they like, and the rest may be in configuration. System/Stream properties from user provided descriptors will be converted to their corresponding configurations and merged with the user provided configuration. Descriptor generated properties will override those provided by the user in configuration.
For simple cases, where system configuration is either unnecessary or provided by default, System implementers may provide convenience factory methods for Input/OutputDescriptors so that they can be obtained using just the system name. For example;
Code Block | ||||
---|---|---|---|---|
| ||||
KafkaInputDescriptor<KV<String, GenericRecord>> pvd = KafkaInputDescriptor.from("PageViewEvent", "tracking"); // properties for the "tracking" system are provided in configuration
KafkaOutputDescriptor<KV<String, PageViewCount>> pvcd = KafkaInputDescriptor.from("PageViewCount", "metrics"); // properties for the "metrics" system are provided in configuration
MessageStream<KV<String, GenericRecord>> pvs = graph.getInputStream(pvd);
OutputStream<KV<String, PageViewCount>> pvcs = graph.getOutputStream(pvcd);
pvs.window(...).map(...).sendTo(pvcs); |
Basic requirements
The basic requirements for the new APIs are:
- Users should be able to set any Samza properties for their input and output streams using these descriptors (e.g., bootstrap status, input priority, offset management, serdes etc.).
- System implementers should be able to provide custom implementations of these descriptors that support additional system-specific properties. For example, a KafkaSystemDescriptor may allow setting Kafka producer and consumer properties.
Users should chose the extended descriptor for their system if one is available. Else, they may use the Samza-provided GenericSystemDescriptor/GenericInputDescriptor/GenericOutputDescriptor implementations and provide system specific properties as the appropriate key-value configurations. - Users should be able to use a fluent API for configuring these descriptors.
- The types of MessageStream/OutputStream created using these descriptors should match their contents. E.g., for simple descriptors, the type of Message/OutputStream should be determined by the system or stream level Serdes, whether provided by the system as a default, or configured by the user explicitly.
- Users should be able to provide a default SystemDescriptor which should be used to configure the properties of the system used for creating intermediate streams in the application. This is equivalent to setting and configuring job.default.system and its properties.
Advanced requirements
We will allow System implementers to extend these Descriptor implementations so that they can provide the following additional features to users:
System Specific Serdes
Basic requirements
The basic requirements for the new APIs are:
- Users should be able to set any Samza properties for their input and output streams using these descriptors (e.g., bootstrap status, input priority, offset management, serdes etc.).
- System implementers should be able to provide custom implementations of these descriptors that support additional system-specific properties. For example, a KafkaSystemDescriptor may allow setting Kafka producer and consumer properties.
Users should chose the extended descriptor for their system if one is available. Else, they may use the Samza-provided GenericSystemDescriptor/GenericInputDescriptor/GenericOutputDescriptor implementations and provide system specific properties as the appropriate key-value configurations. - Users should be able to use a fluent API for configuring these descriptors.
- The types of MessageStream/OutputStream created using these descriptors should match their contents. E.g., for simple descriptors, the type of Message/OutputStream should be determined by the system or stream level Serdes, whether provided by the system as a default, or configured by the user explicitly.
- Users should be able to provide a default SystemDescriptor which should be used to configure the properties of the system used for creating intermediate streams in the application. This is equivalent to setting and configuring job.default.system and its properties.
Advanced requirements
We will allow System implementers to extend these Descriptor implementations so that they can provide the following additional features to users:
System Specific Input/OutputDescriptor Getters
System implementers should be able to provide convenience methods to get Input/OutputDescriptors for their users.
For example, for simple cases system configuration is either unnecessary or provided by default, users should be able to obtain an input descriptor using just the system name:
Code Block | ||||
---|---|---|---|---|
| ||||
KafkaInputDescriptor<KV<String, GenericRecord>> pvd = KafkaInputDescriptor.from("PageViewEvent", "tracking"); // properties for the "tracking" system are provided in configuration
KafkaOutputDescriptor<KV<String, PageViewCount>> pvcd = KafkaInputDescriptor.from("PageViewCount", "tracking"); // properties for the "metrics" system are provided in configuration
MessageStream<KV<String, GenericRecord>> pvs = graph.getInputStream(pvd);
OutputStream<KV<String, PageViewCount>> pvcs = graph.getOutputStream(pvcd);
pvs.window(...).map(...).sendTo(pvcs); |
Similarly, System implementers should be able to dictate a specific serialization format for messages in their streams. For example, if a system enforces Avro as the message format, its , users should be able to obtain an Input/OutputDescriptor without providing an explicit serde, and the input and output descriptors should have a KV<String, GenericRecord> message type by default. The API will always allow users to provide a Stream level serde for input/output streams that may be different than the System specific serde. System implementors may choose to disallow this by throwing RuntimeException in these methods
Also, System implementers of systems that only support either input or output, but not both, should be able to disallow creating input or output descriptors.
Input Transformers
InputTransformer functions allow System implementers to apply custom transformations to messages in their input streams before they're delivered to the applicationHigh Level API StreamApplication. The primary use case of this feature is to allow systems to add additional metadata to each incoming message by extending IncomingMessageEnvelope. In this use case, the transformer is provided by the System implementors as a system level default, and is not visible to the user.
...
The InputTransformer function is applied at runtime in InputOperatorImpl on each IncomingMessageEnvelope after the key and value have been deserialized. For a InputDescriptor with a InputTransformer, the type of messages in their MessageStreams will always be the result type of the InputTransformer function. I.e., providing stream level serdes should not change the type of the MessageStream generated from the descriptor, since the InputTransformer is applied after deserialization and overrides the Serde type.
Users may optionally override the system default transformer or provide one for an input stream themselves. This might be useful, for example, if they need access to some information in the IncomingMessageEnvelope like the offset or the message timestampWe will not support user-provided stream level transformers in the initial API, but we'll keep the option of doing so in future available.
Stream Expanders
StreamExpander functions allow System implementors to expand a single user-provided input descriptor into multiple input descriptors, apply some custom MessageStream transforms on the expanded input streams thus obtained, and then return a new MessageStream to the application High Level API StreamApplication for further processing. The primary use case of this feature is to allow system implementors to resolve a "logical" user provided input stream to multiple underlying "physical" input stream sources at execution planning time and still present a unified logical view to the user for further processing. The interface is:
Code Block | ||||
---|---|---|---|---|
| ||||
/** * Expands the provided {@link InputDescriptor} to a sub-DAG of one or more operators on the {@link StreamGraphStreamApplicationDescriptor}, * and returns a new {@link MessageStream} with the combined results. Called when {@link StreamGraph#getInputStreamStreamApplicationDescriptor#getInputStream} * is being used to get a {@link MessageStream} using an {@link InputDescriptor} from an * {@link org.apache.samza.operators.descriptors.base.system.ExpandingSystemDescriptor} * <p> * This is provided by default by {@link org.apache.samza.operators.descriptors.base.system.ExpandingSystemDescriptor} * implementations and can not be overridden or set on a per stream level. * * @param <OM> type of the messages in the resultant {@link MessageStream} */ public interface StreamExpander<OM> extends Serializable { /** * Expands the provided {@link InputDescriptor} to a sub-DAG of one or more operators on the {@link StreamGraphStreamApplicationDescriptor}, * and returns a new {@link MessageStream} with the combined results. Called when the {@link InputDescriptor} * is being used to get an {@link MessageStream} using {@link StreamGraph#getInputStreamStreamApplicationDescriptor#getInputStream}. * <p> * Note: Take care to avoid infinite recursion in the implementation; e.g., by ensuring that it doesn't call * {@link StreamGraph#getInputStreamStreamApplicationDescriptor#getInputStream} with an {@link InputDescriptor} from the an * {@link org.apache.samza.operators.descriptors.base.system.ExpandingSystemDescriptor} (like this one) again. * * @param streamGraphStreamApplicationDescriptor the {@link StreamGraphStreamApplicationDescriptor} to register the expanded sub-DAG of operators on * @param inputDescriptor the {@link InputDescriptor} to be expanded * @return the {@link MessageStream} containing the combined results of the sub-DAG of operators */ MessageStream<OM> apply(StreamGraphStreamApplicationDescriptor streamGraphStreamApplicationDescriptor, InputDescriptor inputDescriptor); } |
Unlike InputTransformers, these functions cannot be provided or overridden by users. These functions These functions are executed once in the ApplicationRunner when creating/describing the StreamGraphStreamApplication.
We do not plan to support user provided stream level StreamExpanders.
Similar to a SystemDescriptor with a InputTransformer, the type of the messages in their MessageStream (but not OutputStream) obtained from their InputDescriptors will always be the result type of the StreamExpander. Same parameterization rules apply.
...
A Fluent API for describing inputs and outputs: This requires each super class method to return an instance of the concrete sub-class it was called on for further chaining. This means that each super class needs to carry the type of its extending class in its type parameters. E.g.:
Code Block language java title Descriptor Sub-type public abstract class SystemDescriptor<SystemMessageType, SubClass extends SystemDescriptor<SystemMessageType, SubClass>> public final class GenericSystemDescriptor<SystemMessageType> extends SystemDescriptor<SystemMessageType, GenericSystemDescriptor<SystemMessageType>>
Support both System and Stream level serdes: If users or system implementers want to set a default serde for a SystemDescriptor, any Input/OutputDescriptors created from it should inherit the default serde type. However Samza also They may or may not allows users to provide stream level serdes that override the system level serde. If users provide a Stream level serde, the Input/OutputDescriptors should have the same type as the stream level serde. The simplest way to do this is to require stream-level serdes to be provided while creating the Input/OutputDescriptors.
Code Block language java title BaseSystemDescriptor API class SystemDescriptor<SystemMessageType> { public InputDescriptor<A> getInputDescriptor(String id) // case 1 public InputDescriptor<B> getInputDescriptor(String id, Serde<StreamMessageType> s) // case 2 }
Stream level serde should not affect InputDescriptor type when used with StreamTransformerInputTransformer. This is in conflict with (or an exception to) the requirement above. For a InputDescriptor with a InputTransformer, the type of messages in their MessageStreams will always be the result type of the InputTransformer function. I.e., providing system/stream level serdes should not affect the type of the descriptor, since the InputTransformer is applied after deserialization and overrides the Serde type. User provided StreamTransformer overrides System provided default StreamTransformer.
Code Block class SystemDescriptor<SystemMessageType> { ... // 1 and 2 as above, plus: public InputDescriptor<C> getInputDescriptor(String id, InputTransformer<StreamTransformerType> t) // case 3 public InputDescriptor<D> getInputDescriptor(String id, InputTransformer<StreamTransformerType> t, Serde<StreamMessageType> s) // case 4 }
While we will consider the use case of stream-level InputTransformer functions in the API design here, we can choose to will not add them to the API until we have concrete use casesAPI in the initial version. This will simplify the SystemDescriptor so that users have 1-2 ways of getting an input stream from a system descriptor instead of 4.
- System or stream level serde should not affect InputDescriptor type when used with StreamExpander. The type of such InputDescriptors will always be the result type of the StreamExpander function.
...
In other words, the type patterns above are:
getISD(id) | getISD(id, serde) | getISD(id, txfn) | getISD(id, txfn, serde) | |
---|---|---|---|---|
Non Transforming System | SystemMessageType | StreamMessageType | StreamTransformerType | StreamTransformerType |
Transforming System | SystemTransformerType | SystemTransformerType | StreamTransformerType | StreamTransformerType |
Expanding System | SystemExpanderType | SystemExpanderType | SystemExpanderType | SystemExpanderType |
Expanding Transforming System | SystemExpanderType | SystemExpanderType | SystemExpanderType | SystemExpanderType |
Clearly, there are no single SystemDescriptor class that can satisfy the type constraints for all four cases. In fact, there are 3 different sets of interface method signatures above (Case 1-3).
...
Note that if we add new features that affects the message type of InputDescriptors, we may need yet another set of getInputStream methods. In general there will have to be 2^n ways of getting an InputDescriptor to support all combinations of features, where n is the number of features where the type of the input stream is affected by the feature. Currently n == 1 if we only support stream level serde overrides, or 2 if we support stream level transform functions as well. This might be a concern for future API evolution, but its not as bad as it appears to be because:
...
To satisfy the requirements above, we propose the following class hierarchy of System and Stream Descriptors.
Base Classes:
SystemDescriptor
SimpleSystemDescriptor (Case 1: supports default system level serdes)
TransformingSystemDescriptor (Case 2: supports default system level serdes and default system level transformer functions)
ExpandingSystemDescriptor (Case 3, 4: supports default system level serdes, default system level transformer functions and default system level input expansion functions)- StreamDescriptor
We also provide the following generic implementations of System, InputStream and OutputStream Descriptors. If a system doesn't provide a more specific implementation, users can use these to configure any samza properties and provide the system specific properties as a Map<String, String> of configs.
Generic (System Agnostic) Implementations:
...
- StreamDescriptor
- InputDescriptor
- OutputDescriptor
To provide flexibility to system implementers to customize whether and how users get input and output descriptors (e.g. advanced requirement 1), we do not enforce a particular API for getInputDescriptor and getOutputDescriptor. System implementers are free to extend the base SystemDescriptor class and add any getInputDescriptor/getOutputDescriptor variants. We do provide the following interfaces as guidelines for System implementers based on the discussion above:
OutputDescriptorProvider: Interface for simple SystemDescriptors that return OutputDescriptors parameterized by the type of the provided stream level serde.
SimpleInputDescriptorProvider: Interface for simple SystemDescriptors that return InputDescriptors parameterized by the type of the provided stream level serde. (Case 1 above)
- TransformingInputDescriptorProvider: Interface for advanced SystemDescriptors that constrain the type of returned InputDescriptors to their own InputTransformer or StreamExpander function result types. Note that this interface is used for both transforming and expanding system descriptors for now, since in the absence of stream level transformer functions the method signature for getInputStream(id, serde) is the same in both cases. We may split this into two interfaces when we support stream level transformer functions.
We also provide the following generic implementations of System, InputStream and OutputStream Descriptors. If a system doesn't provide a more specific implementation, users can use these to configure any Samza properties and provide the system specific properties as a Map<String, String> of configs.
Generic (System Agnostic) Implementations:
- GenericSystemDescriptor
- GenericInputDescriptor
- GenericOutputDescriptor
The new StreamApplicationDescriptor
...
These APIs (WIP) for the new descriptors can be found at: https://github.com/prateekm/samza/tree/stream-descriptor/samza-api/src/main/java/org/apache/samza/operators/descriptors
Sample High Level API code (WIP) using these APIs can be found at: https://github.com/prateekm/samza/tree/stream-descriptor/samza-test/src/main/java/org/apache/samza/example
Sample implementations and usage patterns for these APIs (WIP) can be found at: https://github.com/prateekm/samza/tree/stream-descriptor/samza-api/src/test/java/org/apache/samza/operators/descriptors
The new StreamGraph interface with these classes is:
Code Block |
---|
public interface StreamGraphStreamApplicationDescriptor { void setDefaultSystem(SystemDescriptor<?, ?> defaultSystemDescriptor); <M> MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor); <M> OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor); <K, V> Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor); StreamGraph withContextManager(ContextManager contextManager); } |
Default System Descriptor and Intermediate Streams
The "default system" for a job, currently specified by setting the "job.default.system" configuration and its properties, is used for creating intermediate streams (e.g. for partitionBy and broadcast), as well as some internal streams like coordinator, checkpoint and store changelog streams (, unless their system has been explicitly set using job.coordinator.system, task.checkpoint.system, or job.changelog.system respectively).
We will add a new StreamGraph#setDefaultSystemApplicationDescriptor#setDefaultSystem(SystemDescriptor) API that can be used to configure the default system used by the job. For example, to provide the default system name and set and configure the default system serde. Like used by the job. Like everything else in the Descriptors API, user should be able to set default system properties in config instead of in code, including the serdes, if they wish to. If they choose to rely on the default system's serde or setting the serde for job.default.system in configs instead of providing one explicitly when using partitionBy, they'll be responsible for figuring out the type of messages in the repartitioned streams and casting them correctly in their applicationin config instead of in code if they wish to.
We will not provide a way to set intermediate stream properties in code using a StreamDescriptor since we don't expect this to be a common use case. If users need to manually configure them, they'll have to find the intermediate stream name and set its properties in configuration.
...
This implies that, unlike other configurations, input/output stream serdes *must* be specified using descriptors, and cannot be specified in configuration. Users and SystemDescriptors may specify an KVSerde<NoOpSerde, NoOpSerde> if they don't want messages in the system/stream to be deserialized by Samza. This is may be useful when the System ConsumerSystemConsumer/Producer handle handles the serialization themselves.
itself.
Going forward, serdes will be required for intermediate streamsSerdes are optional for intermediate streams. If users don't provide an explicit serde for MessageStream#partitionBy, and don't set a default system for the application, they must set the serde for job.default.system and its system-level serdes in their configuration instead. If a default system descriptor is set and its serde is used for intermediate streams, it must be a KVSerde.
Stream Configuration Generation
During StreamApplication#init(StreamGraphStreamApplicationDescriptor), users create System and StreamDescriptors, and get input/output streams for them using StreamGraph#getInputStreamStreamApplicationDescriptor#getInputStream(StreamDescriptor). The StreamGraph tracks StreamApplicationDescriptor tracks all Input/OutputDescriptors and the default SystemDescriptor used by the application. When the planner runs, it generates configurations for the input output streams in the following way:
...
Step 1: Create an instance of the StreamGraph and StreamApplicationDescriptor and the StreamApplication, and call StreamApplication#init(StreamGraphStreamApplicationDescriptor)
Step 2: Collect all the user provided System/StreamDescriptors from the StreamGraphStreamApplicationDescriptor.
Step 3: Merge the user provided configuration (application.cfg) with the Descriptor generated configuration (System/Input/OutputDescriptor#toConfig). Descriptor generated configuration has higher precedence and overrides user provided : Merge the user provided configuration with the Descriptor generated configuration.
Step 4: Run the planning phase using this merged configuration. The planner internally creates StreamSpec instances based on this merged configuration, mutates these StreamSpecs to set additional inferred stream properties, creates/validates internal streams using them, and then merges StreamSpec generated configuration with previously merged configuration from Step 3. StreamSpec generated configuration has higher precedence.
...
Note: System and Stream descriptor generated configuration have higher precedence than user provided configuration (from cfg2). We will need to offer a mechanism to override these descriptor generated configs in cfg2. This is useful, for example, to support fabric/instance level overrides. This can currently be done by specifying them with a special jobs.job-idname.* override scope. For example, the value for config "foo.bar" generated in step 4 can be overridden using the config "jobs.job-idname.foo.bar".
Note: In the case where there are "expanded" streams, configs will be generated for the "expanded" streams only, not the initial logical stream.
...
For these reasons, if the user hasn't explicitly provided a default value for a property in the descriptor, the descriptor will not generate any configuration for the property, not even the default value. In the example above, if the user has not explicitly set bootstrap status for a stream as true or false, the correct behavior is for InputDescriptor#toConfig() to not generate the "streams.stream-id.samza.bootstrap" configuration at all, not to set it to false.
...
Compatibility, Deprecation, and Migration Plan
The changes in the StreamGraph StreamApplicationDescriptor interface are backwards incompatible. Specifically, instead of getting an input/output stream using a streamId, users will now have to get an input/output stream using StreamDescriptors.
The migration is as simple as changing:
...
SystemDescriptors and StreamDescriptors.
Rest of the code remains the same, and users can migrate to specifying more of their input/output properties using descriptors instead of configuration over time.