You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by GitBox <gi...@apache.org> on 2020/12/21 23:55:41 UTC
[GitHub] [iceberg] Fokko opened a new pull request #1972: First version of the Iceberg sink for Apache Beam
Fokko opened a new pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972
The first attempt of adding a sink to write data to a DFS using Beam.
Fixes #693
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r735057380
##########
File path: beam/src/test/java/org/apache/iceberg/beam/FileWriterTest.java
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import java.util.Map;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
Review comment:
Not: Now that we have our own shaded guava, can we use that instead or do you need the shaded Beam guava for some reason?
Commenting as their has been some recent interest in this PR and I happen to know there are a group of people that would greatly appreciate an open source Apache Beam connector.
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-762481434
@Fokko, sorry for the delay getting back to this. I've been trying to get everything reviewed for the 0.11.0 release.
I'm excited to see Beam support working, but I don't think that this PR has enough guardrails for general use. Because users create their own Avro files, it would be easy to create files without field IDs and append them to a table. Doing that also requires an extra step to read the Avro file if I'm reading `FilenameToDataFile` correctly. We usually want engines to handle the details of writing data correctly so users don't need to understand them, but this PR delegates the problem to users.
I think we would need to provide more support so that users don't write bad data into their tables, which would also fix the performance problem introduced by `FilenameToDataFile`. There is already support in Iceberg for writing Avro generics to both Avro and Parquet formats. Could this add file writers using that to ensure that file stats and schema are correct? That would easily add Parquet support in the process.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550269126
##########
File path: site/docs/beam.md
##########
@@ -0,0 +1,39 @@
+<!--
+ - 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.
+ -->
+
+# Apache Beam
+
+!!! Warning
+ The Beam API is experimental and in early alpha stage
+
+## Write records
+
+When sinking Avro files to a distributed file system using `FileIO`, we can catch rely the filenames downstream. We append these filenames to an Iceberg table incrementally.
+
+```java
+final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+final FileIO.Write<Void, GenericRecord> avroFileIO = FileIO.<GenericRecord>write()
+ .via(AvroIO.sink(avroSchema))
+ .to("gs://.../../")
+ .withSuffix(".avro");
Review comment:
The problem with converting Avro to Iceberg is that there is no guarantee that the field IDs are assigned consistently (see [the Field IDs section for Avro](http://iceberg.apache.org/spec/#avro)). If the ID annotations aren't present, they are reassigned but that depends on the structure of the schema.
To safely write Avro for an Iceberg table, you should always convert from the table schema to an Avro schema so that the IDs are correct and match the table.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549472232
##########
File path: site/docs/beam.md
##########
@@ -0,0 +1,39 @@
+<!--
+ - 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.
+ -->
+
+# Apache Beam
+
+!!! Warning
+ The Beam API is experimental and in early alpha stage
+
+## Write records
+
+When sinking Avro files to a distributed file system using `FileIO`, we can catch rely the filenames downstream. We append these filenames to an Iceberg table incrementally.
+
+```java
+final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+final FileIO.Write<Void, GenericRecord> avroFileIO = FileIO.<GenericRecord>write()
+ .via(AvroIO.sink(avroSchema))
+ .to("gs://.../../")
+ .withSuffix(".avro");
Review comment:
This would work if the Avro schema passed to the sink were created from the Iceberg table schema because that sets the field IDs correctly. But I think that's not very obvious to users, so I think it would be better to eventually have an Iceberg sink that uses the Iceberg schema directly like Spark and Flink do.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-752742829
@rdblue That was my first thought as well. Unfortunally that's not the case:
```
MacBook-Pro-van-Fokko:dbt-data-ai-summit fokkodriesprong$ avro-tools tojson file:/tmp/fokko/1970-01-01T00-00-00.000+00:00-1970-01-01T00-01-00.000+00:00-00000-of-00001.avro
20/12/30 21:05:11 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Exception in thread "main" java.lang.IllegalArgumentException: java.net.URISyntaxException: Relative path in absolute URI: .1970-01-01T00-00-00.000+00:00-1970-01-01T00-01-00.000+00:00-00000-of-00001.avro.crc
at org.apache.hadoop.fs.Path.initialize(Path.java:263)
at org.apache.hadoop.fs.Path.<init>(Path.java:221)
at org.apache.hadoop.fs.Path.<init>(Path.java:129)
at org.apache.hadoop.fs.ChecksumFileSystem.getChecksumFile(ChecksumFileSystem.java:94)
at org.apache.hadoop.fs.ChecksumFileSystem$ChecksumFSInputChecker.<init>(ChecksumFileSystem.java:149)
at org.apache.hadoop.fs.ChecksumFileSystem.open(ChecksumFileSystem.java:347)
at org.apache.hadoop.fs.FileSystem.open(FileSystem.java:899)
at org.apache.avro.tool.Util.openFromFS(Util.java:88)
at org.apache.avro.tool.Util.fileOrStdin(Util.java:64)
at org.apache.avro.tool.DataFileReadTool.run(DataFileReadTool.java:87)
at org.apache.avro.tool.Main.run(Main.java:67)
at org.apache.avro.tool.Main.main(Main.java:56)
Caused by: java.net.URISyntaxException: Relative path in absolute URI: .1970-01-01T00-00-00.000+00:00-1970-01-01T00-01-00.000+00:00-00000-of-00001.avro.crc
at java.net.URI.checkPath(URI.java:1823)
at java.net.URI.<init>(URI.java:745)
at org.apache.hadoop.fs.Path.initialize(Path.java:260)
... 11 more
```
I've added a new Naming Policy to the PR that resolves the issue. Besides that, I've added an assert that catches the semicolon, to make sure that you don't write data, that you can't read using Spark/Hive/etc.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r547432651
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ public static void write(TableIdentifier table, Schema schema, String hiveMetastoreUrl, WriteFilesResult<Void> resultFiles) {
Review comment:
This doesn't conform to the normal style we use in the project, which is 2 spaces for an indent.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] iemejia edited a comment on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
iemejia edited a comment on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-768099869
@Fokko on the interest of making Beam more object store friendly don't hesitate to bring ideas/discussion into the Beam community. I think we need to improve in this area. Does this have an associated JIRA in the Iceberg side? I want to link it to [BEAM-10160](https://issues.apache.org/jira/browse/BEAM-10160)
Also do you plan to add the read part too?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r590909197
##########
File path: beam/src/main/java/org/apache/iceberg/beam/FileWriter.java
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.iceberg.beam;
+
+import java.io.IOException;
+import java.util.Locale;
+import java.util.Map;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionKey;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.LocationProvider;
+import org.apache.iceberg.io.OutputFileFactory;
+import org.apache.iceberg.io.PartitionedWriter;
+import org.apache.iceberg.io.TaskWriter;
+import org.apache.iceberg.io.UnpartitionedWriter;
+import org.apache.iceberg.relocated.com.google.common.annotations.VisibleForTesting;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FileWriter<T extends GenericRecord> extends DoFn<T, DataFile> {
+ private static final Logger LOG = LoggerFactory.getLogger(FileWriter.class);
+
+ private final PartitionSpec spec;
+ private final TableIdentifier tableIdentifier;
+ private final String hiveMetastoreUrl;
+
+ private Map<String, String> properties;
+ private Schema schema;
+ private LocationProvider locations;
+ private FileIO io;
+ private EncryptionManager encryptionManager;
+ private HiveCatalog catalog;
+ private FileFormat fileFormat;
+
+ private transient TaskWriter<T> writer;
+ private transient BoundedWindow lastSeenWindow;
+
+ public FileWriter(
+ TableIdentifier tableIdentifier,
+ Schema schema,
+ PartitionSpec spec,
+ String hiveMetastoreUrl,
+ Map<String, String> properties
+ ) {
+ this.tableIdentifier = tableIdentifier;
+ this.spec = spec;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ this.schema = schema;
+ this.properties = properties;
+ }
+
+ @StartBundle
+ public void startBundle(StartBundleContext sbc) {
+ start();
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, BoundedWindow window) {
+ appendRecord(
+ context.element(),
+ window,
+ (int) context.pane().getIndex(),
+ context.pane().getIndex()
+ );
+ }
+
+ @FinishBundle
+ public void finishBundle(FinishBundleContext fbc) {
+ DataFile[] files = finish();
+ for (DataFile file : files) {
+ fbc.output(file, Instant.now(), lastSeenWindow);
+ }
+ }
+
+ @VisibleForTesting
+ public void start() {
+ Configuration conf = new Configuration();
+ for (String key : this.properties.keySet()) {
+ conf.set(key, this.properties.get(key));
+ }
+ catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ conf
+ );
+ Table table = HiveCatalogHelper.loadOrCreateTable(
+ catalog,
+ tableIdentifier,
+ schema,
+ spec,
+ properties
+ );
+ this.schema = table.schema();
+ this.locations = table.locationProvider();
+ this.properties = table.properties();
+ this.io = table.io();
+ this.encryptionManager = table.encryption();
+ String formatString = table.properties().getOrDefault(
+ TableProperties.DEFAULT_FILE_FORMAT, TableProperties.DEFAULT_FILE_FORMAT_DEFAULT);
+ this.fileFormat = FileFormat.valueOf(formatString.toUpperCase(Locale.ENGLISH));
+ }
+
+ @VisibleForTesting
+ public void appendRecord(T element, BoundedWindow window, int partitionId, long taskId) {
+ if (writer == null) {
+ LOG.info("Setting up the writer");
+ // We would rather do this in the startBundle, but we don't know the pane
+
+ BeamAppenderFactory<T> appenderFactory = new BeamAppenderFactory<>(schema, properties, spec);
+ OutputFileFactory fileFactory = new OutputFileFactory(
+ spec, fileFormat, locations, io, encryptionManager, partitionId, taskId);
+
+ if (spec.isUnpartitioned()) {
+ writer = new UnpartitionedWriter<>(spec, fileFormat, appenderFactory, fileFactory, io, Long.MAX_VALUE);
+ } else {
+ writer = new PartitionedWriter<T>(
+ spec, fileFormat, appenderFactory, fileFactory, io, Long.MAX_VALUE) {
+ @Override
+ protected PartitionKey partition(T row) {
+ return new PartitionKey(spec, schema);
+ }
+ };
+ }
+ }
+ try {
+ lastSeenWindow = window;
+ writer.write(element);
Review comment:
What if we get records for multiple partitions? I think `PartitionedWriter` will fail with a runtime exception.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550270117
##########
File path: api/src/main/java/org/apache/iceberg/catalog/Namespace.java
##########
@@ -19,13 +19,14 @@
package org.apache.iceberg.catalog;
+import java.io.Serializable;
import java.util.Arrays;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
/**
* A namespace in a {@link Catalog}.
*/
-public class Namespace {
Review comment:
Sounds good. I have no problem making these serializable.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-794749577
@madhubabubits, let us know if you have any other comments.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] iemejia commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-768591095
Please go ahead with the Read @Fokko I was just curious. I do have just too much in my backlog too to even risk to add more stuff. Let's sync for the Read part because I have a [WIP branch for Delta](https://github.com/iemejia/beam/tree/deltaio) and I suppose both can end up being similar (ok saying this with my poor knowledge of both Delta and Iceberg).
We can also discuss once you have finished the write (and/or read) on how to integrate it with Beam's SQL, it is quite straight-forward.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi edited a comment on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
aokolnychyi edited a comment on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-794707731
Thanks for the great work, @Fokko!
We have a person who is also interested in a Beam sink for Iceberg. It would be great to collaborate. @RussellSpitzer and I can help with reviews. I feel like this is not far from being ready.
A couple of questions from me (I have just a basic understanding of Beam):
- Is it correct that this PR supports only fixed windows in Beam?
- I head it is pretty common to write to multiple tables or determine which table to write to dynamically in Beam. Is it worth supporting such cases in the Iceberg sink? Does not have to be done in the first version, apparently.
- Does Beam have any way to control the distribution and ordering of data that is passed for write? For example, we have added a way to ask Spark to distribute and order records according to the partition spec and sort order. A similar effort is in progress for Flink. The main reason is to reduce the number of produced files and order records accordingly to benefit from min/max skipping and better compression. Can people shuffle and sort data manually? Can we do that automatically?
The first two points are minor but the last one can be a big deal at scale. For example, if we don't distribute and order data on write, we will either produce a lot of small files or keep a lot of files open at the same time.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549472780
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
Review comment:
Users may choose to do this, but I would not recommend it as the default behavior. It is too easy to write bad data to a table and get the union of two schemas instead of having the write rejected. Since we generally try to make Iceberg work like a normal SQL table, I think that this should not be the default.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550269735
##########
File path: beam/src/test/java/org/apache/iceberg/beam/IcebergIOTest.java
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class IcebergIOTest {
+ private static final List<String> SENTENCES =
+ Arrays.asList(
+ "Beam window 1 1",
+ "Beam window 1 2",
+ "Beam window 1 3",
+ "Beam window 1 4",
+ "Beam window 2 1",
+ "Beam window 2 2");
+ private static final Instant START_TIME = new Instant(0);
+
+ private static final List<String> FIRST_WIN_WORDS = SENTENCES.subList(0, 4);
+ private static final List<String> SECOND_WIN_WORDS = SENTENCES.subList(4, 6);
+ private static final Duration WINDOW_DURATION = Duration.standardMinutes(1);
+
+ @Rule
+ public final transient TestPipeline pipeline = TestPipeline.create();
+ final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+
+ private static final PipelineOptions options = TestPipeline.testingPipelineOptions();
+
+ private static final String stringSchema = "{\n" +
Review comment:
Okay, so the initial conversion to Iceberg assigns some IDs, then you pass that Iceberg schema into the source. The IDs are reassigned to ensure consistency when a table is created, and you're probably converting that table schema back to Iceberg to configure the Avro writer?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550996360
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
+ }
+
+ // Append the new files
+ final AppendFiles app = table.newAppend();
+ // We need to get the statistics, not easy to get them through Beam
Review comment:
Avro is a first-class citizen in Beam, this is more or less the equivalent of the InternalRow of Spark. We could replace the AvroIO with an IcebergIO, but this feels like reinventing the wheel to me. Then we still need to current logic of doing the commits. So, let's keep this option open for now.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-753598331
@rdblue Happy new year! :) Looked into this, it was much more interesting (and time consuming :-) than I've anticipated. You were close. Allow me to summarize:
Also good to know, is that Beam does not output the crc files for the Avro data files, and therefore it silently ignores them when it comes to the names without the semicolons:
However, with the semicolons, it fails at line 145. So it is unable to open the file. It indeed takes the part before as the scheme:
At `Path(Path parent, String child)` in `Path.java` the parent is `/tmp/test_streaming_crc`, and the child `.output-1970-01-01T00:00:00.000Z-1970-01-01T00:01:00.000Z-00000-of-00001.avro.crc`. Which is being split by the Path constructor. It detects the first semicolon, and will proceed if there is no slash. So this will make `.output-1970-01-01T00` the schema, and `00:00.000Z-1970-01-01T00:01:00.000Z-00000-of-00001.avro.crc` the file:
However, Java doesn't not accept the semicolons. It boils down to this:
```java
import java.net.URI;
// This works:
new URI(null, null, normalizePath(null, ".00004-e82668ab-25ec-4190-864e-e4d140eb83b5.metadata.json.crc"), null, null);
// This breaks:
new URI(null, null, normalizePath(null, ".output-1970-01-01T00:00:00.000Z-1970-01-01T00:01:00.000Z-00000-of-00001.avro.crc"), null, null);
```
Throwing the Relative path in absolute URI error. This is still an open issue:
- https://issues.apache.org/jira/browse/HADOOP-3257
- https://issues.apache.org/jira/browse/HADOOP-7945
- https://issues.apache.org/jira/browse/HADOOP-12455
- https://issues.apache.org/jira/browse/HADOOP-14217
I'll add a note to the new naming policy, referring to this issue, and leave this as is for now.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550996360
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
+ }
+
+ // Append the new files
+ final AppendFiles app = table.newAppend();
+ // We need to get the statistics, not easy to get them through Beam
Review comment:
Avro is a first-class citizen in Beam, this is more or less the equivalent of the InternalRow of Spark. We could replace the AvroIO with an IcebergIO, but this feels like reinventing the wheel to me. I've noticed that there is quite some effort in the writers for Beam. For example, tuning the number and size of the output buffers kind of stuff. I would like to reuse this logic.
Apart from the writers themselves. We still need to current logic of doing the commits. So, I think we can keep this open for now.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-750490365
@rdblue @HeartSaVioR Thanks for the review. This is indeed still a draft. I'll add some more tests soon 👍
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] HeartSaVioR commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r547702165
##########
File path: hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
##########
@@ -58,9 +58,11 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class HiveCatalog extends BaseMetastoreCatalog implements Closeable, SupportsNamespaces {
+public class HiveCatalog extends BaseMetastoreCatalog implements AutoCloseable, SupportsNamespaces {
Review comment:
I'm not sure this diff "technically" change anything. Sounds unnecessary unless explained.
##########
File path: hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
##########
@@ -58,9 +58,11 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class HiveCatalog extends BaseMetastoreCatalog implements Closeable, SupportsNamespaces {
+public class HiveCatalog extends BaseMetastoreCatalog implements AutoCloseable, SupportsNamespaces {
Review comment:
I'm not sure this diff "technically" changes anything. Sounds unnecessary unless explained.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549472916
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
Review comment:
Schemas can be semantically compared using `table.schema().asStruct().equals(schema.asStruct())`.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-751397938
Able to read the results with Spark:
```
MacBook-Pro-van-Fokko:incubator-iceberg fokkodriesprong$ spark-sql --packages org.apache.iceberg:iceberg-spark3:0.10.0 --conf "spark.sql.catalog.hive_prod=org.apache.iceberg.spark.SparkCatalog" --conf "spark.sql.catalog.hive_prod.type=hive" --conf "spark.sql.catalog.hive_prod.uri=thrift://localhost:9083"
spark-sql> SELECT * FROM hive_prod.default.test;
20/12/26 22:07:46 INFO BaseMetastoreTableOperations: Refreshing table metadata from new version: file:/tmp/test/metadata/00004-ed964aeb-8924-492b-8225-eb2102c22f03.metadata.json
20/12/26 22:07:46 INFO BaseMetastoreCatalog: Table loaded by catalog: hive_prod.default.test
20/12/26 22:07:47 INFO MemoryStore: Block broadcast_0 stored as values in memory (estimated size 315.8 KiB, free 366.0 MiB)
20/12/26 22:07:48 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes in memory (estimated size 27.4 KiB, free 366.0 MiB)
20/12/26 22:07:48 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory on 192.168.1.113:56438 (size: 27.4 KiB, free: 366.3 MiB)
20/12/26 22:07:48 INFO SparkContext: Created broadcast 0 from broadcast at SparkScanBuilder.java:138
20/12/26 22:07:48 INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 40.0 B, free 366.0 MiB)
20/12/26 22:07:48 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes in memory (estimated size 116.0 B, free 366.0 MiB)
20/12/26 22:07:48 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory on 192.168.1.113:56438 (size: 116.0 B, free: 366.3 MiB)
20/12/26 22:07:48 INFO SparkContext: Created broadcast 1 from broadcast at SparkScanBuilder.java:139
20/12/26 22:07:48 INFO V2ScanRelationPushDown:
Pushing operators to hive_prod.default.test
Pushed Filters:
Post-Scan Filters:
Output: word#0
20/12/26 22:07:48 INFO TableScan: Scanning table hive_prod.default.test snapshot 4784653135298349855 created at 2020-12-26 22:07:38.530 with filter true
20/12/26 22:07:49 INFO CodeGenerator: Code generated in 226.493784 ms
20/12/26 22:07:49 INFO SparkContext: Starting job: main at NativeMethodAccessorImpl.java:0
20/12/26 22:07:49 INFO DAGScheduler: Got job 0 (main at NativeMethodAccessorImpl.java:0) with 1 output partitions
20/12/26 22:07:49 INFO DAGScheduler: Final stage: ResultStage 0 (main at NativeMethodAccessorImpl.java:0)
20/12/26 22:07:49 INFO DAGScheduler: Parents of final stage: List()
20/12/26 22:07:49 INFO DAGScheduler: Missing parents: List()
20/12/26 22:07:49 INFO DAGScheduler: Submitting ResultStage 0 (MapPartitionsRDD[3] at main at NativeMethodAccessorImpl.java:0), which has no missing parents
20/12/26 22:07:49 INFO MemoryStore: Block broadcast_2 stored as values in memory (estimated size 7.1 KiB, free 366.0 MiB)
20/12/26 22:07:49 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes in memory (estimated size 3.5 KiB, free 366.0 MiB)
20/12/26 22:07:49 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory on 192.168.1.113:56438 (size: 3.5 KiB, free: 366.3 MiB)
20/12/26 22:07:49 INFO SparkContext: Created broadcast 2 from broadcast at DAGScheduler.scala:1223
20/12/26 22:07:49 INFO DAGScheduler: Submitting 1 missing tasks from ResultStage 0 (MapPartitionsRDD[3] at main at NativeMethodAccessorImpl.java:0) (first 15 tasks are for partitions Vector(0))
20/12/26 22:07:49 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks
20/12/26 22:07:49 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, 192.168.1.113, executor driver, partition 0, PROCESS_LOCAL, 11598 bytes)
20/12/26 22:07:49 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
...
20/12/26 22:07:50 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 1580 bytes result sent to driver
20/12/26 22:07:50 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 1132 ms on 192.168.1.113 (executor driver) (1/1)
20/12/26 22:07:50 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
20/12/26 22:07:50 INFO DAGScheduler: ResultStage 0 (main at NativeMethodAccessorImpl.java:0) finished in 1,235 s
20/12/26 22:07:50 INFO DAGScheduler: Job 0 is finished. Cancelling potential speculative or zombie tasks for this job
20/12/26 22:07:50 INFO TaskSchedulerImpl: Killing all running tasks in stage 0: Stage finished
20/12/26 22:07:50 INFO DAGScheduler: Job 0 finished: main at NativeMethodAccessorImpl.java:0, took 1,274538 s
hands
yes
my
go
are
typing
words
Time taken: 4.501 seconds, Fetched 7 row(s)
```
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] yushuyao commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
yushuyao commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-950252168
@Fokko I see this has been open for a while. Wondering when do you plan to merge it?
--
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.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550268330
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
+ }
+
+ // Append the new files
+ final AppendFiles app = table.newAppend();
+ // We need to get the statistics, not easy to get them through Beam
Review comment:
Iceberg provides common write classes in the core module that we extend for Flink and Spark object models, so if you do go with Iceberg writers, it would probably not be too much code. For example, take a look at the [`SparkPartitionedWriter`](https://github.com/apache/iceberg/blob/master/spark/src/main/java/org/apache/iceberg/spark/source/SparkPartitionedWriter.java).
Using Iceberg writers would help with stats as well as writing delete files if you later want to support CDC or other row-level use cases.
But, one challenge is integrating those writers with an object model. Spark and Flink have row classes that are directly supported in the integration. It looks like you're using Avro generics for Avro, and using an object model specific to a data format is going to be a little difficult because Iceberg needs additional metadata, like field IDs. It sounds difficult to make sure that each format writer is producing the right information for Iceberg tables.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r547433464
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
Review comment:
Can you use the Iceberg version of Guava? Unless you're passing Guava objects to the Beam API, this should use Iceberg's bundled Guava version. (And hopefully there are no places where Beam requires relocated Guava classes!)
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r592027533
##########
File path: api/src/main/java/org/apache/iceberg/catalog/TableIdentifier.java
##########
@@ -28,7 +29,7 @@
/**
* Identifies a table in iceberg catalog.
*/
-public class TableIdentifier {
+public class TableIdentifier implements Serializable {
Review comment:
For some of these classes where they are implementing Serializable, should we consider adding a `serialVersionUID`?
I have many users (specifically for Flink) that encounter subtle issues when some Serializable class does not specify `serialVersionUID` and the compiler changes the assigned seriallVersionUID. Very often, these users need to allow some or all of their jobs existing state to be discarded when this happens.
I'm not sure if `TableIdentifier` gets placed into state anywhere ourselves, but for classes that are part of the public api is there an argument against including them?
Additionally, does anybody have any arguments for or against including serial version uid on the interfaces which are also extending serializable? For example, `DataFile` stands out to me in particular as things that could potentially cause issue. I could imagine that something along the lines of `Class<DataFile> clazz = DataFile.class` could potentially get serialized and cause issues when the interface changes.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r592018073
##########
File path: api/src/main/java/org/apache/iceberg/DataFile.java
##########
@@ -35,7 +36,7 @@
/**
* Interface for data files listed in a table manifest.
*/
-public interface DataFile extends ContentFile<DataFile> {
+public interface DataFile extends ContentFile<DataFile>, Serializable {
Review comment:
If we are implementing Serializable, should we add serial version uid?
In particular, I have users that run into issues with things that get stored in state having changed version uids when they're not explicitly pinned.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi edited a comment on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
aokolnychyi edited a comment on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-794749577
@madhubabubits, let us know if you have any other comments. I know you are giving this PR a try.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-752752371
Here's the relevant check in URI that is failing:
```java
if ((path != null)
&& ((path.length() > 0) && (path.charAt(0) != '/')))
throw new URISyntaxException(s,
"Relative path in absolute URI");
```
Looks like the location is parsed in `Path`:
```java
// parse uri scheme, if any
int colon = pathString.indexOf(':');
int slash = pathString.indexOf('/');
if ((colon != -1) &&
((slash == -1) || (colon < slash))) { // has a scheme
scheme = pathString.substring(0, colon);
start = colon+1;
}
```
String's `indexOf` returns on the first occurrence of `:`, so it looks like it must be that the `file:` is getting stripped out somehow. Can you debug and see what's actually passed?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-764889511
@rdblue I agree, first writing with Beam, and computing statistics is very wasteful. Also, Beam isn't native to object stores (some renaming going on), so I'd have to fix that in the Beam codebase as well.
Got the Iceberg writers working today. Took less code than I anticipated. It is still very rough, so please don't look at the code :) Will clean this up tomorrow and add tests! Thanks!
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-764889511
@rdblue I agree, first writing with Beam, and computing statistics is very wasteful. Also, Beam isn't native to object stores (some renaming going on), so I'd have to fix that in the Beam codebase as well.
Got the Iceberg writers working today. Took less code than I anticipated. It is still very rough, so please don't look at the code :) Will clean this up tomorrow and add tests! Thanks!
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r550269735
##########
File path: beam/src/test/java/org/apache/iceberg/beam/IcebergIOTest.java
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class IcebergIOTest {
+ private static final List<String> SENTENCES =
+ Arrays.asList(
+ "Beam window 1 1",
+ "Beam window 1 2",
+ "Beam window 1 3",
+ "Beam window 1 4",
+ "Beam window 2 1",
+ "Beam window 2 2");
+ private static final Instant START_TIME = new Instant(0);
+
+ private static final List<String> FIRST_WIN_WORDS = SENTENCES.subList(0, 4);
+ private static final List<String> SECOND_WIN_WORDS = SENTENCES.subList(4, 6);
+ private static final Duration WINDOW_DURATION = Duration.standardMinutes(1);
+
+ @Rule
+ public final transient TestPipeline pipeline = TestPipeline.create();
+ final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+
+ private static final PipelineOptions options = TestPipeline.testingPipelineOptions();
+
+ private static final String stringSchema = "{\n" +
Review comment:
Okay, so the initial conversion to Iceberg assigns some IDs, then you pass that Iceberg schema into the source. The IDs are reassigned to ensure consistency when a table is created, and you're probably converting that table schema back to Iceberg to configure the Avro writer? Or maybe you're just getting lucky and the IDs are assigned the same way in the tests.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r548268733
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,102 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableList;
Review comment:
Auch, great catch!
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r592018073
##########
File path: api/src/main/java/org/apache/iceberg/DataFile.java
##########
@@ -35,7 +36,7 @@
/**
* Interface for data files listed in a table manifest.
*/
-public interface DataFile extends ContentFile<DataFile> {
+public interface DataFile extends ContentFile<DataFile>, Serializable {
Review comment:
If we are implementing Serializable, should we add serial version uid?
In particular, I have Flink users that run into issues with things that get stored in state having changed version uids when they're not explicitly pinned.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-762439441
How do we feel about this PR? Only Avro write support enough to get this in, or should we also include reading Iceberg? That would take some more time.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-768515075
@iemejia I would like to sort this out, before opening a new discussion :) But I think it is an import thing to address, as it slows down the throughput of the pipeline, and increases the costs (because of the increased number of operations). Iceberg doesn't rely on renames, as it is build with object stores in mind.
I'm happy to add the read part later on, but if you feel like it, feel free to pick it up. I got quite some backlog.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] iemejia commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-768099869
@Fokko on the interest of making Beam more object store friendly don't hesitate to bring ideas/discussion into the Beam community. I think we need to improve in this area. Does this have an associated JIRA in the Iceberg side? I want to link it to [BEAM-10160](https://issues.apache.org/jira/browse/BEAM-10160)
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-751849766
Looks like a good start, @Fokko! I'd definitely recommend relying a bit more on Iceberg helpers to do things like write data files, but this does demonstrate the basic elements of writing to an Iceberg table.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r547432390
##########
File path: hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
##########
@@ -58,9 +58,11 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class HiveCatalog extends BaseMetastoreCatalog implements Closeable, SupportsNamespaces {
+public class HiveCatalog extends BaseMetastoreCatalog implements AutoCloseable, SupportsNamespaces {
Review comment:
It doesn't look like these changes are needed. Could we move the changes in this file to a separate PR?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r548268009
##########
File path: hive-metastore/src/main/java/org/apache/iceberg/hive/HiveCatalog.java
##########
@@ -58,9 +58,11 @@
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class HiveCatalog extends BaseMetastoreCatalog implements Closeable, SupportsNamespaces {
+public class HiveCatalog extends BaseMetastoreCatalog implements AutoCloseable, SupportsNamespaces {
Review comment:
I was under the impression that we couldn't use `with { } `. But this seems incorrect. I'll revert this change. Thanks for pointing it out, so we can keep the history clean 👍
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] madhubabubits commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
madhubabubits commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r591053345
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,103 @@
+/*
+ * 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.iceberg.beam;
+
+
+import java.util.Map;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.coders.SerializableCoder;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Maps;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.catalog.TableIdentifier;
+
+public class IcebergIO {
+ private IcebergIO() {
+ }
+
+ public static final class Builder {
+ private TableIdentifier tableIdentifier;
+ private Schema schema;
+ private String hiveMetaStoreUrl;
+
+ private final Map<String, String> properties = Maps.newHashMap();
+
+ public Builder withTableIdentifier(TableIdentifier newTableIdentifier) {
+ this.tableIdentifier = newTableIdentifier;
+ return this;
+ }
+
+ public Builder withSchema(Schema newSchema) {
+ this.schema = newSchema;
+ return this;
+ }
+
+ public Builder withHiveMetastoreUrl(String newHiveMetaStoreUrl) {
+ assert newHiveMetaStoreUrl.startsWith("thrift://");
+ this.hiveMetaStoreUrl = newHiveMetaStoreUrl;
+ return this;
+ }
+
+ public Builder conf(String key, String value) {
+ this.properties.put(key, value);
+ return this;
+ }
+
+ public <T extends GenericRecord> PCollection<Snapshot> build(PCollection<T> avroRecords) {
+ return IcebergIO.write(
+ avroRecords,
+ this.tableIdentifier,
+ this.schema,
+ this.hiveMetaStoreUrl,
+ this.properties
+ );
+ }
+ }
+
+ private static <T extends GenericRecord> PCollection<Snapshot> write(
+ PCollection<T> avroRecords,
+ TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ Map<String, String> properties
+ ) {
+ // We take the filenames that are emitted by the FileIO
+ final PCollection<DataFile> dataFiles = avroRecords
+ .apply(
+ "Write DataFiles",
+ ParDo.of(new FileWriter<>(table, schema, PartitionSpec.unpartitioned(), hiveMetastoreUrl, properties))
+ )
+ .setCoder(SerializableCoder.of(DataFile.class));
+
+ // We use a combiner, to combine all the files to a single commit in
+ // the Iceberg log
+ final IcebergDataFileCommitter combiner = new IcebergDataFileCommitter(table, hiveMetastoreUrl, properties);
+ final Combine.Globally<DataFile, Snapshot> combined = Combine.globally(combiner).withoutDefaults();
Review comment:
@Fokko - firstly, great work with the implementation.
Do you see any use for returning the snapshot here ? Instead, can we implement something on the lines of FileIO sink ? that way user will have full control on windowing, dynamic tablename ( based on the avro schema of each GenericRecord in PCollection<T extends GenericRecord> ) and support both bounded & unbounded collection.
Also, do you see value in providing windowing for both creating files & windowing for committing ( like create files based on 5min window, but final commit is based on 1hr window ). This might be a use-case for unbounded collection
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549489600
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
Review comment:
Good point, and fully agree. I'll remove this one for now, this should be handled outside of the sink.
##########
File path: site/docs/beam.md
##########
@@ -0,0 +1,39 @@
+<!--
+ - 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.
+ -->
+
+# Apache Beam
+
+!!! Warning
+ The Beam API is experimental and in early alpha stage
+
+## Write records
+
+When sinking Avro files to a distributed file system using `FileIO`, we can catch rely the filenames downstream. We append these filenames to an Iceberg table incrementally.
+
+```java
+final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+final FileIO.Write<Void, GenericRecord> avroFileIO = FileIO.<GenericRecord>write()
+ .via(AvroIO.sink(avroSchema))
+ .to("gs://.../../")
+ .withSuffix(".avro");
Review comment:
What is the implication of doing this the other way around? In the current setup we have the Avro schema, and distill the Iceberg schema from it.
##########
File path: beam/src/test/java/org/apache/iceberg/beam/IcebergIOTest.java
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class IcebergIOTest {
+ private static final List<String> SENTENCES =
+ Arrays.asList(
+ "Beam window 1 1",
+ "Beam window 1 2",
+ "Beam window 1 3",
+ "Beam window 1 4",
+ "Beam window 2 1",
+ "Beam window 2 2");
+ private static final Instant START_TIME = new Instant(0);
+
+ private static final List<String> FIRST_WIN_WORDS = SENTENCES.subList(0, 4);
+ private static final List<String> SECOND_WIN_WORDS = SENTENCES.subList(4, 6);
+ private static final Duration WINDOW_DURATION = Duration.standardMinutes(1);
+
+ @Rule
+ public final transient TestPipeline pipeline = TestPipeline.create();
+ final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+
+ private static final PipelineOptions options = TestPipeline.testingPipelineOptions();
+
+ private static final String stringSchema = "{\n" +
Review comment:
Yes, I've started by implementing Avro files, because this is our use case. But with the Beam API, Parquet and ORC are also supported.
For the actual interface, we only use the Iceberg schema. The `IcebergIO` only accepts `org.apache.iceberg.Schema`. For the test we convert the Avro schema to iceberg using `AvroSchemaUtil.toIceberg(avroSchema);`.
translates into:
I need to clean the test up a bit.
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
+ }
+
+ // Append the new files
+ final AppendFiles app = table.newAppend();
+ // We need to get the statistics, not easy to get them through Beam
Review comment:
Thanks for pointing out. I'll track PR https://github.com/apache/iceberg/pull/1963/files. I think we have to make a decision here. I do like the fact that we fully re-use the writers from Beam, so we don't have to maintain that part.
##########
File path: api/src/main/java/org/apache/iceberg/catalog/Namespace.java
##########
@@ -19,13 +19,14 @@
package org.apache.iceberg.catalog;
+import java.io.Serializable;
import java.util.Arrays;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
/**
* A namespace in a {@link Catalog}.
*/
-public class Namespace {
Review comment:
Unfortunately, yes. With Beam, there is no Master/Worker as with Spark or Task/Job manager as with Flink. Everything happens on a worker, requiring everything to be serialized. For example, the interaction with the Hive metastore is also done by the worker. This requires everything to be serialized.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-752699397
For the file issue, what is the full location that you're passing into Iceberg? Are you sure it isn't missing `file:/` to signal that it is the local FS? I think maybe that's the problem.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-749704363
@Fokko, thanks for working on this! This looks like it may not be complete. Do you think it should be a draft?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549473210
##########
File path: api/src/main/java/org/apache/iceberg/catalog/Namespace.java
##########
@@ -19,13 +19,14 @@
package org.apache.iceberg.catalog;
+import java.io.Serializable;
import java.util.Arrays;
import org.apache.iceberg.relocated.com.google.common.base.Joiner;
/**
* A namespace in a {@link Catalog}.
*/
-public class Namespace {
Review comment:
Are these changes 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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] Fokko commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
Fokko commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-760930790
Works like a charm:
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] aokolnychyi commented on pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
aokolnychyi commented on pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#issuecomment-794707731
Thanks for the great work, @Fokko!
We have a person who is also interested in a Beam sink for Iceberg. It would be great to collaborate. @RussellSpitzer and I can help with reviews. I feel like this is not far from being ready.
A couple of questions from me (I have just a basic understanding of Beam):
- Is it correct that this PR supports only fixed windows in Beam?
- I head it is pretty common to write to multiple tables or determine which table to write to dynamically in Beam. Is it worth supporting such cases in the Iceberg sink? Does not have to be done in the first version, apparently.
- Does Beam have any way to control the distribution and ordering of data that is passed for write? For example, we have added a way to ask Spark to distribute and order records according to the partition spec and sort order. A similar effort is in progress for Flink. The main reason is to reduce the number of produced files and order records accordingly to benefit from min/max skipping and better compression. Can people shuffle and sort data manually? Can we do that automatically?
The first two points are minor but the last one can be a big deal at scale. For example, if we don't distribute and order data on write, we either need to produce a lot of small files or keep a lot of files open.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r592018073
##########
File path: api/src/main/java/org/apache/iceberg/DataFile.java
##########
@@ -35,7 +36,7 @@
/**
* Interface for data files listed in a table manifest.
*/
-public interface DataFile extends ContentFile<DataFile> {
+public interface DataFile extends ContentFile<DataFile>, Serializable {
Review comment:
If we are implementing Serializable, should we add serial version uid?
In particular, I have Flink users that run into issues with things that get stored in state having changed version uids when they're not explicitly pinned.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] kbendick commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
kbendick commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r592031133
##########
File path: core/src/main/java/org/apache/iceberg/io/OutputFileFactory.java
##########
@@ -26,6 +26,7 @@
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.encryption.EncryptedOutputFile;
import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
Review comment:
Nit: This seems to be unused.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549471754
##########
File path: beam/src/test/java/org/apache/iceberg/beam/IcebergIOTest.java
##########
@@ -0,0 +1,166 @@
+/*
+ * Licensed 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.iceberg.beam;
+
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.AvroCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.AvroIO;
+import org.apache.beam.sdk.io.FileIO;
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.testing.TestStream;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.TimestampedValue;
+import org.apache.iceberg.avro.AvroSchemaUtil;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.joda.time.Duration;
+import org.joda.time.Instant;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class IcebergIOTest {
+ private static final List<String> SENTENCES =
+ Arrays.asList(
+ "Beam window 1 1",
+ "Beam window 1 2",
+ "Beam window 1 3",
+ "Beam window 1 4",
+ "Beam window 2 1",
+ "Beam window 2 2");
+ private static final Instant START_TIME = new Instant(0);
+
+ private static final List<String> FIRST_WIN_WORDS = SENTENCES.subList(0, 4);
+ private static final List<String> SECOND_WIN_WORDS = SENTENCES.subList(4, 6);
+ private static final Duration WINDOW_DURATION = Duration.standardMinutes(1);
+
+ @Rule
+ public final transient TestPipeline pipeline = TestPipeline.create();
+ final String hiveMetastoreUrl = "thrift://localhost:9083/default";
+
+ private static final PipelineOptions options = TestPipeline.testingPipelineOptions();
+
+ private static final String stringSchema = "{\n" +
Review comment:
I'm not sure why this would result in readable data files. Iceberg requires either a schema in the Avro file with field IDs or a name mapping in the Iceberg table to get the field IDs from Avro field names. How did you read the table produced by this test?
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] rdblue commented on a change in pull request #1972: First version of the Iceberg sink for Apache Beam
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1972:
URL: https://github.com/apache/iceberg/pull/1972#discussion_r549473125
##########
File path: beam/src/main/java/org/apache/iceberg/beam/IcebergIO.java
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.iceberg.beam;
+
+import org.apache.beam.sdk.io.WriteFilesResult;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.values.KV;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Snapshot;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NoSuchTableException;
+import org.apache.iceberg.hive.HiveCatalog;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+
+import java.util.LinkedList;
+import java.util.List;
+
+
+public class IcebergIO {
+
+ private IcebergIO() {
+ }
+
+ static void write(TableIdentifier table,
+ Schema schema,
+ String hiveMetastoreUrl,
+ WriteFilesResult<Void> resultFiles) {
+ resultFiles.getPerDestinationOutputFilenames().apply(
+ Combine.globally(new FileCombiner(table, schema, hiveMetastoreUrl)).withoutDefaults()
+ );
+ }
+
+ private static class FileCombiner extends Combine.CombineFn<KV<Void, String>, List<String>, Snapshot> {
+ private final TableIdentifier tableIdentifier;
+ private final Schema schema;
+ private final String hiveMetastoreUrl;
+
+ FileCombiner(TableIdentifier table, Schema schema, String hiveMetastoreUrl) {
+ this.tableIdentifier = table;
+ this.schema = schema;
+ this.hiveMetastoreUrl = hiveMetastoreUrl;
+ }
+
+ @Override
+ public List<String> createAccumulator() {
+ return new LinkedList<>();
+ }
+
+ @Override
+ public List<String> addInput(List<String> mutableAccumulator, KV<Void, String> input) {
+ mutableAccumulator.add(input.getValue());
+ return mutableAccumulator;
+ }
+
+ @Override
+ public List<String> mergeAccumulators(Iterable<List<String>> accumulators) {
+ return ImmutableList.copyOf(Iterables.concat(accumulators));
+ }
+
+ @Override
+ public Snapshot extractOutput(List<String> accumulator) {
+ try (HiveCatalog catalog = new HiveCatalog(
+ HiveCatalog.DEFAULT_NAME,
+ this.hiveMetastoreUrl,
+ 1,
+ new Configuration()
+ )) {
+ Table table;
+ try {
+ table = catalog.loadTable(this.tableIdentifier);
+ } catch (NoSuchTableException e) {
+ // If it doesn't exist, we just create the table
+ table = catalog.createTable(this.tableIdentifier, schema);
+ }
+
+ // In case the schema has changed
+ if (table.schema() != this.schema) {
+ table.updateSchema().unionByNameWith(this.schema).commit();
+ }
+
+ // Append the new files
+ final AppendFiles app = table.newAppend();
+ // We need to get the statistics, not easy to get them through Beam
Review comment:
There is a PR to aggregate the stats while writing an Avro file. You would get them for free using an Iceberg-based file writer instead of the generic Avro one. You'd also get correct record counts and file sizes.
----------------------------------------------------------------
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
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org