You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@gobblin.apache.org by GitBox <gi...@apache.org> on 2020/12/23 14:56:58 UTC
[GitHub] [incubator-gobblin] hanghangliu opened a new pull request #3184: Gobblin 1320 add iceberg writer
hanghangliu opened a new pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184
Dear Gobblin maintainers,
This pr is a still ongoing development. Any comment and advise is welcomed!
### JIRA
- [ ] My PR addresses the following [Gobblin JIRA](https://issues.apache.org/jira/browse/GOBBLIN/) issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-1320
### Description
- [ ] Here are some details about my PR, including screenshots (if applicable):
This PR is trying to add an Iceberg module, that enable Gobblin to write as Iceberg table format. It wraps Iceberg task writer into FsDataWriter. It accept Avro, ORC and Parquet format. This PR only addresses the writer part of the Iceberg module, and the source part will be in another PR.
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] autumnust commented on a change in pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
autumnust commented on a change in pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#discussion_r561455188
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergUtil.java
##########
@@ -0,0 +1,47 @@
+package org.apache.gobblin.writer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import java.util.Map;
+
+
+@Slf4j
+public class IcebergUtil {
+
+ public static Table createTable(String path, Map<String, String> properties, boolean partitioned, Schema schema) {
+ PartitionSpec spec;
+ if (partitioned) {
+ spec = PartitionSpec.builderFor(schema).identity("data").build();
Review comment:
Shall we get more specific identity value here?
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergTaskWriterFactory.java
##########
@@ -0,0 +1,64 @@
+package org.apache.gobblin.writer;
+
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.io.*;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import java.util.Map;
+
+/**
+ * Factory to create {@link TaskWriter} that is.
+ * responsible for writing generic data
+ * into Iceberg table.
+ *
+ */
+public class IcebergTaskWriterFactory {
+
+ private final Schema schema;
+ private final PartitionSpec partitionSpec;
+ private final LocationProvider locations;
+ private final FileIO io;
+ private final EncryptionManager encryptionManager;
+ private final long targetFileSizeBytes;
+ private final FileFormat format;
+ private final IcebergFileAppenderFactory appenderFactory;
+
+ private transient OutputFileFactory outputFileFactory;
+
+ public IcebergTaskWriterFactory(Schema schema,
+ PartitionSpec partitionSpec,
+ LocationProvider locations,
+ FileIO io,
+ EncryptionManager encryptionManager,
+ long targetFileSizeBytes,
+ FileFormat format,
+ Map<String, String> tableProperties) {
+ this.schema = schema;
+ this.partitionSpec = partitionSpec;
+ this.locations = locations;
+ this.io = io;
+ this.encryptionManager = encryptionManager;
+ this.targetFileSizeBytes = targetFileSizeBytes;
+ this.format = format;
+ this.appenderFactory = new IcebergFileAppenderFactory(schema);
+ }
+
+ public void initialize(int taskId, int attemptId) {
+ this.outputFileFactory = new OutputFileFactory(partitionSpec, format, locations, io, encryptionManager, taskId, attemptId);
+ }
+
+ public TaskWriter create() {
+ Preconditions.checkNotNull(outputFileFactory,
+ "The outputFileFactory shouldn't be null if we have invoked the initialize().");
+
+ if (partitionSpec.fields().isEmpty()) {
Review comment:
So why we need this if-branch ?
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergFileAppenderFactory.java
##########
@@ -0,0 +1,84 @@
+package org.apache.gobblin.writer;
+
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+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.io.TaskWriter;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+
+
+/**
+ * Factory to create a new {@link FileAppender} that
+ * write generic data. The created {@link FileAppender}
+ * will be wrapped in {@link TaskWriter}
+ */
+public class IcebergFileAppenderFactory implements FileAppenderFactory {
Review comment:
shall we rename it to `GobblinFileAppender` if following those implementation in Iceberg codebase?
##########
File path: gobblin-modules/gobblin-iceberg/src/test/java/org/apache/gobblin/writer/IcebergTestUtil.java
##########
@@ -0,0 +1,130 @@
+package org.apache.gobblin.writer;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.*;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopInputFile;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.testng.Assert;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.iceberg.hadoop.HadoopOutputFile.fromPath;
+
+public class IcebergTestUtil {
+
+ public static final Schema SCHEMA = new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get())
+ );
+
+ public static final Record RECORD = GenericRecord.create(SCHEMA);
+
+ public static Table createTable(String path, Map<String, String> properties, boolean partitioned) {
+ PartitionSpec spec;
+ if (partitioned) {
+ spec = PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ } else {
+ spec = PartitionSpec.unpartitioned();
+ }
+ return new HadoopTables().create(SCHEMA, spec, properties, path);
+ }
+
+ public static Record createRecord(Integer id, String data) {
+ Record record = RECORD.copy();
+ record.setField("id", id);
+ record.setField("data", data);
+ return record;
+ }
+
+ public static void assertTableRecords(Table table, List<Record> expected) throws IOException {
+ table.refresh();
+ try (CloseableIterable<Record> iterable = IcebergGenerics.read(table).build()) {
+ Assert.assertEquals(
+ Sets.newHashSet(expected), Sets.newHashSet(iterable));
+ }
+ }
+
+ public static void assertTableRecords(String tablePath, List<Record> expected) throws IOException {
+ Preconditions.checkArgument(expected != null, "expected records shouldn't be null");
+ assertTableRecords(new HadoopTables().load(tablePath), expected);
+ }
+
+ public static org.apache.avro.Schema generateAvroSchema() {
+ org.apache.avro.Schema avroSchema = (new org.apache.avro.Schema.Parser()).parse("{\n" +
+ " \"namespace\": \"com.linkedin.orc\",\n" +
+ " \"type\": \"record\",\n" +
+ " \"name\": \"IcebergTest\",\n" +
+ " \"fields\": [\n" +
+ " {\n" +
+ " \"name\": \"id\",\n" +
+ " \"type\": \"int\"\n" +
+ " },\n" +
+ " {\n" +
+ " \"name\": \"data\",\n" +
+ " \"type\": \"string\"\n" +
+ " }\n" +
+ " ]\n" +
+ "}\n");
+ return avroSchema;
+ }
+
+ public static List<org.apache.avro.generic.GenericRecord> genericAvroRecords() throws IOException {
+ List<org.apache.avro.generic.GenericRecord> list = new ArrayList<>();
+ org.apache.avro.Schema avroSchema = generateAvroSchema();
+
+ GenericRecordBuilder builder_0 = new GenericRecordBuilder(avroSchema);
+ builder_0.set("id", 1);
+ builder_0.set("data", "alice");
+ list.add(builder_0.build());
+
+ GenericRecordBuilder builder_1 = new GenericRecordBuilder(avroSchema);
+ builder_1.set("id", 2);
+ builder_1.set("data", "bob");
+ list.add(builder_1.build());
+ return list;
+ }
+
+
+
+ public static List<org.apache.avro.generic.GenericRecord> deserializeAvroRecords(org.apache.avro.Schema schema, String schemaPath)
Review comment:
Where is this method being used? In fact I thought we will need to directly deal with avro file for deserialization ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] autumnust commented on pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
autumnust commented on pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#issuecomment-764178784
Also, there's a travis failure
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] codecov-io commented on pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
codecov-io commented on pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#issuecomment-750346721
# [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=h1) Report
> Merging [#3184](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=desc) (579aadc) into [master](https://codecov.io/gh/apache/incubator-gobblin/commit/fb5e40fcffa978fdde7e22efb851f73ddd7a3587?el=desc) (fb5e40f) will **decrease** coverage by `36.83%`.
> The diff coverage is `0.00%`.
[](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #3184 +/- ##
============================================
- Coverage 45.94% 9.10% -36.84%
+ Complexity 9602 1726 -7876
============================================
Files 1997 2019 +22
Lines 76097 77180 +1083
Branches 8469 8563 +94
============================================
- Hits 34960 7029 -27931
- Misses 37873 69469 +31596
+ Partials 3264 682 -2582
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...pache.gobblin/writer/IcebergDataWriterBuilder.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ0RhdGFXcml0ZXJCdWlsZGVyLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...che.gobblin/writer/IcebergFileAppenderFactory.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ0ZpbGVBcHBlbmRlckZhY3RvcnkuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...pache.gobblin/writer/IcebergTaskWriterFactory.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1Rhc2tXcml0ZXJGYWN0b3J5LmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...in/java/org.apache.gobblin/writer/IcebergUtil.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1V0aWwuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../java/org.apache.gobblin/writer/IcebergWriter.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1dyaXRlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...c/main/java/org/apache/gobblin/util/FileUtils.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi11dGlsaXR5L3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3V0aWwvRmlsZVV0aWxzLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-4.00%)` | |
| [...n/java/org/apache/gobblin/fork/CopyableSchema.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2ZvcmsvQ29weWFibGVTY2hlbWEuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-2.00%)` | |
| [...java/org/apache/gobblin/stream/ControlMessage.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vc3RyZWFtL0NvbnRyb2xNZXNzYWdlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| [...va/org/apache/gobblin/dataset/DatasetResolver.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1hcGkvc3JjL21haW4vamF2YS9vcmcvYXBhY2hlL2dvYmJsaW4vZGF0YXNldC9EYXRhc2V0UmVzb2x2ZXIuamF2YQ==) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-3.00%)` | |
| [...va/org/apache/gobblin/converter/EmptyIterable.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1jb3JlL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NvbnZlcnRlci9FbXB0eUl0ZXJhYmxlLmphdmE=) | `0.00% <0.00%> (-100.00%)` | `0.00% <0.00%> (-1.00%)` | |
| ... and [1083 more](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=footer). Last update [fb5e40f...579aadc](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-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
[GitHub] [incubator-gobblin] codecov-io edited a comment on pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
codecov-io edited a comment on pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#issuecomment-750346721
# [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=h1) Report
> Merging [#3184](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=desc) (579aadc) into [master](https://codecov.io/gh/apache/incubator-gobblin/commit/fb5e40fcffa978fdde7e22efb851f73ddd7a3587?el=desc) (fb5e40f) will **increase** coverage by `0.24%`.
> The diff coverage is `29.62%`.
[](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## master #3184 +/- ##
============================================
+ Coverage 45.94% 46.18% +0.24%
- Complexity 9602 9762 +160
============================================
Files 1997 2019 +22
Lines 76097 77180 +1083
Branches 8469 8563 +94
============================================
+ Hits 34960 35648 +688
- Misses 37873 38226 +353
- Partials 3264 3306 +42
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=tree) | Coverage Δ | Complexity Δ | |
|---|---|---|---|
| [...pache.gobblin/writer/IcebergDataWriterBuilder.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ0RhdGFXcml0ZXJCdWlsZGVyLmphdmE=) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...in/java/org.apache.gobblin/writer/IcebergUtil.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1V0aWwuamF2YQ==) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [.../java/org.apache.gobblin/writer/IcebergWriter.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1dyaXRlci5qYXZh) | `0.00% <0.00%> (ø)` | `0.00 <0.00> (?)` | |
| [...che.gobblin/writer/IcebergFileAppenderFactory.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ0ZpbGVBcHBlbmRlckZhY3RvcnkuamF2YQ==) | `32.25% <32.25%> (ø)` | `2.00 <2.00> (?)` | |
| [...pache.gobblin/writer/IcebergTaskWriterFactory.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1tb2R1bGVzL2dvYmJsaW4taWNlYmVyZy9zcmMvbWFpbi9qYXZhL29yZy5hcGFjaGUuZ29iYmxpbi93cml0ZXIvSWNlYmVyZ1Rhc2tXcml0ZXJGYWN0b3J5LmphdmE=) | `87.50% <87.50%> (ø)` | `3.00 <3.00> (?)` | |
| [...apache/gobblin/runtime/api/JobCatalogListener.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL0pvYkNhdGFsb2dMaXN0ZW5lci5qYXZh) | `76.92% <0.00%> (-23.08%)` | `0.00% <0.00%> (ø%)` | |
| [...n/runtime/job\_catalog/JobCatalogListenersList.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvam9iX2NhdGFsb2cvSm9iQ2F0YWxvZ0xpc3RlbmVyc0xpc3QuamF2YQ==) | `63.63% <0.00%> (-10.05%)` | `10.00% <0.00%> (ø%)` | |
| [...pache/gobblin/cluster/JobConfigurationManager.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvSm9iQ29uZmlndXJhdGlvbk1hbmFnZXIuamF2YQ==) | `81.39% <0.00%> (-6.11%)` | `10.00% <0.00%> (ø%)` | |
| [.../apache/gobblin/runtime/api/MutableJobCatalog.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1ydW50aW1lL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL3J1bnRpbWUvYXBpL011dGFibGVKb2JDYXRhbG9nLmphdmE=) | `81.25% <0.00%> (-5.42%)` | `0.00% <0.00%> (ø%)` | |
| [...ache/gobblin/cluster/GobblinHelixJobScheduler.java](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree#diff-Z29iYmxpbi1jbHVzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9nb2JibGluL2NsdXN0ZXIvR29iYmxpbkhlbGl4Sm9iU2NoZWR1bGVyLmphdmE=) | `34.48% <0.00%> (-4.74%)` | `6.00% <0.00%> (ø%)` | |
| ... and [54 more](https://codecov.io/gh/apache/incubator-gobblin/pull/3184/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=footer). Last update [fb5e40f...579aadc](https://codecov.io/gh/apache/incubator-gobblin/pull/3184?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-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
[GitHub] [incubator-gobblin] autumnust commented on a change in pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
autumnust commented on a change in pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#discussion_r561455188
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergUtil.java
##########
@@ -0,0 +1,47 @@
+package org.apache.gobblin.writer;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
+import org.apache.iceberg.hadoop.HadoopTables;
+
+import java.util.Map;
+
+
+@Slf4j
+public class IcebergUtil {
+
+ public static Table createTable(String path, Map<String, String> properties, boolean partitioned, Schema schema) {
+ PartitionSpec spec;
+ if (partitioned) {
+ spec = PartitionSpec.builderFor(schema).identity("data").build();
Review comment:
Shall we get more specific identity value here?
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergTaskWriterFactory.java
##########
@@ -0,0 +1,64 @@
+package org.apache.gobblin.writer;
+
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.encryption.EncryptionManager;
+import org.apache.iceberg.io.*;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import java.util.Map;
+
+/**
+ * Factory to create {@link TaskWriter} that is.
+ * responsible for writing generic data
+ * into Iceberg table.
+ *
+ */
+public class IcebergTaskWriterFactory {
+
+ private final Schema schema;
+ private final PartitionSpec partitionSpec;
+ private final LocationProvider locations;
+ private final FileIO io;
+ private final EncryptionManager encryptionManager;
+ private final long targetFileSizeBytes;
+ private final FileFormat format;
+ private final IcebergFileAppenderFactory appenderFactory;
+
+ private transient OutputFileFactory outputFileFactory;
+
+ public IcebergTaskWriterFactory(Schema schema,
+ PartitionSpec partitionSpec,
+ LocationProvider locations,
+ FileIO io,
+ EncryptionManager encryptionManager,
+ long targetFileSizeBytes,
+ FileFormat format,
+ Map<String, String> tableProperties) {
+ this.schema = schema;
+ this.partitionSpec = partitionSpec;
+ this.locations = locations;
+ this.io = io;
+ this.encryptionManager = encryptionManager;
+ this.targetFileSizeBytes = targetFileSizeBytes;
+ this.format = format;
+ this.appenderFactory = new IcebergFileAppenderFactory(schema);
+ }
+
+ public void initialize(int taskId, int attemptId) {
+ this.outputFileFactory = new OutputFileFactory(partitionSpec, format, locations, io, encryptionManager, taskId, attemptId);
+ }
+
+ public TaskWriter create() {
+ Preconditions.checkNotNull(outputFileFactory,
+ "The outputFileFactory shouldn't be null if we have invoked the initialize().");
+
+ if (partitionSpec.fields().isEmpty()) {
Review comment:
So why we need this if-branch ?
##########
File path: gobblin-modules/gobblin-iceberg/src/main/java/org.apache.gobblin/writer/IcebergFileAppenderFactory.java
##########
@@ -0,0 +1,84 @@
+package org.apache.gobblin.writer;
+
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.MetricsConfig;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+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.io.TaskWriter;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+import java.io.IOException;
+import java.io.UncheckedIOException;
+import java.util.Map;
+
+
+/**
+ * Factory to create a new {@link FileAppender} that
+ * write generic data. The created {@link FileAppender}
+ * will be wrapped in {@link TaskWriter}
+ */
+public class IcebergFileAppenderFactory implements FileAppenderFactory {
Review comment:
shall we rename it to `GobblinFileAppender` if following those implementation in Iceberg codebase?
##########
File path: gobblin-modules/gobblin-iceberg/src/test/java/org/apache/gobblin/writer/IcebergTestUtil.java
##########
@@ -0,0 +1,130 @@
+package org.apache.gobblin.writer;
+
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.io.Decoder;
+import org.apache.avro.io.DecoderFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.*;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.hadoop.HadoopInputFile;
+import org.apache.iceberg.hadoop.HadoopTables;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileAppenderFactory;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Types;
+import org.testng.Assert;
+
+import java.io.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.iceberg.hadoop.HadoopOutputFile.fromPath;
+
+public class IcebergTestUtil {
+
+ public static final Schema SCHEMA = new Schema(
+ Types.NestedField.optional(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(2, "data", Types.StringType.get())
+ );
+
+ public static final Record RECORD = GenericRecord.create(SCHEMA);
+
+ public static Table createTable(String path, Map<String, String> properties, boolean partitioned) {
+ PartitionSpec spec;
+ if (partitioned) {
+ spec = PartitionSpec.builderFor(SCHEMA).identity("data").build();
+ } else {
+ spec = PartitionSpec.unpartitioned();
+ }
+ return new HadoopTables().create(SCHEMA, spec, properties, path);
+ }
+
+ public static Record createRecord(Integer id, String data) {
+ Record record = RECORD.copy();
+ record.setField("id", id);
+ record.setField("data", data);
+ return record;
+ }
+
+ public static void assertTableRecords(Table table, List<Record> expected) throws IOException {
+ table.refresh();
+ try (CloseableIterable<Record> iterable = IcebergGenerics.read(table).build()) {
+ Assert.assertEquals(
+ Sets.newHashSet(expected), Sets.newHashSet(iterable));
+ }
+ }
+
+ public static void assertTableRecords(String tablePath, List<Record> expected) throws IOException {
+ Preconditions.checkArgument(expected != null, "expected records shouldn't be null");
+ assertTableRecords(new HadoopTables().load(tablePath), expected);
+ }
+
+ public static org.apache.avro.Schema generateAvroSchema() {
+ org.apache.avro.Schema avroSchema = (new org.apache.avro.Schema.Parser()).parse("{\n" +
+ " \"namespace\": \"com.linkedin.orc\",\n" +
+ " \"type\": \"record\",\n" +
+ " \"name\": \"IcebergTest\",\n" +
+ " \"fields\": [\n" +
+ " {\n" +
+ " \"name\": \"id\",\n" +
+ " \"type\": \"int\"\n" +
+ " },\n" +
+ " {\n" +
+ " \"name\": \"data\",\n" +
+ " \"type\": \"string\"\n" +
+ " }\n" +
+ " ]\n" +
+ "}\n");
+ return avroSchema;
+ }
+
+ public static List<org.apache.avro.generic.GenericRecord> genericAvroRecords() throws IOException {
+ List<org.apache.avro.generic.GenericRecord> list = new ArrayList<>();
+ org.apache.avro.Schema avroSchema = generateAvroSchema();
+
+ GenericRecordBuilder builder_0 = new GenericRecordBuilder(avroSchema);
+ builder_0.set("id", 1);
+ builder_0.set("data", "alice");
+ list.add(builder_0.build());
+
+ GenericRecordBuilder builder_1 = new GenericRecordBuilder(avroSchema);
+ builder_1.set("id", 2);
+ builder_1.set("data", "bob");
+ list.add(builder_1.build());
+ return list;
+ }
+
+
+
+ public static List<org.apache.avro.generic.GenericRecord> deserializeAvroRecords(org.apache.avro.Schema schema, String schemaPath)
Review comment:
Where is this method being used? In fact I thought we will need to directly deal with avro file for deserialization ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [incubator-gobblin] autumnust commented on pull request #3184: Gobblin 1320 add iceberg writer
Posted by GitBox <gi...@apache.org>.
autumnust commented on pull request #3184:
URL: https://github.com/apache/incubator-gobblin/pull/3184#issuecomment-764178784
Also, there's a travis failure
----------------------------------------------------------------
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