Parallelized read-write operations in Hoodie Merge phase #370

n3nash · 2018-04-02T05:00:44Z

Parallelized read-write operations in Hoodie Merge phase
Make BufferedIterator generic enough to be able to Buffer any type of payload, not just HoodieRecord.

ovj · 2018-04-02T05:20:34Z

hoodie-client/src/main/java/com/uber/hoodie/func/BufferedIterator.java

    // It caches the exception seen while fetching insert value.
    public Optional<Exception> exception = Optional.empty();

    public BufferedIteratorPayload(T record, Schema schema) {
      this.record = record;
-      try {
-        this.insertValue = record.getData().getInsertValue(schema);


why do we need to remove this? This is an expensive operation which we want to offload to reader thread?

Please avoid commenting on "WIP" PR's :)

n3nash · 2018-04-03T00:23:01Z

@vinothchandar Please take a pass at the approach, comments and java docs and code cleaning coming soon after we agree on the approach.

bvaradar

Decoupling Reading and Writing Parquet files from/to potentially remote server makes sense. Code changes looks good in general. I have added some minor comments.

bvaradar · 2018-04-04T01:50:36Z

hoodie-client/src/main/java/com/uber/hoodie/func/LazyInsertIterable.java

+                logger.info("starting hoodie writer thread");
+                // Passing parent thread's TaskContext to newly launched thread for it to access original TaskContext
+                // properties.
+                TaskContext$.MODULE$.setTaskContext(sparkThreadTaskContext);


Minor: It looks like there is a contract in using threads inside Spark Task and is repeated in update case too. Can we make a first class type (like SparkTaskThread) and handle this boiler-plate in one place.

TaskContext$.MODULE$.setTaskContext(sparkThreadTaskContext); this is the only boiler plate code here. The other code around is a little different in each scenario unfortunately. Trying to avoid introducing new first class types if possible.

bvaradar · 2018-04-04T02:02:05Z

hoodie-client/src/main/java/com/uber/hoodie/func/BufferedIterator.java


  public BufferedIterator(final Iterator<T> iterator, final long bufferMemoryLimit,
-      final Schema schema) {
+      final Schema schema, final PayloadFunction<T, R> payloadFunc) {


Minor: Rename PayloadFunction to BufferedIteratorPayloadFactory as it is used only in the context of BufferedIterator

bvaradar · 2018-04-04T02:04:11Z

hoodie-client/src/main/java/com/uber/hoodie/func/payload/AbstractBufferedIteratorPayload.java

+
+package com.uber.hoodie.func.payload;
+
+public class AbstractBufferedIteratorPayload<I, O> {


bvaradar · 2018-04-04T02:22:03Z

...e-client/src/main/java/com/uber/hoodie/func/payload/HoodieRecordBufferedIteratorPayload.java

+  public HoodieRecordBufferedIteratorPayload(HoodieRecord record, Schema schema) {
+    this.inputPayload = record;
+    try {
+      this.outputPayload = record.getData().getInsertValue(schema);


An observation: I am assuming this serialization is cpu intensive and with your changes, this work is decoupled from parquet write for new file-group generation case (Hoodie Create Handle). For the merge-case though, the deserialization seems to be tied up with parquet data fetch but they together are decoupled from parquet write.

Yes, that's true. The issue is that for CreateHandle the process isn't dependent on anything apart from the input, whereas in the MergeHandle, a merge process with the new records takes place. A larger refactor could help address this but I'm hoping to keep that for later.

lets capture the AIs from here into an issue?

bvaradar · 2018-04-04T02:30:04Z

hoodie-client/src/main/java/com/uber/hoodie/func/BufferedIterator.java

@@ -40,8 +38,8 @@
 * internally samples every {@link #RECORD_SAMPLING_RATE}th record and adjusts number of records in
 * buffer accordingly. This is done to ensure that we don't OOM.
 */
-public class BufferedIterator<K extends HoodieRecordPayload, T extends HoodieRecord<K>> implements
-    Iterator<BufferedIterator.BufferedIteratorPayload<T>> {
+public class BufferedIterator<T, R>


Can you add metrics around ingress, egress throughput and latency.

bvaradar · 2018-04-04T02:31:09Z

hoodie-client/src/main/java/com/uber/hoodie/func/PayloadFunction.java

+import com.uber.hoodie.func.payload.AbstractBufferedIteratorPayload;
+import java.util.function.Function;
+
+public abstract class PayloadFunction<T, R>


Minor: Make it interface as there is no implementation ?

n3nash · 2018-04-04T04:54:47Z

@bvaradar Thanks for your comments. Unfortunately, the PR wasn't ready for review yet apart from just discussing the approach. Since we agree on the approach and hopefully so does @vinothchandar, I'm going to make necessary code changes now. You might need to do another pass because of this, thanks!

n3nash · 2018-04-04T06:27:38Z

@bvaradar @vinothchandar Cleaned up the code, please take a pass at it.

vinothchandar

Overall fine with approach. On top of comments I left, like to see if we can avoid introducing a single abstraction BufferedIteratorTransform<I,O,F> which contains input, output and also the function. Function can just be a lambda as well.

vinothchandar · 2018-04-04T15:13:51Z

hoodie-client/src/main/java/com/uber/hoodie/func/ParquetReaderWrappedIterator.java

+  @Override
+  public boolean hasNext() {
+    try {
+      this.next = parquetReader.read();


lets handle the case where hasNext is handled without next being called and vice versa..

i.e make it true to iterator contract

vinothchandar · 2018-04-04T15:14:45Z

hoodie-client/src/main/java/com/uber/hoodie/func/payload/BufferedIteratorPayloadFunction.java

+ */
+public interface BufferedIteratorPayloadFunction<I, O>
+    extends Function<I, AbstractBufferedIteratorPayload<I, O>> {
+


remove extra line

vinothchandar · 2018-04-04T15:17:52Z

...e-client/src/main/java/com/uber/hoodie/func/payload/HoodieRecordBufferedIteratorPayload.java

+  public HoodieRecordBufferedIteratorPayload(HoodieRecord record, Schema schema) {
+    this.inputPayload = record;
+    try {
+      this.outputPayload = record.getData().getInsertValue(schema);


lets capture the AIs from here into an issue?

vinothchandar · 2018-04-04T15:19:39Z

hoodie-client/src/main/java/com/uber/hoodie/io/HoodieMergeHandle.java

@@ -136,7 +136,8 @@ private String init(String fileId, Iterator<HoodieRecord<T>> newRecordsItr) {
      // Load the new records in a map
      logger.info("MaxMemoryPerPartitionMerge => " + config.getMaxMemoryPerPartitionMerge());
      this.keyToNewRecords = new ExternalSpillableMap<>(config.getMaxMemoryPerPartitionMerge(),
-          Optional.empty(), new StringConverter(), new HoodieRecordConverter(schema, config.getPayloadClass()));
+          Optional.empty(), new StringConverter(),


we are here again..
@bvaradar @n3nash what could cause this folding/formatting change, if we have checkstyle & already pre formatted code?

These are wraps that happened column width 100. Checkstyle and code-style configuration under style/ folder have 120 as wrap point.
@n3nash : If you have not already imported style/intellij-java-google-style.xml, Can you use them.

I already use them, not sure what happened here.

ok, Digged a bit. @n3nash: Can you verify IntelliJ -> Preferences-> Code Style -> Java -> Wrapping and Braces is set to 100 (instead of 120).

Found an issue in IntelliJ where if you have existing code-styles (other than default) which are active, importing a new code-style does not seem to change this value. I am not able to reproduce this consistently though. Deleting all non-default code-styles and/or activating the Default Code-Style and then importing our code-style does the trick.

Let me know how it goes.

Balaji.V

I have the following :

Changing this to 120 fixes it.

vinothchandar · 2018-04-04T15:23:16Z

hoodie-client/src/main/java/com/uber/hoodie/table/HoodieCopyOnWriteTable.java

+        Future writerResult =
+            writerService.submit(
+                () -> {
+                  logger.info("starting hoodie writer thread");


this code seems repeated, is there a way to modularize and share code? Also revisit logger statements to make then more contextualized?

vinothchandar · 2018-04-04T15:24:04Z

hoodie-common/src/main/java/com/uber/hoodie/common/util/collection/DiskBasedMap.java

@@ -94,7 +95,9 @@ private void initFile(File writeOnlyFileHandle) throws IOException {
    }
    writeOnlyFileHandle.createNewFile();

-    log.info("Spilling to file location " + writeOnlyFileHandle.getAbsolutePath());
+    log.info("Spilling to file location " + writeOnlyFileHandle.getAbsolutePath() + " in machine ("


nit: rename machine to host

n3nash · 2018-04-04T22:35:42Z

@vinothchandar @bvaradar addressed CR comments.

@vinothchandar I started making changes for BufferedIteratorTransform but realized it's easier to reason in the code with function and payload separated, kept it that way.

bvaradar

Looks good to me.

vinothchandar · 2018-04-05T15:18:26Z

hoodie-client/src/main/java/com/uber/hoodie/func/LazyInsertIterable.java

-        handle = new HoodieCreateHandle(hoodieConfig, commitTime, hoodieTable,
-            payload.record.getPartitionPath());
-        handle.write(payload.record, payload.insertValue,
+        handle =


is this wrapping at 80 or 120?

handle = new HoodieCreateHandle(hoodieConfig, commitTime, hoodieTable, insertPayload.getPartitionPath());

is only 108 chars..

@bvaradar does checkstyle just check for < 120.

@n3nash if its formatting incorrectly, can you please fix and make a pass on all files again

it's formatting incorrectly, I made a pass on all files..

vinothchandar · 2018-04-05T15:25:51Z

hoodie-client/src/main/java/com/uber/hoodie/func/ParquetReaderWrappedIterator.java

+  // Holds the next entry returned by the parquet reader
+  private T next;
+  // Holds the current entry returned by the parquet reader
+  private T prev;


I think you can implement this using a single variable next ?

should we add a test for this file.

yeah, done.

vinothchandar · 2018-04-05T15:28:15Z

hoodie-client/src/main/java/com/uber/hoodie/func/BufferedIteratorWrapper.java

+ * needs to be buffered, the runnable function that needs to be executed in the reader thread and
+ * return the transformed output based on the writer function
+ */
+public class BufferedIteratorWrapper<I, O, E> {


lets rename to something more specific.. It feels like we are overloading Wrapper a lot. How about BufferedIteratorExecutor

Add a test here?

vinothchandar · 2018-04-05T15:29:15Z

hoodie-client/src/main/java/com/uber/hoodie/func/ParquetReaderWrappedIterator.java

+ * This class wraps a parquet reader and provides an iterator based api to
+ * read from a parquet file. This is used in {@link BufferedIterator}
+ */
+public class ParquetReaderWrappedIterator<T> implements Iterator<T> {


rename : ParquetReaderIterator or ParquetIterator

vinothchandar · 2018-04-05T15:30:09Z

hoodie-client/src/main/java/com/uber/hoodie/func/payload/BufferedIteratorPayloadFunction.java

+ * @param <I> input payload data type
+ * @param <O> output payload data type
+ */
+public interface BufferedIteratorPayloadFunction<I, O>


rename : BufferedIteratorTransform
(what it does is to transform I to O)

I think we should reconsider merging the Payload and the Transform here.. Current way introduces too many classes/interfaces for what we need to achieve.

I do agree it introduces extra classes to achieve something more simple...I like the cleaner separation of payload and transformer though.

Is there a unique advantage for having them separate? In this scenario, you are just looking for a a transform to turn I to O and it seems straightforward to have them together. Even if you look up something like Adapter pattern (https://en.wikipedia.org/wiki/Adapter_pattern) , the I,O,F are together.. (we don't need the target object to reflect source object changes though)

vinothchandar · 2018-04-05T15:36:07Z

...-client/src/main/java/com/uber/hoodie/func/payload/GenericRecordBufferedIteratorPayload.java

+    extends AbstractBufferedIteratorPayload<GenericRecord, GenericRecord> {
+
+  public GenericRecordBufferedIteratorPayload(GenericRecord record) {
+    this.inputPayload = record;


can we move this constructor to the abstract class above.. thats the typical pattern . or did you intend for it to be a Interface originally?

yeah, missed to refactor that after other changes, thanks for pointing that out.

vinothchandar · 2018-04-05T15:38:11Z

...src/main/java/com/uber/hoodie/func/payload/GenericRecordBufferedIteratorPayloadFunction.java

+    implements BufferedIteratorPayloadFunction<GenericRecord, AbstractBufferedIteratorPayload> {
+
+  @Override
+  public AbstractBufferedIteratorPayload apply(GenericRecord t) {


This class for e.g is why I strongly feel we should merge payload and transform into a single entity. We always use them to together and end up with a more complex Generic interfaces and micro implementations in the current way

in any case, lets name this class shorter? the Transform rename would make this GenericRecordBufferedIteratorTransform which is a tad shorter.

Addressed rename

vinothchandar · 2018-04-06T14:25:46Z

...e-client/src/main/java/com/uber/hoodie/func/payload/HoodieRecordBufferedIteratorPayload.java

+  // It caches the exception seen while fetching insert value.
+  public Optional<Exception> exception = Optional.empty();
+
+  public HoodieRecordBufferedIteratorPayload(HoodieRecord record, Schema schema) {


I think we should consider using a static method lambda if possible to capture the transform.. This whole class structure with constructor chaining/inheritance all seem overkill for the amount of work thats done in converting I to O. This is one of the reasons function passing was introduced in java - to avoid such boilerplate code. Can we take adv of that?

Refactored, please take a pass.

n3nash · 2018-04-09T20:25:32Z

@vinothchandar Refactored, please take a pass.

Co-authored-by: lokesh-lingarajan-0310 <[email protected]>

ovj reviewed Apr 2, 2018

View reviewed changes

n3nash force-pushed the parallelize_merge branch 3 times, most recently from 0b9c336 to 21a9166 Compare April 3, 2018 00:22

bvaradar reviewed Apr 4, 2018

View reviewed changes

n3nash force-pushed the parallelize_merge branch from 21a9166 to 3146eaa Compare April 4, 2018 06:27

n3nash changed the title ~~(WIP) Parallelized read-write operations in Hoodie Merge phase~~ Parallelized read-write operations in Hoodie Merge phase Apr 4, 2018

n3nash force-pushed the parallelize_merge branch from 3146eaa to f005b30 Compare April 4, 2018 06:32

vinothchandar reviewed Apr 4, 2018

View reviewed changes

n3nash force-pushed the parallelize_merge branch 2 times, most recently from 00c0f8a to 80fa5c6 Compare April 4, 2018 22:32

n3nash force-pushed the parallelize_merge branch from 80fa5c6 to e16d6c2 Compare April 4, 2018 22:39

bvaradar reviewed Apr 5, 2018

View reviewed changes

vinothchandar requested changes Apr 5, 2018

View reviewed changes

n3nash force-pushed the parallelize_merge branch 2 times, most recently from 699805e to 90ba1c3 Compare April 6, 2018 06:31

vinothchandar reviewed Apr 6, 2018

View reviewed changes

Parallelized read-write operations in Hoodie Merge phase

967a338

n3nash force-pushed the parallelize_merge branch from 90ba1c3 to 967a338 Compare April 9, 2018 20:22

vinothchandar approved these changes Apr 12, 2018

View reviewed changes

vinothchandar merged commit 720e42f into apache:master Apr 12, 2018

vinothchandar mentioned this pull request Apr 12, 2018

Replace BufferedIteratorPayload abstraction with function passing #375

Closed

vinishjail97 pushed a commit to vinishjail97/hudi that referenced this pull request Dec 15, 2023

Adding support s3 incr for snapshotLoadQuerySplitter (apache#370)

c27de42

Co-authored-by: lokesh-lingarajan-0310 <[email protected]>


		package com.uber.hoodie.func.payload;

		public class AbstractBufferedIteratorPayload<I, O> {

Parallelized read-write operations in Hoodie Merge phase #370

Parallelized read-write operations in Hoodie Merge phase #370

Conversation

n3nash commented Apr 2, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3nash commented Apr 3, 2018 • edited Loading

bvaradar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3nash Apr 4, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3nash commented Apr 4, 2018

n3nash commented Apr 4, 2018

vinothchandar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3nash commented Apr 4, 2018

bvaradar left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

n3nash commented Apr 9, 2018

n3nash commented Apr 2, 2018 •

edited

Loading

n3nash commented Apr 3, 2018 •

edited

Loading

n3nash Apr 4, 2018 •

edited

Loading