[BEAM-59] Convert WriteFiles/FileBasedSink from IOChannelFactory to FileSystems #2779

dhalperi · 2017-04-29T03:56:15Z

This converts FileBasedSink from IOChannelFactory to FileSystems, with
fallout changes on all existing Transforms that use WriteFiles.

We preserve the existing semantics of most transforms, simply adding the
ability for users to provide ResourceId in addition to String when
setting the outputPrefix.

Other changes:

Make DefaultFilenamePolicy its own top-level class and move
IOChannelUtils#constructName into it. This the default FilenamePolicy
used by FilebasedSource.
Rethink FilenamePolicy as a function from ResourceId (base directory)
to ResourceId (output file), moving the base directory into the
context. This way, FilenamePolicy logic is truly independent from the
base directory. Using ResourceId#resolve, a filename policy can add
multiple path components, say, base/YYYY/MM/DD/file.txt, in a
fileystem independent way.

(Also add an optional extension parameter to the function, enabling an
owning transform to pass in the suffix from a separately-configured
compression factory or similar.)

TODO:

I cleaned up TextIO and AvroIO, but XmlIO and TFRecordIO need more.
Review test coverage.
REALLY review testing and javadoc.

But getting this out to be able to look at the comprehensive diff.

CC: @davorbonaci @lukecwik @vikkyrk @jkff @reuvenlax @kennknowles

dhalperi · 2017-04-29T04:01:03Z

...ion-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java

-            .matches(appliedWrite(withCustomSharding)),
-        is(false));
-  }
+//  @Test


@dhalperi todo

kennknowles · 2017-04-29T04:13:39Z

This seems like a bunch of different changes in one commit. Can you split?

dhalperi · 2017-04-29T04:40:17Z

That was cc, not R. And I have been splitting

…

On Fri, Apr 28, 2017 at 21:13 Kenn Knowles ***@***.***> wrote: This seems like a bunch of different changes in one commit. Can you split? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#2779 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAgITxM7LY_fvbYYvMCCXDZxMNadoWMnks5r0rj2gaJpZM4NMIzO> .

reuvenlax · 2017-04-29T18:17:49Z

...ion-java/src/test/java/org/apache/beam/runners/core/construction/PTransformMatchersTest.java

+//        PTransformMatchers.writeWithRunnerDeterminedSharding()
+//            .matches(appliedWrite(withCustomSharding)),
+//        is(false));
+//  }


ping when tests are done, and I will come back to this file

reuvenlax · 2017-04-29T18:17:59Z

...s/direct-java/src/test/java/org/apache/beam/runners/direct/WriteWithShardingFactoryTest.java

+//    public FileBasedWriteOperation<Object> createWriteOperation() {
+//      throw new IllegalArgumentException("Should not be used");
+//    }
+//  }


reuvenlax · 2017-04-29T18:22:53Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+     * {@link ResourceId#getFilename() filename} of the output resource will be used to set the file
+     * prefix. Files are additionally named by a shard identifier (see
+     * {@link Bound#withNumShards(int)}) and can be configured with a common suffix using
+     * {@link Bound#withSuffix(String)}.


some of this (e.g. withSuffix) applies only to DefaultFilenamePolicy

reuvenlax · 2017-04-29T18:23:41Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

      final boolean windowedWrites;
-      FileBasedSink.FilenamePolicy filenamePolicy;
+      @Nullable final FileBasedSink.FilenamePolicy filenamePolicy;


why Nullable?

reuvenlax · 2017-04-29T18:25:35Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+          // /some/dir as the output directory and file- as the prefix.
+          //
+          // But if this fails, say for /some/dir/ , fallback to treating it as a directory.
+          return to(FileSystems.matchNewResource(outputPrefix, false /* isDirectory */));


AFAICT FileSystems.matchNewResource does not throw

reuvenlax · 2017-04-29T18:27:00Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

@@ -694,11 +614,10 @@ private Read() {}
       */
      public <X> Bound<X> withSchema(Class<X> type) {
        return new Bound<>(
-            name,
-            filenamePrefix,
+            outputPrefix,


Might want to consider AutoValue (I believe jkff already has that coded up). Right now this feels a bit error prone, especially with multiple string arguments that are easy to mix up

reuvenlax · 2017-04-29T18:30:06Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+      @Override
+      public ResourceId apply(ResourceId input) {
+        return input.getCurrentDirectory();
+      }


Maybe belongs somewhere more common? Seems like it might be useful elsewhere (e.g. TextIO)

reuvenlax · 2017-04-29T18:32:34Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/DefaultFilenamePolicy.java

+            .withLabel("Filename Pattern"));
+  }
+
+  private static class ExtractFilename implements SerializableFunction<ResourceId, String> {


same as ExtractDirectory - I feel like this could be somewhere more common

reuvenlax · 2017-04-29T18:33:38Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

@@ -297,19 +253,19 @@ public int getNumShards() {
     * well as sharding information. The policy must return unique and consistent filenames
     * for different windows and panes.
     */
-    public abstract String windowedFilename(WindowedContext c);
+    public abstract ResourceId windowedFilename(
+        ResourceId outputDirectory, WindowedContext c, String extension);


Can you explain extension?

Is there a reason this can't be in WindowedContext? Where does extension come from?

Alternatively, maybe this better belongs in DefaultFilenamePolicy

The FileBasedSink already has to know the CompressionFactory to open the channel. It doesn't make sense to make every policy support passing in the compression factory just to make it know about the extension. This is contextual information, in the sense that every sink needs it and it's not valid across different applications of the policy.

I believe @kennknowles is deleting the contexts here; this PR simply makes that not-worse.

reuvenlax · 2017-04-29T18:37:08Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

      // The intent of the code is to have a consistent value of tempDirectory across
      // all workers, which wouldn't happen if now() was called inline.
-      Instant now = Instant.now();
+      private final String timestamp = Instant.now().toString(TEMPDIR_TIMESTAMP);
+      private final Long tempId = TEMP_COUNT.getAndIncrement();


Assume this is to handle the case of multiple FileBasedSinks that all get evaluated within the same millisecond at graph-construction time. If so, please add a comment to that effect.

reuvenlax · 2017-04-29T18:39:18Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

      // writerResults won't contain destination filenames, so we dynamically generate them here.
      if (files.size() > 0) {
        checkArgument(outputFilenames.isEmpty());
        // Sort files for idempotence.
-        files = Ordering.natural().sortedCopy(files);
-        FilenamePolicy filenamePolicy = getSink().fileNamePolicy;
+        files = Ordering.usingToString().sortedCopy(files);


Better to inject a comparator here. People may in the future change toString without realizing that we depend on it for semantics

ResourceId#toString has documented semantics, so this is not a valid concern.

reuvenlax · 2017-04-29T18:41:26Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

@@ -739,7 +614,7 @@ protected final void removeTemporaryFiles(Set<String> knownFiles,
     * Provides a coder for {@link FileBasedSink.FileResult}.
     */
    public final Coder<FileResult> getFileResultCoder() {
-      return FileResultCoder.of();
+      return SerializableCoder.of(FileResult.class);


Why did you change this back? FileResultCoder was explicitly added, because SerializableCoder is somewhat broken in streaming.

reuvenlax · 2017-04-29T18:45:15Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-  /**
-   * A coder for FileResult objects.
-   */
-  public static final class FileResultCoder extends CustomCoder<FileResult> {


Did you delete this because of ResourceId? This will break streaming update in a fundamental way. Compatibility checks will succeed (as the coder hasn't changed), but Java may fail to deserialize objects after update.

I can't find any Beam documentation that mentions update compatibility. Can you link the spec? I'm not sure what guiding principles you're using here.

Great question!

"Update" is a feature of the Dataflow runner, that actually predates Beam. It allows updating a running streaming pipeline with new code, even allowing certain changes to the graph structure. This has not yet been proposed as part of the Beam model, though it could be and should be.

While update is not yet formally a part of Beam, we should endeavor not to gratuitously break runner features - especially features that are widely used. Similarly Flink supports resuming a pipeline from a savepoint, and we should try not to break that.

In this particular case "update compatibility" (which means ensuring that a new graph structure can resume from an old graph) is not the issue. The issue is that the Java Serializable structure for objects is extremely prone to changing. The user might make simple code changes that do not at all change the graph structure of the pipeline, yet will change the serialized format of the object. This means that the new pipeline will start crashing at runtime, because it will no longer be able to deserialize it's own stored data.

Supporting undocumented runner features is not sustainable. We really need to make sure all such things are explicitly captured in the Beam model.

reuvenlax · 2017-04-29T18:47:57Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-    private final String filename;
-    private final String destinationFilename;
+  public static final class FileResult implements Serializable {
+    private final ResourceId filename;


Can we resolve these into Strings before creating the FileResult? That way we can keep the coder.

An alternate idea - add a getCoder() method to ResourceId that subclasses must implement.

ResourceId does not have a coder as the type of resource is not known – and not knowable! – at construction time.

dhalperi · 2017-04-30T17:05:04Z

Thanks for initial comments, but as I said in the PR desc this is just CC: folks to let them know where things are at.

Makes the changes in apache#2779 more standalone

Makes the changes in #2779 more standalone

coveralls · 2017-05-01T08:21:28Z

Coverage decreased (-0.3%) to 69.812% when pulling 93e2898 on dhalperi:convert-file-based-sink into 1197bef on apache:master.

dhalperi · 2017-05-01T23:26:20Z

retest this please

coveralls · 2017-05-02T01:17:22Z

Coverage decreased (-0.04%) to 70.349% when pulling 0f9db04 on dhalperi:convert-file-based-sink into cd813fb on apache:master.

dhalperi · 2017-05-02T17:27:39Z

Restored file result coder, rebased on top of AvroIO changes, and tests are green locally.

I'm still reviewing test coverage, but this is ready for a bigger look.

coveralls · 2017-05-02T18:18:28Z

Changes Unknown when pulling 441b9d8 on dhalperi:convert-file-based-sink into ** on apache:master**.

coveralls · 2017-05-02T19:43:12Z

Coverage decreased (-3.4%) to 66.499% when pulling 441b9d8 on dhalperi:convert-file-based-sink into 4c3174b on apache:master.

reuvenlax · 2017-05-02T19:51:05Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

@@ -297,19 +253,19 @@ public int getNumShards() {
     * well as sharding information. The policy must return unique and consistent filenames
     * for different windows and panes.
     */
-    public abstract String windowedFilename(WindowedContext c);
+    public abstract ResourceId windowedFilename(
+        ResourceId outputDirectory, WindowedContext c, String extension);


Is there a reason this can't be in WindowedContext? Where does extension come from?

reuvenlax · 2017-05-02T19:52:24Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java


+  private static class ExtractDirectory implements SerializableFunction<ResourceId, ResourceId> {


Have a generic ResourceIdHelpers class?

I'm not currently in the business of providing users with reusable utilities. We can certainly do that in follow-up work if we want.

reuvenlax · 2017-05-02T19:54:28Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

@@ -297,19 +253,19 @@ public int getNumShards() {
     * well as sharding information. The policy must return unique and consistent filenames
     * for different windows and panes.
     */
-    public abstract String windowedFilename(WindowedContext c);
+    public abstract ResourceId windowedFilename(
+        ResourceId outputDirectory, WindowedContext c, String extension);


Alternatively, maybe this better belongs in DefaultFilenamePolicy

reuvenlax · 2017-05-02T20:53:29Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-    protected final void copyToOutputFiles(Map<String, String> filenames,
-                                           PipelineOptions options)
+    @VisibleForTesting
+    final void copyToOutputFiles(Map<ResourceId, ResourceId> filenames)


s/filenames/resourceIds/

reuvenlax · 2017-05-02T20:54:52Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-      return FileResultCoder.of();
-    }
-
-    /**


Have we come up with a solution forFileResultCoder?

reuvenlax · 2017-05-02T21:07:25Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

@@ -173,9 +172,9 @@

  private static <T> Write.Builder<T> defaultWriteBuilder() {
    return new AutoValue_AvroIO_Write.Builder<T>()
-        .setFilenameSuffix("")
+        .setFilenameSuffix(null)
+        .setShardTemplate(null)


why change this to null?

So we know whether they're set.

reuvenlax · 2017-05-02T21:08:47Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

    /**
-     * Forces a single file as output.
+     * Forces a single file as output. This option is only compatible with unwindowed writes.


why? I think it's compatible with windowed writes.

coveralls · 2017-05-02T21:26:38Z

Coverage decreased (-3.4%) to 66.496% when pulling 81c0db4 on dhalperi:convert-file-based-sink into 4c3174b on apache:master.

coveralls · 2017-05-03T02:05:36Z

Changes Unknown when pulling c8c451b on dhalperi:convert-file-based-sink into ** on apache:master**.

dhalperi · 2017-05-03T03:36:30Z

PTAL @jkff

jkff

Overall looks good. All my comments are minor enough that it's fine to self-merge after addressing them.

jkff · 2017-05-03T20:00:46Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-     * the filename. The {@link WindowedContext} object gives access to the window and pane, as
-     * well as sharding information. The policy must return unique and consistent filenames
+     * the file {@link ResourceId resource} to be created given the base output directory and an
+     * optional extension from {@link FileBasedSink} configuration (e.g., {@link CompressionType}).


In what sense is the extension optional: can it be null? if yes, mark nullable; if no, clarify

Fixed from optional to (possibly empty).

jkff · 2017-05-03T20:01:54Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-     * @return The base filename for all output files.
+     * When a sink has not requested windowed or triggered output, this method will be invoked to
+     * return the file {@link ResourceId resource} to be created given the base output directory and
+     * an optional extension applied by additional {@link FileBasedSink} configuration


Same about the extension

jkff · 2017-05-03T20:02:15Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

+     * <p>The {@link Context} object only provides sharding information, which is used by the policy
+     * to generate unique and consistent filenames.
+     *
+     * <p>Expected to be {@code null} when the {@link Context} has {@link Context#getNumShards()}


You mean this is expected to return null? Is there any symmetric expectation on windowedFilename?

Deleted and solved in FileBasedSink itself. This was a wart from Reuven's earlier work; now we'll simply not call the function to compute filename if numShards <= 0.

jkff · 2017-05-03T20:04:46Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

-
-  public FilenamePolicy getFileNamePolicy() {
-    return fileNamePolicy;
+  public FilenamePolicy getFilenamePolicy() {


jkff · 2017-05-03T20:10:54Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/FileBasedSink.java

+          srcFiles.add(srcDestPair.getKey());
+          dstFiles.add(srcDestPair.getValue());
+        }
+        FileSystems.copy(srcFiles, dstFiles, StandardMoveOptions.IGNORE_MISSING_FILES);


Was IGNORE_MISSING_FILES also the semantics of channelFactory.copy? Either way worth a comment.

yes it was; added a comment. Would like to clean up the logic here however to be more explicit and better support non-copying filesystems like HDFS.

jkff · 2017-05-03T20:45:30Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java

     */
    public Write to(String filenamePrefix) {
+      try {
+        ResourceId fileResource =


Same comments as in Avro.

jkff · 2017-05-03T20:46:46Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java

+     *
+     * <p>The name of the output files will be determined by the {@link FilenamePolicy} used.
+     *
+     * <p>By default, a {@link DefaultFilenamePolicy} will be used built using the specified prefix


Optional: I'm generally not thrilled about this much javadoc duplication - it tends to rot and get inconsistent. Here and in other IOs, I think it'd be nice to have only one of the overloads be "authoritative" and have others refer to it like Like {@link #thatOtherOverload()}, but ...

jkff · 2017-05-03T20:48:54Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java

-    @Override
-    public String windowedFilename(WindowedContext input) {
+    public ResourceId windowedFilename(
+        ResourceId outputDirectory, WindowedContext input, String extension) {
      String filename = outputFilePrefix + "-" + input.getWindow().toString() +  "-"


nit: this and the unwindowed one could benefit from some String.format

jkff · 2017-05-03T20:53:02Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOTest.java

-  @Test
-  @Category(ValidatesRunner.class)
-  @Ignore("[BEAM-436] DirectRunner ValidatesRunner tempLocation configuration insufficient")
-  public void testPrimitiveWriteDisplayData() throws IOException {


Why was this one deleted?

It used to be a test that display data is preserved by runners, exploiting the fact that p.apply() is intercepted by the runner. It never worked (hence it was @Ignored) and now with runner API the entire premise of the test is flawed.

jkff · 2017-05-03T20:54:00Z

sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java

-      return toBuilder().setFilenamePrefix(filenamePrefix).build();
+    public Write<T> to(String filenamePrefix) {
+      ResourceId resourceId;
+      try {


Same comments as Avro.

done with helper.

dhalperi

Thanks!

dhalperi · 2017-05-03T22:52:30Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/DefaultFilenamePolicy.java

+/**
+ * A default {@link FilenamePolicy} for unwindowed files. This policy takes four parameters:
+ *
+ * <ul>


dhalperi · 2017-05-03T22:52:51Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+          getFilenamePrefix() != null,
+          "Unable to populate DisplayData for invalid AvroIO.Write (unset output prefix).");
+      String outputPrefixString = null;
+      if (getFilenamePrefix().isAccessible()) {


ack. I know @swegner made an explicit choice here, though. Will look into again after.

dhalperi · 2017-05-03T22:53:24Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+     * a common suffix (if supplied using {@link #withSuffix(String)}). This default can be
+     * overridden using {@link #withFilenamePolicy(FilenamePolicy)}.
+     */
+    public Write<T> to(String outputPrefix) {


done. to(VP) and toResource(VP) since they need to have different names in type-erased Java or they're treated as the same function.

dhalperi · 2017-05-03T22:53:36Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/AvroIO.java

+     */
+    public Write<T> to(String outputPrefix) {
+      try {
+        ResourceId fileResource =


done and pulled this logic into a single helper in FileBasedSink.

dhalperi · 2017-05-03T22:56:00Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/DefaultFilenamePolicy.java

+   *
+   * <p>If provided, the suffix will be used; otherwise the files will have an empty suffix.
+   */
+  public static DefaultFilenamePolicy constructUsingStandardParameters(


dhalperi · 2017-05-03T23:01:21Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java

     */
    public Write to(String filenamePrefix) {
+      try {
+        ResourceId fileResource =


dhalperi · 2017-05-03T23:03:04Z

sdks/java/core/src/main/java/org/apache/beam/sdk/io/TextIO.java

+     *
+     * <p>The name of the output files will be determined by the {@link FilenamePolicy} used.
+     *
+     * <p>By default, a {@link DefaultFilenamePolicy} will be used built using the specified prefix


dhalperi · 2017-05-03T23:07:04Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/AvroIOTest.java

-    @Override
-    public String windowedFilename(WindowedContext input) {
+    public ResourceId windowedFilename(
+        ResourceId outputDirectory, WindowedContext input, String extension) {
      String filename = outputFilePrefix + "-" + input.getWindow().toString() +  "-"


dhalperi · 2017-05-03T23:08:14Z

sdks/java/core/src/test/java/org/apache/beam/sdk/io/TextIOTest.java

-  @Test
-  @Category(ValidatesRunner.class)
-  @Ignore("[BEAM-436] DirectRunner ValidatesRunner tempLocation configuration insufficient")
-  public void testPrimitiveWriteDisplayData() throws IOException {


It used to be a test that display data is preserved by runners, exploiting the fact that p.apply() is intercepted by the runner. It never worked (hence it was @Ignored) and now with runner API the entire premise of the test is flawed.

dhalperi · 2017-05-03T23:09:44Z

sdks/java/io/xml/src/main/java/org/apache/beam/sdk/io/xml/XmlIO.java

-      return toBuilder().setFilenamePrefix(filenamePrefix).build();
+    public Write<T> to(String filenamePrefix) {
+      ResourceId resourceId;
+      try {


done with helper.

This converts FileBasedSink from IOChannelFactory to FileSystems, with fallout changes on all existing Transforms that use WriteFiles. We preserve the existing semantics of most transforms, simply adding the ability for users to provide ResourceId in addition to String when setting the outputPrefix. Other changes: * Rethink FilenamePolicy as a function from ResourceId (base directory) to ResourceId (output file), moving the base directory into the context. This way, FilenamePolicy logic is truly independent from the base directory. Using ResourceId#resolve, a filename policy can add multiple path components, say, base/YYYY/MM/DD/file.txt, in a fileystem independent way. (Also add an optional extension parameter to the function, enabling an owning transform to pass in the suffix from a separately-configured compression factory or similar.) * Make DefaultFilenamePolicy its own top-level class and move IOChannelUtils#constructName into it. This the default FilenamePolicy used by FileBasedSink.

coveralls · 2017-05-04T00:44:53Z

Coverage increased (+0.09%) to 69.964% when pulling e190c7d on dhalperi:convert-file-based-sink into 2d22485 on apache:master.

coveralls · 2017-05-04T02:51:36Z

Coverage decreased (-0.08%) to 70.554% when pulling 7aaa59e on dhalperi:convert-file-based-sink into b4bafd0 on apache:master.

Makes the changes in apache#2779 more standalone

dhalperi commented Apr 29, 2017

View reviewed changes

reuvenlax reviewed Apr 29, 2017

View reviewed changes

lukecwik mentioned this pull request Apr 29, 2017

[BEAM-2005, BEAM-2030, BEAM-2031, BEAM-2032, BEAM-2033, BEAM-2070] Base implementation of HadoopFileSystem. #2777

Closed

4 tasks

dhalperi force-pushed the convert-file-based-sink branch from 1897a87 to 32b2dc3 Compare April 30, 2017 17:46

dhalperi added a commit to dhalperi/beam that referenced this pull request Apr 30, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile

4cd313a

Makes the changes in apache#2779 more standalone

dhalperi mentioned this pull request Apr 30, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile #2792

Closed

dhalperi force-pushed the convert-file-based-sink branch 3 times, most recently from 164bf51 to 435b21a Compare April 30, 2017 18:53

dhalperi added a commit to dhalperi/beam that referenced this pull request Apr 30, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile

440378d

Makes the changes in apache#2779 more standalone

dhalperi force-pushed the convert-file-based-sink branch 2 times, most recently from a76c223 to f4ab836 Compare April 30, 2017 23:01

dhalperi added a commit to dhalperi/beam that referenced this pull request Apr 30, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile

6aeaed3

Makes the changes in apache#2779 more standalone

asfgit pushed a commit that referenced this pull request May 1, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile

1ecc6eb

Makes the changes in #2779 more standalone

dhalperi force-pushed the convert-file-based-sink branch from 578d20b to 93e2898 Compare May 1, 2017 07:28

dhalperi force-pushed the convert-file-based-sink branch from 0fb538b to 85be61b Compare May 1, 2017 22:48

dhalperi force-pushed the convert-file-based-sink branch from 85be61b to 5cb56e5 Compare May 1, 2017 23:35

dhalperi force-pushed the convert-file-based-sink branch from 0f9db04 to 2029851 Compare May 2, 2017 01:55

dhalperi mentioned this pull request May 2, 2017

[BEAM-59] Minor changes that can be separated out from big PR #2814

Closed

dhalperi force-pushed the convert-file-based-sink branch from 2029851 to aa267d4 Compare May 2, 2017 16:47

dhalperi force-pushed the convert-file-based-sink branch from b7305d5 to 441b9d8 Compare May 2, 2017 18:15

reuvenlax reviewed May 2, 2017

View reviewed changes

dhalperi force-pushed the convert-file-based-sink branch 3 times, most recently from fcda6ab to c8c451b Compare May 3, 2017 01:02

jkff approved these changes May 3, 2017

View reviewed changes

dhalperi commented May 4, 2017

View reviewed changes

dhalperi added 2 commits May 3, 2017 17:07

review comments

7aaa59e

dhalperi force-pushed the convert-file-based-sink branch from e190c7d to 7aaa59e Compare May 4, 2017 00:44

asfgit closed this in 1bc50d6 May 4, 2017

dhalperi deleted the convert-file-based-sink branch May 4, 2017 17:36

peihe pushed a commit to peihe/incubator-beam that referenced this pull request Jul 21, 2017

[BEAM-59] Minor style cleanups to WriteOneWindowPerFile

9bd8bb1

Makes the changes in apache#2779 more standalone


		private static class ExtractDirectory implements SerializableFunction<ResourceId, ResourceId> {

[BEAM-59] Convert WriteFiles/FileBasedSink from IOChannelFactory to FileSystems #2779

[BEAM-59] Convert WriteFiles/FileBasedSink from IOChannelFactory to FileSystems #2779

Conversation

dhalperi commented Apr 29, 2017 • edited

Choose a reason for hiding this comment

kennknowles commented Apr 29, 2017

dhalperi commented Apr 29, 2017 via email

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

dhalperi Apr 30, 2017 • edited

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

dhalperi Apr 30, 2017 • edited

Choose a reason for hiding this comment

dhalperi commented Apr 30, 2017 • edited

coveralls commented May 1, 2017

dhalperi commented May 1, 2017

coveralls commented May 2, 2017

dhalperi commented May 2, 2017

coveralls commented May 2, 2017

coveralls commented May 2, 2017

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

coveralls commented May 2, 2017

coveralls commented May 3, 2017

dhalperi commented May 3, 2017

jkff 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

dhalperi 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

dhalperi commented Apr 29, 2017 •

edited

dhalperi Apr 30, 2017 •

edited

dhalperi Apr 30, 2017 •

edited

dhalperi commented Apr 30, 2017 •

edited