Implement the parquet value reader & writer for apache flink #1125

openinx · 2020-06-18T11:33:57Z

No description provided.

openinx · 2020-06-22T01:48:01Z

Ping @rdblue , Thanks.

rdblue · 2020-06-23T00:48:56Z

Thanks! I'll give this a thorough review tomorrow.

rdblue · 2020-06-24T00:14:33Z

flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReaderWriter.java

+
+  @Test
+  public void testCorrectness() throws IOException {
+    int numRows = 2500;


How high have you tested? I usually test locally with ~1m rows.

Could you also add tests with the new generation methods that @samarthjain added for Parquet vectorization? Those allow you to generate data that will be dictionary encoded and that will fall back to non-dictionary after a few dictionary pages.

iceberg/spark/src/test/java/org/apache/iceberg/spark/data/RandomData.java

Lines 99 to 106 in 705da1b

public static Iterable<Record> generateFallbackData(Schema schema, int numRecords, long seed, long numDictRecords) {

return newIterable(() -> new FallbackDataGenerator(schema, seed, numDictRecords), schema, numRecords);

}

public static Iterable<GenericData.Record> generateDictionaryEncodableData(

Schema schema, int numRecords, long seed, float nullPercentage) {

return newIterable(() -> new DictionaryEncodedDataGenerator(schema, seed, nullPercentage), schema, numRecords);

}

OK, let me think about how to add those unit tests. Thanks.

rdblue · 2020-06-24T00:16:51Z

flink/src/test/java/org/apache/iceberg/flink/data/RandomData.java

+    return rows;
+  }
+
+  private static class RandomDataGenerator extends TypeUtil.CustomOrderSchemaVisitor<Object> {


Should this inherit from the one for Iceberg generics to avoid duplicating the list and map methods? Primitive and struct will need to be implemented, but this could reuse a lot there as well.

rdblue · 2020-06-24T00:24:35Z

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java

    }
  }

+  public interface StructWriterFactory {


Instead of adding this interface, what about adding WriteBuilder.createStructWriter to do this? That way, the Flink builder could just inherit from the generic builder and override that one method.

I think that would be cleaner because adding this interface requires also adding public methods to pass the factory. I'd rather not add those public methods if we can avoid it by adding a protected method and change the builder to a protected class.

I guess we will also need to mark the WriteBuilder from private to public because it will be accessed by flink classes from outside package.

It could be protected, right?

protected is not enough for the outside flink package, the WriterBuilder is a static class in org.apache.iceberg.data.parquet , it must be public so that other packages could inherit it.

You're right. The class should be public, but the method that will be overridden should be protected.

Em, I've implemented two versions. the first version provides a public interfaces with protected StructureWriterFactory and StructReaderFactory, the second version use the inherit WriterBuilder & ReaderBuilder #1125. For me , seems the first version looks much more concise. I plan to change to version#1 (with the public interface and protected methods ). Thanks.

Could you post the branch with the inheritance version? I'd like to see it to compare. I don't like the extra public classes and methods, and I think that inheritance would be a cleaner public API. I'm curious why you think the other approach looks more concise, though.

The inheritance version is here.

I'm curious why you think the other approach looks more concise, though.

Because the inheritance version will need to expose the buildReader and buildWriter logic , for example this. If we change those logic, will also need to change in flink reader/writers. seems the same logic with the GenericParquetReader.

Okay, how about inheriting from the outer class, then? That way, the interface and the method that accepts the factory could be protected. Mainly, I don't think these should be public.

rdblue · 2020-06-24T00:27:22Z

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java

+    StructReader<T, T> create(List<Type> types, List<ParquetValueReader<?>> readers, StructType struct);
+  }
+
+  static class RecordReader extends StructReader<GenericRecord, GenericRecord> {


Why change this to use the GenericRecord instance rather than the Record interface?

I don't see much value in this change. We will just need to change it back if we want to add implementations of Record that are not generic, like we do with our internal classes that extend Avro's IndexedRecord. Ideally, I'd like to change those over to use our generics readers eventually.

I changed this because the buildReader method return the GenericRecord ParquetValueReader. Seems better to change the ParquetValueReader to be Record reader.

rdblue · 2020-06-24T00:29:46Z

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java

  }

-  static class RecordReader extends StructReader<Record, Record> {
+  public interface StructReaderFactory<T> {


Similar to the write path, I think it would be nice to refactor this to avoid exposing new public methods and interfaces.

rdblue · 2020-06-24T00:30:31Z

Thanks @openinx! Overall this looks pretty close to ready to go in. I like that we can reuse so much between Flink and Iceberg generics.

openinx · 2020-06-25T15:12:07Z

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java


+  public interface StructWriterFactory {
+
+    StructWriter<?> create(List<ParquetValueWriter<?>> writers);


I tried to mark this method to be protected but seems java8 don't allow to do that....

Interface methods are always public.

rdblue · 2020-06-26T17:05:07Z

data/src/test/java/org/apache/iceberg/data/RandomGenericData.java

-
-    private RandomDataGenerator(long seed) {
-      this.random = new Random(seed);
+  private static class RandomRecordDataGenerator extends RandomDataGenerator<Record> {


How about RandomRecordGenerator instead of RandomRecordDataGenerator? I think that's more descriptive since it returns random records.

rdblue · 2020-06-26T17:08:50Z

flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkParquetReaderWriter.java

+import static org.apache.iceberg.types.Types.NestedField.required;
+
+public class TestFlinkParquetReaderWriter {
+  private static final int NUM_RECORDS = 1_000_000;


Good to know that this passes, but I don't think we need to run with this high of a record count every time in CI. Could you reduce this to 20_000?

rdblue · 2020-06-26T17:14:36Z

@openinx, I think this is ready to commit other than the public interfaces and methods to pass struct factories. I'd like to make those protected or go with the inheritance approach. Personally, I think the inheritance approach is cleaner and I don't mind the method you pointed out, but either way I think can work.

…r writer.

openinx · 2020-06-28T00:12:35Z

Ping @rdblue , Mind to take another look ? I think it's prepared to be merged now, will pull request the next patch about flink sink operator as we described in #1096

openinx · 2020-06-28T02:46:06Z

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

+  }
+
+  @SuppressWarnings("unchecked")
+  public static ParquetValueReader<Row> buildRowReader(Schema expectedSchema,


I changed the buildReader to buildRowReader because the parent buildReader will return with a ParquetValueReader <Record> data type, which clashes with this FlinkParquetReaders 's buildReader returned ParquetValueReader <Row>.

rdblue · 2020-06-29T17:58:51Z

Thanks, @openinx!

I appreciate you going with the inheritance approach for now. I'll prototype the alternative I suggested above, since I think that addresses your duplication concern while not making the factory interface public. I'm merging this so that it unblocks you and we can find a good solution to that problem in parallel.

openinx · 2020-07-02T04:14:36Z

@rdblue , Seems it's more clearer to abstract the generic logics in a separate BaseParquetReader<T> and make the RecordParquetReader to extend it. I implemented this in #1158, you may want to take a look. Thanks.

openinx added 2 commits June 18, 2020 19:29

Implement the parquet value reader & writer for apache flink

ab0c7e4

Fix the failure unit tests.

e33d2cd

rdblue reviewed Jun 24, 2020

View reviewed changes

openinx added 5 commits June 24, 2020 11:46

Remove the StructWriterFactory & StructReaderFactory.

1c37f6d

Revert the generic type in GenericParquetReaders#buildReader

f4799a0

Add test suits for DictionaryEncodedData and FallbackData

0350b9b

Revert to user the StructWriterFactory & StructReaderFactory

17845b1

Use the Record instead of GenericRecord

95e197a

openinx commented Jun 25, 2020

View reviewed changes

rdblue reviewed Jun 26, 2020

View reviewed changes

rdblue added this to the Flink Sink milestone Jun 27, 2020

Make the flink parquet reader/write inherit the generic parquet reade…

5bffced

…r writer.

openinx commented Jun 28, 2020

View reviewed changes

rdblue merged commit edb8d77 into apache:master Jun 29, 2020

openinx deleted the flink-parquet-io branch June 30, 2020 02:03

openinx mentioned this pull request Jul 2, 2020

Flink: Abstract the BaseParquetReaders so that GenericParquetReaders and FlinkParquetReaders could share the max part of the codes. #1162

Merged

rdblue mentioned this pull request Jul 2, 2020

Why do we need two avro record readers & writers ? #1152

Closed

cmathiesen pushed a commit to ExpediaGroup/iceberg that referenced this pull request Aug 19, 2020

Flink: Add Parquet value reader, writer implementations (apache#1125)

604996b

	public static Iterable<Record> generateFallbackData(Schema schema, int numRecords, long seed, long numDictRecords) {
	return newIterable(() -> new FallbackDataGenerator(schema, seed, numDictRecords), schema, numRecords);
	}

	public static Iterable<GenericData.Record> generateDictionaryEncodableData(
	Schema schema, int numRecords, long seed, float nullPercentage) {
	return newIterable(() -> new DictionaryEncodedDataGenerator(schema, seed, nullPercentage), schema, numRecords);
	}


		public interface StructWriterFactory {

		StructWriter<?> create(List<ParquetValueWriter<?>> writers);

Implement the parquet value reader & writer for apache flink #1125

Implement the parquet value reader & writer for apache flink #1125

Uh oh!

Conversation

openinx commented Jun 18, 2020

Uh oh!

openinx commented Jun 22, 2020

Uh oh!

rdblue commented Jun 23, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue Jun 25, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue Jun 24, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Jun 24, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Jun 26, 2020

Uh oh!

openinx commented Jun 28, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rdblue commented Jun 29, 2020

Uh oh!

openinx commented Jul 2, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

rdblue Jun 25, 2020 •

edited

Loading

rdblue Jun 24, 2020 •

edited

Loading