forked from apache/samza
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
`Initial draft of top-level fluent API for operator DAGs Author: Yi Pan (Data Infrastructure) <nickpan47@gmail.com> Reviewers: Xinyu Liu <xiliu@linkedin.com>, Jacob Maes <jmaes@linkedin.com>, Prateek Maheshwari <pmaheshw@linkedin.com> Closes apache#51 from nickpan47/samza-fluent-api-v1 and squashes the following commits: 001be63 [Yi Pan (Data Infrastructure)] SAMZA-1073: Addressing review feedbacks. Change StreamGraphFactory to StreamGraphBuilder. 373048a [Yi Pan (Data Infrastructure)] SAMZA-1073: top-level fluent API `
- Loading branch information
Showing
73 changed files
with
3,080 additions
and
1,006 deletions.
There are no files selected for viewing
47 changes: 47 additions & 0 deletions
47
samza-api/src/main/java/org/apache/samza/operators/ContextManager.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.samza.operators; | ||
|
|
||
| import org.apache.samza.annotation.InterfaceStability; | ||
| import org.apache.samza.config.Config; | ||
| import org.apache.samza.task.TaskContext; | ||
|
|
||
|
|
||
| /** | ||
| * Interface class defining methods to initialize and finalize the context used by the transformation functions. | ||
| */ | ||
| @InterfaceStability.Unstable | ||
| public interface ContextManager { | ||
| /** | ||
| * The initialization method to create shared context for the whole task in Samza. Default to NO-OP | ||
| * | ||
| * @param config the configuration object for the task | ||
| * @param context the {@link TaskContext} object | ||
| * @return User-defined task-wide context object | ||
| */ | ||
| default TaskContext initTaskContext(Config config, TaskContext context) { | ||
| return context; | ||
| } | ||
|
|
||
| /** | ||
| * The finalize method to allow users to close resource initialized in {@link #initTaskContext} method. Default to NO-OP. | ||
| * | ||
| */ | ||
| default void finalizeTaskContext() { } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
41 changes: 41 additions & 0 deletions
41
samza-api/src/main/java/org/apache/samza/operators/OutputStream.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.samza.operators; | ||
|
|
||
| import org.apache.samza.annotation.InterfaceStability; | ||
| import org.apache.samza.operators.functions.SinkFunction; | ||
|
|
||
|
|
||
| /** | ||
| * The interface class defining the specific {@link SinkFunction} for a system {@link OutputStream}. | ||
| * | ||
| * @param <M> The type of message to be send to this output stream | ||
| */ | ||
| @InterfaceStability.Unstable | ||
| public interface OutputStream<M> { | ||
|
|
||
| /** | ||
| * Returns the specific {@link SinkFunction} for this output stream. The {@link OutputStream} is created | ||
| * via {@link StreamGraph#createOutStream(StreamSpec, Serde, Serde)} or {@link StreamGraph#createIntStream(StreamSpec, Serde, Serde)}. | ||
| * Hence, the proper types of serdes for key and value are instantiated and are used in the {@link SinkFunction} returned. | ||
| * | ||
| * @return The pre-defined {@link SinkFunction} to apply proper serdes before sending the message to the output stream. | ||
| */ | ||
| SinkFunction<M> getSinkFunction(); | ||
| } |
94 changes: 94 additions & 0 deletions
94
samza-api/src/main/java/org/apache/samza/operators/StreamGraph.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,94 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.samza.operators; | ||
|
|
||
| import org.apache.samza.annotation.InterfaceStability; | ||
| import org.apache.samza.operators.data.MessageEnvelope; | ||
| import org.apache.samza.serializers.Serde; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
|
|
||
| /** | ||
| * Job-level programming interface to create an operator DAG and run in various different runtime environments. | ||
| */ | ||
| @InterfaceStability.Unstable | ||
| public interface StreamGraph { | ||
| /** | ||
| * Method to add an input {@link MessageStream} from the system | ||
| * | ||
| * @param streamSpec the {@link StreamSpec} describing the physical characteristics of the input {@link MessageStream} | ||
| * @param keySerde the serde used to serialize/deserialize the message key from the input {@link MessageStream} | ||
| * @param msgSerde the serde used to serialize/deserialize the message body from the input {@link MessageStream} | ||
| * @param <K> the type of key in the input message | ||
| * @param <V> the type of message in the input message | ||
| * @param <M> the type of {@link MessageEnvelope} in the input {@link MessageStream} | ||
| * @return the input {@link MessageStream} object | ||
| */ | ||
| <K, V, M extends MessageEnvelope<K, V>> MessageStream<M> createInStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde); | ||
|
|
||
| /** | ||
| * Method to add an output {@link MessageStream} from the system | ||
| * | ||
| * @param streamSpec the {@link StreamSpec} describing the physical characteristics of the output {@link MessageStream} | ||
| * @param keySerde the serde used to serialize/deserialize the message key from the output {@link MessageStream} | ||
| * @param msgSerde the serde used to serialize/deserialize the message body from the output {@link MessageStream} | ||
| * @param <K> the type of key in the output message | ||
| * @param <V> the type of message in the output message | ||
| * @param <M> the type of {@link MessageEnvelope} in the output {@link MessageStream} | ||
| * @return the output {@link MessageStream} object | ||
| */ | ||
| <K, V, M extends MessageEnvelope<K, V>> OutputStream<M> createOutStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde); | ||
|
|
||
| /** | ||
| * Method to add an intermediate {@link MessageStream} from the system | ||
| * | ||
| * @param streamSpec the {@link StreamSpec} describing the physical characteristics of the intermediate {@link MessageStream} | ||
| * @param keySerde the serde used to serialize/deserialize the message key from the intermediate {@link MessageStream} | ||
| * @param msgSerde the serde used to serialize/deserialize the message body from the intermediate {@link MessageStream} | ||
| * @param <K> the type of key in the intermediate message | ||
| * @param <V> the type of message in the intermediate message | ||
| * @param <M> the type of {@link MessageEnvelope} in the intermediate {@link MessageStream} | ||
| * @return the intermediate {@link MessageStream} object | ||
| */ | ||
| <K, V, M extends MessageEnvelope<K, V>> OutputStream<M> createIntStream(StreamSpec streamSpec, Serde<K> keySerde, Serde<V> msgSerde); | ||
|
|
||
| /** | ||
| * Method to get the input {@link MessageStream}s | ||
| * | ||
| * @return the input {@link MessageStream} | ||
| */ | ||
| Map<StreamSpec, MessageStream> getInStreams(); | ||
|
|
||
| /** | ||
| * Method to get the {@link OutputStream}s | ||
| * | ||
| * @return the map of all {@link OutputStream}s | ||
| */ | ||
| Map<StreamSpec, OutputStream> getOutStreams(); | ||
|
|
||
| /** | ||
| * Method to set the {@link ContextManager} for this {@link StreamGraph} | ||
| * | ||
| * @param manager the {@link ContextManager} object | ||
| * @return this {@link StreamGraph} object | ||
| */ | ||
| StreamGraph withContextManager(ContextManager manager); | ||
|
|
||
| } |
38 changes: 38 additions & 0 deletions
38
samza-api/src/main/java/org/apache/samza/operators/StreamGraphBuilder.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,38 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.samza.operators; | ||
|
|
||
| import org.apache.samza.annotation.InterfaceStability; | ||
| import org.apache.samza.config.Config; | ||
|
|
||
|
|
||
| /** | ||
| * This interface defines a factory class that user will implement to create user-defined operator DAG in a {@link StreamGraph} object. | ||
| */ | ||
| @InterfaceStability.Unstable | ||
| public interface StreamGraphBuilder { | ||
| /** | ||
| * Users are required to implement this abstract method to initialize the processing logic of the application, in terms | ||
| * of a DAG of {@link org.apache.samza.operators.MessageStream}s and operators | ||
| * | ||
| * @param graph an empty {@link StreamGraph} object to be initialized | ||
| * @param config the {@link Config} of the application | ||
| */ | ||
| void init(StreamGraph graph, Config config); | ||
| } |
Oops, something went wrong.