LROB-86: HadoopOutputFileIO #41

b923 · 2018-09-25T08:59:32Z

No description provided.

dmvk

This definitely heads the right direction! I've added a few comments. Good job 👍

...ut-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopOutputFormatIO.java

dmvk · 2018-09-25T10:19:37Z

...ut-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopOutputFormatIO.java

+      try {
+        JobID jobId = HadoopUtils.createJobId();
+        TaskAttemptContext setupTaskContext =
+            HadoopUtils.createSetupTaskContext(getConfiguration().get(), jobId);


I reckon this method should work with the configuration object from the passed argument. Maybe we can make this method static?

dmvk · 2018-09-25T10:28:51Z

...ut-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopOutputFormatIO.java

+      PCollectionView<SerializableConfiguration> configView = null;
+      if (input.getWindowingStrategy().equals(WindowingStrategy.globalDefault())) {
+        configView = createGlobalConfigCollectionView(input);
+      }


we should throw an exception for non default windowing

I will fix it in the streaming iteration

dmvk · 2018-09-25T10:31:47Z

...ut-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopOutputFormatIO.java

+      return PDone.in(
+          input
+              .apply(
+                  TRANSFORM_NAME + "/TaskAssignee",


there is not need to prefix transformation names. These will be always nested under user provided name.

dmvk · 2018-09-25T11:01:09Z

...ut-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopOutputFormatIO.java

+   * @param <KeyT>
+   * @param <ValueT>
+   */
+  private static class TaskIdContextHolder<KeyT, ValueT> {


How about TaskContext?

...doop-output-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopUtils.java

dmvk · 2018-09-25T12:08:12Z

...doop-output-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/HadoopUtils.java

+    }
+  }
+
+  static int getReducersCountFromConfig(Configuration conf) {


shorter getReducersCount(Configuration conf) & getPartitioner(Configuration conf) method names are ok :)

dmvk · 2018-09-25T12:08:48Z

...tput-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/IterableCombinerFn.java

+
+  /**
+   * Coder for {@link CollectionAccumulator} class.
+   * @param <T>


missing doc (javadoc task would fail)

dmvk · 2018-09-25T12:09:04Z

...tput-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/IterableCombinerFn.java

+    }
+  }
+
+}


dmvk · 2018-09-25T12:09:29Z

...tput-format/src/main/java/org/apache/beam/sdk/io/hadoop/outputformat/IterableCombinerFn.java

+ * Collects all items of defined type into one {@link Iterable} container.
+ * @param <T> Type of the elements to collect
+ */
+public class IterableCombinerFn<T>


👍 I guess this should be package private though

dmvk

This looks awesome, good job 👍 There are few more issues to address, but we are getting closer ;)

...java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java

dmvk · 2018-10-10T10:43:25Z

...java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java

+      Integer taskID = element.getKey();
+
+      if (bundleTaskContext == null) {
+        bundleTaskContext = setupTask(taskID, conf);


IMPORTANT: this seems weird, I think that we can receive multiple windowed taskids in the same bundle, am I missing something?

This will be probably more complicated..
I have found this article: https://beam.apache.org/documentation/execution-model/ which describes how are the data distributed across bundles. I will probably have to move task setup and commit from WriteFn to separate ParDos.
I will look on that more closely tommorow.

I still doubt if the WriteFn is written correctly.
There are following points which should be changed by my opinion:

Setup of the tasks should be moved into taskIds creation

Hash branch can do this for every partition after job setup

NonHash branch can do it after taskId lock acquisition

Commit of tasks should be done differently for Hash and non Hash branch:

Hash branch should commit task after all writes are done - for example before commitJob

NonHash branch should commit task after bundle processing (As it is now)

My final dubiety is releated to dependent parallellism failures hadnling in relation to WriteFn. If I understand to the document correctly than we cannot expect that KV pairs of one taksId will be processed in one bundle (at least for Hashing branch).

I am not sure about my thoughts, so I will need some advice probably...

...java/io/hadoop-format/src/main/java/org/apache/beam/sdk/io/hadoop/format/HadoopFormatIO.java

b923 requested a review from dmvk September 25, 2018 08:59

dmvk suggested changes Sep 25, 2018

View reviewed changes

dmvk mentioned this pull request Oct 4, 2018

[BEAM-3912] Add HadoopOutputFormatIO support apache/beam#6306

Closed

2 tasks

dmvk force-pushed the BEAM-3912-HadoopOutputFormatIO branch from 074af32 to 044a5b6 Compare October 4, 2018 16:18

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch from 81aa1ec to d298e6b Compare October 5, 2018 14:49

dmvk suggested changes Oct 10, 2018

View reviewed changes

dmvk approved these changes Oct 22, 2018

View reviewed changes

aromanenko-dev added 6 commits October 24, 2018 16:03

[BEAM-3912] Add HadoopOutputFormatIO support

8f15b88

[BEAM-3912] Remove useless dep

406f8d7

[BEAM-3912] Add HadoopOutputFormatIO support

86f723e

[BEAM-3912] Remove useless dep

9863c79

Fix typo in test name

fa9cc4a

[BEAM-3912] Implement HadoopFormatIO.Write

757b71e

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch from 933f417 to e457e04 Compare October 25, 2018 07:52

b923 changed the base branch from BEAM-3912-HadoopOutputFormatIO to master October 25, 2018 07:55

[BEAM-5309] Add streaming support for HadoopFormatIO

20e3e24

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch from 0150688 to 20e3e24 Compare October 31, 2018 08:28

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch 3 times, most recently from 4193ac9 to 721a422 Compare November 13, 2018 12:25

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch from 721a422 to b13e55d Compare November 19, 2018 15:44

[BEAM-5309] Add streaming support for HadoopFormatIO

4adc254

b923 force-pushed the b923/LROB-86-HadoopOutputFormatIO branch from b13e55d to 4adc254 Compare November 29, 2018 11:02

+                  }
+                }
+              }

                
                    No newline at end of file

LROB-86: HadoopOutputFileIO #41

Are you sure you want to change the base?

LROB-86: HadoopOutputFileIO #41

Uh oh!

Conversation

b923 commented Sep 25, 2018

Uh oh!

dmvk left a comment

Choose a reason for hiding this comment

Uh oh!

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!

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!

dmvk left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

b923 Oct 11, 2018 • 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!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

b923 Oct 11, 2018 •

edited

Loading