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/08/14 06:53:13 UTC
[GitHub] [iceberg] JingsongLi opened a new pull request #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
JingsongLi opened a new pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340
In #1293, we found lot of places use `switch case avro orc parquet` to create appender, these logicals can be extract to `GenericAppenderFactory`.
----------------------------------------------------------------
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 #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
Posted by GitBox <gi...@apache.org>.
rdblue commented on pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340#issuecomment-676826333
Thanks @JingsongLi, looks good.
----------------------------------------------------------------
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] openinx commented on a change in pull request #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340#discussion_r471894729
##########
File path: data/src/test/java/org/apache/iceberg/data/GenericAppenderHelper.java
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.data;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.List;
+import org.apache.iceberg.AppendFiles;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.DataFiles;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.StructLike;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.junit.Assert;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Helper for appending {@link DataFile} to a table or appending {@link Record}s to a table.
+ */
+public class GenericAppenderHelper {
+
+ private final Table table;
+ private final FileFormat fileFormat;
+ private final TemporaryFolder tmp;
+
+ public GenericAppenderHelper(Table table, FileFormat fileFormat, TemporaryFolder tmp) {
+ this.table = table;
+ this.fileFormat = fileFormat;
+ this.tmp = tmp;
+ }
+
+ public void appendToTable(DataFile... dataFiles) {
+ Preconditions.checkNotNull(table, "table not set");
+
+ AppendFiles append = table.newAppend();
+
+ for (DataFile dataFile : dataFiles) {
+ append = append.appendFile(dataFile);
+ }
+
+ append.commit();
+ }
+
+ public void appendToTable(List<Record> records) throws IOException {
+ appendToTable(null, records);
+ }
+
+ public void appendToTable(StructLike partition, List<Record> records) throws IOException {
+ appendToTable(writeFile(partition, records));
+ }
+
+ public DataFile writeFile(StructLike partition, List<Record> records) throws IOException {
+ Preconditions.checkNotNull(table, "table not set");
+ File file = tmp.newFile();
+ Assert.assertTrue(file.delete());
+ return appendToLocalFile(table, file, fileFormat, partition, records);
+ }
+
+ private static DataFile appendToLocalFile(
+ Table table, File file, FileFormat format, StructLike partition, List<Record> records)
+ throws IOException {
+ FileAppender<Record> appender = new GenericAppenderFactory(table.schema()).newAppender(
+ Files.localOutput(file), format);
+ try (FileAppender<Record> fileAppender = appender) {
+ fileAppender.addAll(records);
+ }
+
+ DataFiles.Builder builder = DataFiles.builder(table.spec())
+ .withRecordCount(records.size())
+ .withFileSizeInBytes(file.length())
+ .withPath(file.toURI().toString())
+ .withMetrics(appender.metrics())
+ .withFormat(format);
+ if (partition != null) {
Review comment:
We could call the `withPartition(partition)` directly without this nullable check, because `builder` will handle this inside 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.
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 merged pull request #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
Posted by GitBox <gi...@apache.org>.
rdblue merged pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340
----------------------------------------------------------------
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 #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
Posted by GitBox <gi...@apache.org>.
rdblue commented on a change in pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340#discussion_r473467777
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.data;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Factory to create a new {@link FileAppender} to write {@link Record}s.
+ */
+public class GenericAppenderFactory implements FileAppenderFactory<Record> {
+
+ private final Schema schema;
+ private final Map<String, String> config = Maps.newHashMap();
+
+ public GenericAppenderFactory(Schema schema) {
+ this.schema = schema;
+ }
+
+ public GenericAppenderFactory set(String property, String value) {
+ config.put(property, value);
+ return this;
+ }
+
+ public GenericAppenderFactory setAll(Map<String, String> properties) {
+ config.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public FileAppender<Record> newAppender(OutputFile outputFile, FileFormat fileFormat) {
+ try {
+ switch (fileFormat) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(DataWriter::create)
+ .named(fileFormat.name())
+ .setAll(config)
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(GenericParquetWriter::buildWriter)
+ .named(fileFormat.name())
+ .setAll(config)
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .setAll(config)
Review comment:
Yes, they 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] openinx commented on a change in pull request #1340: Introduce GenericAppenderFactory and GenericAppenderHelper
Posted by GitBox <gi...@apache.org>.
openinx commented on a change in pull request #1340:
URL: https://github.com/apache/iceberg/pull/1340#discussion_r471887232
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.data;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Factory to create a new {@link FileAppender} to write {@link Record}s.
+ */
+public class GenericAppenderFactory implements FileAppenderFactory<Record> {
+
+ private final Schema schema;
+ private final Map<String, String> config = Maps.newHashMap();
+
+ public GenericAppenderFactory(Schema schema) {
+ this.schema = schema;
+ }
+
+ public GenericAppenderFactory set(String property, String value) {
+ config.put(property, value);
+ return this;
+ }
+
+ public GenericAppenderFactory setAll(Map<String, String> properties) {
+ config.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public FileAppender<Record> newAppender(OutputFile outputFile, FileFormat fileFormat) {
+ try {
+ switch (fileFormat) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(DataWriter::create)
+ .named(fileFormat.name())
Review comment:
The `name` is usually a table name for the given schema. not `fileFormat`. we may need to remove this line, or set it with a reasonable table name.
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.data;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Factory to create a new {@link FileAppender} to write {@link Record}s.
+ */
+public class GenericAppenderFactory implements FileAppenderFactory<Record> {
+
+ private final Schema schema;
+ private final Map<String, String> config = Maps.newHashMap();
+
+ public GenericAppenderFactory(Schema schema) {
+ this.schema = schema;
+ }
+
+ public GenericAppenderFactory set(String property, String value) {
+ config.put(property, value);
+ return this;
+ }
+
+ public GenericAppenderFactory setAll(Map<String, String> properties) {
+ config.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public FileAppender<Record> newAppender(OutputFile outputFile, FileFormat fileFormat) {
+ try {
+ switch (fileFormat) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(DataWriter::create)
+ .named(fileFormat.name())
+ .setAll(config)
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(GenericParquetWriter::buildWriter)
+ .named(fileFormat.name())
Review comment:
ditto
##########
File path: data/src/main/java/org/apache/iceberg/data/GenericAppenderFactory.java
##########
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iceberg.data;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.data.avro.DataWriter;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * Factory to create a new {@link FileAppender} to write {@link Record}s.
+ */
+public class GenericAppenderFactory implements FileAppenderFactory<Record> {
+
+ private final Schema schema;
+ private final Map<String, String> config = Maps.newHashMap();
+
+ public GenericAppenderFactory(Schema schema) {
+ this.schema = schema;
+ }
+
+ public GenericAppenderFactory set(String property, String value) {
+ config.put(property, value);
+ return this;
+ }
+
+ public GenericAppenderFactory setAll(Map<String, String> properties) {
+ config.putAll(properties);
+ return this;
+ }
+
+ @Override
+ public FileAppender<Record> newAppender(OutputFile outputFile, FileFormat fileFormat) {
+ try {
+ switch (fileFormat) {
+ case AVRO:
+ return Avro.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(DataWriter::create)
+ .named(fileFormat.name())
+ .setAll(config)
+ .build();
+
+ case PARQUET:
+ return Parquet.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(GenericParquetWriter::buildWriter)
+ .named(fileFormat.name())
+ .setAll(config)
+ .build();
+
+ case ORC:
+ return ORC.write(outputFile)
+ .schema(schema)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .setAll(config)
Review comment:
To keep the consistent semantics, I suggest to set `overwrite=true`. Both flink and spark's FileAppenderFactory have enabled the overwrite.
----------------------------------------------------------------
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