You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2020/12/20 13:18:19 UTC
[GitHub] [beam] Fokko opened a new pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Fokko opened a new pull request #13585:
URL: https://github.com/apache/beam/pull/13585
**Please** add a meaningful description for your change here
Exploring how to write Iceberg tables using Beam.
By removing the PDone, and emit the files that are written, we can add these to the Table Format, such as Iceberg.
Exploring the possible approaches, but I think this is a very fruitful direction since we can fully re-use the existing Avro/Parquet/.. writers, and emit the WriteFilesResult to the next operator, which will append the new files to the Iceberg log.
My suggestion would be to change the API in Apache Beam so we can use the `WriteFilesResult`. Add the Iceberg extension in the Iceberg repository itself. This will be a `PTransform<WriteFilesResult<?>, PDone>`. Maybe I'll change the `PDone` to a `Table` (Iceberg Table) as well. So you can signal downstream consumers. I'll open this PR somewhere next week, but would like to know your idea's as well! :)
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
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] [beam] Fokko commented on pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #13585:
URL: https://github.com/apache/beam/pull/13585#issuecomment-748868672
R: @iemejia 👍
----------------------------------------------------------------
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] [beam] TheNeuralBit commented on a change in pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
TheNeuralBit commented on a change in pull request #13585:
URL: https://github.com/apache/beam/pull/13585#discussion_r546840398
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSchemaIOProvider.java
##########
@@ -109,12 +107,12 @@ public Schema schema() {
}
@Override
- public PTransform<PCollection<Row>, POutput> buildWriter() {
+ public PTransform<PCollection<Row>, WriteFilesResult<?>> buildWriter() {
Review comment:
Ack, thank you
----------------------------------------------------------------
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] [beam] iemejia commented on pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13585:
URL: https://github.com/apache/beam/pull/13585#issuecomment-749434090
Thanks Fokko I will follow track on the Iceberg side, do not hesitate to ping me if anything needed!
----------------------------------------------------------------
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] [beam] Fokko commented on pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #13585:
URL: https://github.com/apache/beam/pull/13585#issuecomment-749261080
Thanks for the pointers. Turns out that this wasn't needed. I'm able to wire everything together using the `FileIO` API. First attempt is in https://github.com/apache/iceberg/pull/1972
----------------------------------------------------------------
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] [beam] jkff commented on a change in pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
jkff commented on a change in pull request #13585:
URL: https://github.com/apache/beam/pull/13585#discussion_r546818013
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
##########
@@ -1741,9 +1740,8 @@ public void populateDisplayData(DisplayData.Builder builder) {
}
@Override
- public PDone expand(PCollection<T> input) {
- input.apply(inner);
- return PDone.in(input.getPipeline());
+ public WriteFilesResult<?> expand(PCollection<T> input) {
Review comment:
Agreed, this should use FileIO.write + AvroIO.sink - the current change is incompatible and will break anybody's transforms of the form:
```
PDone expand(...) {
...
return AvroIO.write()...; // If return type changes to WFR, this stops compiling
}
```
----------------------------------------------------------------
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] [beam] iemejia commented on pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13585:
URL: https://github.com/apache/beam/pull/13585#issuecomment-748983612
@tszerszen Spark Structured Streaming Validates Runner tests are failing so I think we can ignore those for the moment.
----------------------------------------------------------------
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] [beam] tszerszen commented on pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13585:
URL: https://github.com/apache/beam/pull/13585#issuecomment-748978807
Run Spark StructuredStreaming ValidatesRunner
----------------------------------------------------------------
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] [beam] iemejia commented on a change in pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
iemejia commented on a change in pull request #13585:
URL: https://github.com/apache/beam/pull/13585#discussion_r546732773
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java
##########
@@ -1741,9 +1740,8 @@ public void populateDisplayData(DisplayData.Builder builder) {
}
@Override
- public PDone expand(PCollection<T> input) {
- input.apply(inner);
- return PDone.in(input.getPipeline());
+ public WriteFilesResult<?> expand(PCollection<T> input) {
Review comment:
This solution looks ok, but I am wary of the consequences of changing the return type. It looks like we introduced TypedWrite on Beam for the achieve the same goal without breaking backwards compatibility.
The 'modern' preferred way to write files on Beam is via `FileIO.write()` which already returns `WriteFilesResult`.
Can you check if you we can achieve the intended results by relying on FileIO.write() + AvroIO.sink() ? or is there anything missing?
CC: @jkff in case you have some extra detail to add.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroSchemaIOProvider.java
##########
@@ -109,12 +107,12 @@ public Schema schema() {
}
@Override
- public PTransform<PCollection<Row>, POutput> buildWriter() {
+ public PTransform<PCollection<Row>, WriteFilesResult<?>> buildWriter() {
Review comment:
This looks good, here the change has less issues since this class is `@Internal`
CC @TheNeuralBit for awarenes.
##########
File path: sdks/java/core/src/main/java/org/apache/beam/sdk/io/WriteFiles.java
##########
@@ -90,7 +89,8 @@
/**
* A {@link PTransform} that writes to a {@link FileBasedSink}. A write begins with a sequential
* global initialization of a sink, followed by a parallel write, and ends with a sequential
- * finalization of the write. The output of a write is {@link PDone}.
+ * finalization of the write. The output of a write is {@link WriteFilesResult} with the files
Review comment:
:+1:
----------------------------------------------------------------
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] [beam] Fokko closed pull request #13585: [BEAM-10160] Change API to enable to append to Iceberg
Posted by GitBox <gi...@apache.org>.
Fokko closed pull request #13585:
URL: https://github.com/apache/beam/pull/13585
----------------------------------------------------------------
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