Github user satishd commented on a diff in the pull request:
https://github.com/apache/storm/pull/1029#discussion_r50234893
--- Diff: storm-core/src/jvm/org/apache/storm/trident/Stream.java ---
@@ -68,61 +85,160 @@ protected Stream(TridentTopology topology, String
name, Node node) {
_node = node;
_name = name;
}
-
+
+ /**
+ * Applies a label to the stream. Naming a stream will append the
label to the name of the bolt(s) created by
+ * Trident and will be visible in the Storm UI.
+ *
+ * @param name - The label to apply to the stream
+ * @return
+ */
public Stream name(String name) {
return new Stream(_topology, name, _node);
}
-
+
+ /**
+ * Applies a parallelism hint to a stream.
+ *
+ * @param hint
+ * @return
+ */
public Stream parallelismHint(int hint) {
_node.parallelismHint = hint;
return this;
}
-
+
+ /**
+ * Filters out fields from a stream, resulting in a Stream containing
only the fields specified by `keepFields`.
+ *
+ * For example, if you had a Stream `mystream` containing the fields
`["a", "b", "c","d"]`, calling"
+ *
+ * ```java
+ * mystream.project(new Fields("b", "d"))
+ * ```
+ *
+ * would produce a stream containing only the fields `["b", "d"]`.
+ *
+ *
+ * @param keepFields The fields in the Stream to keep
+ * @return
+ */
public Stream project(Fields keepFields) {
projectionValidation(keepFields);
return _topology.addSourcedNode(this, new
ProcessorNode(_topology.getUniqueStreamId(), _name, keepFields, new Fields(),
new ProjectedProcessor(keepFields)));
}
+ /**
+ * ## Grouping Operation
+ *
+ * @param fields
+ * @return
+ */
public GroupedStream groupBy(Fields fields) {
projectionValidation(fields);
return new GroupedStream(this, fields);
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * @param fields
+ * @return
+ */
public Stream partitionBy(Fields fields) {
projectionValidation(fields);
return partition(Grouping.fields(fields.toList()));
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * @param partitioner
+ * @return
+ */
public Stream partition(CustomStreamGrouping partitioner) {
return
partition(Grouping.custom_serialized(Utils.javaSerialize(partitioner)));
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * Use random round robin algorithm to evenly redistribute tuples
across all target partitions
+ *
+ * @return
+ */
public Stream shuffle() {
return partition(Grouping.shuffle(new NullStruct()));
}
+ /**
+ * ## Repartitioning Operation
+ *
+ * Use random round robin algorithm to evenly redistribute tuples
across all target partitions, with a preference
+ * for local tasks.
+ *
+ * @return
+ */
public Stream localOrShuffle() {
return partition(Grouping.local_or_shuffle(new NullStruct()));
}
+
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * All tuples are sent to the same partition. The same partition is
chosen for all batches in the stream.
+ * @return
+ */
public Stream global() {
// use this instead of storm's built in one so that we can specify
a singleemitbatchtopartition
// without knowledge of storm's internals
return partition(new GlobalGrouping());
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * All tuples in the batch are sent to the same partition. Different
batches in the stream may go to different
+ * partitions.
+ *
+ * @return
+ */
public Stream batchGlobal() {
// the first field is the batch id
return partition(new IndexHashGrouping(0));
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * Every tuple is replicated to all target partitions. This can useful
during DRPC â for example, if you need to do
+ * a stateQuery on every partition of data.
+ *
+ * @return
+ */
public Stream broadcast() {
return partition(Grouping.all(new NullStruct()));
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * @return
+ */
public Stream identityPartition() {
return partition(new IdentityGrouping());
}
-
+
+ /**
+ * ## Repartitioning Operation
+ *
+ * This method takes in a custom partitioning function that implements
+ * {@link backtype.storm.grouping.CustomStreamGrouping}
--- End diff --
It should be org.apache.storm.grouping.CustomStreamGrouping
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
