You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/10/22 20:31:04 UTC
[GitHub] [flink] sjwiesman opened a new pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
sjwiesman opened a new pull request #13752:
URL: https://github.com/apache/flink/pull/13752
## What is the purpose of the change
Promote `DataStreamUtils#collect` to a `DataStream#collect`.
## Verifying this change
UT
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / **no**)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (**yes** / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / **no** / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (yes / **no** / don't know)
- The S3 file system connector: (yes / **no** / don't know)
## Documentation
- Does this pull request introduce a new feature? (**yes** / no)
- If yes, how is the feature documented? (not applicable / docs / **JavaDocs** / not documented)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f621e5d663256a794e2bf416e197bd9ad605959d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f621e5d663256a794e2bf416e197bd9ad605959d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] sjwiesman commented on a change in pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
sjwiesman commented on a change in pull request #13752:
URL: https://github.com/apache/flink/pull/13752#discussion_r511956455
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
##########
@@ -1294,6 +1303,92 @@ public ExecutionConfig getExecutionConfig() {
return sink;
}
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect() throws Exception {
+ return executeAndCollect("DataStream Collect");
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect(String jobExecutionName) throws Exception {
+ return executeAndCollectWithClient(jobExecutionName).iterator;
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(int limit) throws Exception {
+ return executeAndCollect("DataStream Collect", limit);
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(String jobExecutionName, int limit) throws Exception {
+ Preconditions.checkState(limit > 0, "Limit must be greater than 0");
+
+ ClientAndIterator<T> clientAndIterator = executeAndCollectWithClient(jobExecutionName);
+
+ try {
+ List<T> results = new ArrayList<>(limit);
+ while (clientAndIterator.iterator.hasNext() && limit > 0) {
+ results.add(clientAndIterator.iterator.next());
+ limit--;
+ }
+
+ return results;
+ } finally {
+ clientAndIterator.iterator.close();
Review comment:
Yeah, that's better
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
}, {
"hash" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8312",
"triggerID" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 975743e788ccde0f3d1cc62b8a1c8c719c852ba4 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8312)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
}, {
"hash" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9cc1022536ed63c253fb890ba55c631c7bdb9c54 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204)
* 975743e788ccde0f3d1cc62b8a1c8c719c852ba4 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f621e5d663256a794e2bf416e197bd9ad605959d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134)
* 9cc1022536ed63c253fb890ba55c631c7bdb9c54 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9cc1022536ed63c253fb890ba55c631c7bdb9c54 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] sjwiesman commented on a change in pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
sjwiesman commented on a change in pull request #13752:
URL: https://github.com/apache/flink/pull/13752#discussion_r511957050
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
##########
@@ -1294,6 +1303,92 @@ public ExecutionConfig getExecutionConfig() {
return sink;
}
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect() throws Exception {
+ return executeAndCollect("DataStream Collect");
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect(String jobExecutionName) throws Exception {
+ return executeAndCollectWithClient(jobExecutionName).iterator;
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(int limit) throws Exception {
+ return executeAndCollect("DataStream Collect", limit);
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(String jobExecutionName, int limit) throws Exception {
+ Preconditions.checkState(limit > 0, "Limit must be greater than 0");
+
+ ClientAndIterator<T> clientAndIterator = executeAndCollectWithClient(jobExecutionName);
+
+ try {
+ List<T> results = new ArrayList<>(limit);
+ while (clientAndIterator.iterator.hasNext() && limit > 0) {
+ results.add(clientAndIterator.iterator.next());
+ limit--;
+ }
+
+ return results;
+ } finally {
+ clientAndIterator.iterator.close();
+ clientAndIterator.client.cancel();
+ }
+ }
+
+ private ClientAndIterator<T> executeAndCollectWithClient(String jobExecutionName) throws Exception {
Review comment:
`DataStream` depending on `DataStreamUtils` seemed dirty to me. I realized I could have the dependency go the other way so I removed the duplication.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
}, {
"hash" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8312",
"triggerID" : "975743e788ccde0f3d1cc62b8a1c8c719c852ba4",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9cc1022536ed63c253fb890ba55c631c7bdb9c54 Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204)
* 975743e788ccde0f3d1cc62b8a1c8c719c852ba4 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8312)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f621e5d663256a794e2bf416e197bd9ad605959d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714745586
Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.
## Automated Checks
Last check on commit f621e5d663256a794e2bf416e197bd9ad605959d (Thu Oct 22 20:34:02 UTC 2020)
**Warnings:**
* No documentation files were touched! Remember to keep the Flink docs up to date!
<sub>Mention the bot in a comment to re-run the automated checks.</sub>
## Review Progress
* ❓ 1. The [description] looks good.
* ❓ 2. There is [consensus] that the contribution should go into to Flink.
* ❓ 3. Needs [attention] from.
* ❓ 4. The change fits into the overall [architecture].
* ❓ 5. Overall code [quality] is good.
Please see the [Pull Request Review Guide](https://flink.apache.org/contributing/reviewing-prs.html) for a full explanation of the review process.<details>
The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required <summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot approve description` to approve one or more aspects (aspects: `description`, `consensus`, `architecture` and `quality`)
- `@flinkbot approve all` to approve all aspects
- `@flinkbot approve-until architecture` to approve everything until `architecture`
- `@flinkbot attention @username1 [@username2 ..]` to require somebody's attention
- `@flinkbot disapprove architecture` to remove an approval you gave earlier
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] aljoscha commented on a change in pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
aljoscha commented on a change in pull request #13752:
URL: https://github.com/apache/flink/pull/13752#discussion_r511978002
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/ClientAndIterator.java
##########
@@ -0,0 +1,48 @@
+/*
+ * 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.flink.streaming.api.operators.collect;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.core.execution.JobClient;
+import org.apache.flink.util.CloseableIterator;
+
+import java.util.Iterator;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * A pair of an {@link Iterator} to receive results from a streaming application and a
+ * {@link JobClient} to interact with the program.
+ */
+@Internal
+public final class ClientAndIterator<E> implements AutoCloseable {
+
+ public final JobClient client;
+ public final CloseableIterator<E> iterator;
+
+ public ClientAndIterator(JobClient client, CloseableIterator<E> iterator) {
+ this.client = checkNotNull(client);
+ this.iterator = checkNotNull(iterator);
+ }
+
+ @Override
+ public void close() throws Exception {
+ iterator.close();
Review comment:
Please ignore this nit, but I think it can happen that `close()` throws and then `client.cancel()` will not be called.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] sjwiesman commented on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
sjwiesman commented on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-716544930
Thanks for taking a look, I think I've addressed all the commends.
@aljoscha you know there's nothing more important to me than a well-loved scala api.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13752:
URL: https://github.com/apache/flink/pull/13752#issuecomment-714761274
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134",
"triggerID" : "f621e5d663256a794e2bf416e197bd9ad605959d",
"triggerType" : "PUSH"
}, {
"hash" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204",
"triggerID" : "9cc1022536ed63c253fb890ba55c631c7bdb9c54",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* f621e5d663256a794e2bf416e197bd9ad605959d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8134)
* 9cc1022536ed63c253fb890ba55c631c7bdb9c54 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=8204)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] aljoscha commented on a change in pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
aljoscha commented on a change in pull request #13752:
URL: https://github.com/apache/flink/pull/13752#discussion_r511910670
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
##########
@@ -1294,6 +1303,92 @@ public ExecutionConfig getExecutionConfig() {
return sink;
}
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect() throws Exception {
+ return executeAndCollect("DataStream Collect");
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect(String jobExecutionName) throws Exception {
+ return executeAndCollectWithClient(jobExecutionName).iterator;
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(int limit) throws Exception {
+ return executeAndCollect("DataStream Collect", limit);
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(String jobExecutionName, int limit) throws Exception {
+ Preconditions.checkState(limit > 0, "Limit must be greater than 0");
+
+ ClientAndIterator<T> clientAndIterator = executeAndCollectWithClient(jobExecutionName);
+
+ try {
+ List<T> results = new ArrayList<>(limit);
+ while (clientAndIterator.iterator.hasNext() && limit > 0) {
+ results.add(clientAndIterator.iterator.next());
+ limit--;
+ }
+
+ return results;
+ } finally {
+ clientAndIterator.iterator.close();
Review comment:
Makes me wonder why `ClientAndIterator` isn't `AutoCloseable` and does these two.
##########
File path: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
##########
@@ -1294,6 +1303,92 @@ public ExecutionConfig getExecutionConfig() {
return sink;
}
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect() throws Exception {
+ return executeAndCollect("DataStream Collect");
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ *
+ *<p><b>IMPORTANT</b> The returned iterator must be closed to free all cluster resources.
+ */
+ public CloseableIterator<T> executeAndCollect(String jobExecutionName) throws Exception {
+ return executeAndCollectWithClient(jobExecutionName).iterator;
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(int limit) throws Exception {
+ return executeAndCollect("DataStream Collect", limit);
+ }
+
+ /**
+ * Triggers the distributed execution of the streaming dataflow and returns an iterator over the elements
+ * of the given DataStream.
+ *
+ * <p>The DataStream application is executed in the regular distributed manner on the target environment,
+ * and the events from the stream are polled back to this application process and thread through
+ * Flink's REST API.
+ */
+ public List<T> executeAndCollect(String jobExecutionName, int limit) throws Exception {
+ Preconditions.checkState(limit > 0, "Limit must be greater than 0");
+
+ ClientAndIterator<T> clientAndIterator = executeAndCollectWithClient(jobExecutionName);
+
+ try {
+ List<T> results = new ArrayList<>(limit);
+ while (clientAndIterator.iterator.hasNext() && limit > 0) {
+ results.add(clientAndIterator.iterator.next());
+ limit--;
+ }
+
+ return results;
+ } finally {
+ clientAndIterator.iterator.close();
+ clientAndIterator.client.cancel();
+ }
+ }
+
+ private ClientAndIterator<T> executeAndCollectWithClient(String jobExecutionName) throws Exception {
Review comment:
You're duplicating the code here because we want to remove `DataStreamUtils` in the future?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] sjwiesman closed pull request #13752: [FLINK-19508][DataStream] Add collect() operation on DataStream
Posted by GitBox <gi...@apache.org>.
sjwiesman closed pull request #13752:
URL: https://github.com/apache/flink/pull/13752
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org