You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by jk...@apache.org on 2017/12/06 22:10:13 UTC
[beam] 02/02: add post-review updates
This is an automated email from the ASF dual-hosted git repository.
jkff pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 7564d822aa28a79f27bd0efa09a213ffd3635708
Author: Łukasz Gajowy <lu...@polidea.com>
AuthorDate: Thu Nov 30 15:59:59 2017 -0800
add post-review updates
---
.../beam/sdk/io/common/IOTestPipelineOptions.java | 3 +-
...FileBasedIOIT.java => FileBasedIOITHelper.java} | 50 +++++++++-------------
.../java/org/apache/beam/sdk/io/text/TextIOIT.java | 16 ++++---
.../apache/beam/sdk/io/tfrecord/TFRecordIOIT.java | 16 ++++---
4 files changed, 43 insertions(+), 42 deletions(-)
diff --git a/sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/IOTestPipelineOptions.java b/sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/IOTestPipelineOptions.java
index d919654..e7b475d 100644
--- a/sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/IOTestPipelineOptions.java
+++ b/sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/IOTestPipelineOptions.java
@@ -19,6 +19,7 @@ package org.apache.beam.sdk.io.common;
import org.apache.beam.sdk.options.Default;
import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.Validation;
import org.apache.beam.sdk.testing.TestPipelineOptions;
/**
@@ -96,7 +97,7 @@ public interface IOTestPipelineOptions extends TestPipelineOptions {
void setNumberOfRecords(Long count);
@Description("Destination prefix for files generated by the test")
- @Default.String("FILEBASEDIOIT")
+ @Validation.Required
String getFilenamePrefix();
void setFilenamePrefix(String prefix);
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/AbstractFileBasedIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/FileBasedIOITHelper.java
similarity index 66%
rename from sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/AbstractFileBasedIOIT.java
rename to sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/FileBasedIOITHelper.java
index 9eb8aea..cf20d8e 100644
--- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/AbstractFileBasedIOIT.java
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/common/FileBasedIOITHelper.java
@@ -18,47 +18,44 @@
package org.apache.beam.sdk.io.common;
-import com.google.common.base.Function;
-import com.google.common.collect.FluentIterable;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterables;
import java.io.IOException;
-import java.util.Collection;
import java.util.Collections;
import java.util.Date;
+import java.util.HashSet;
import java.util.Map;
-import org.apache.beam.sdk.io.Compression;
+import java.util.Set;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.fs.MatchResult;
import org.apache.beam.sdk.io.fs.ResourceId;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.PipelineOptionsValidator;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.DoFn;
/**
- * Abstract class for file based IO Integration tests.
+ * Contains helper methods for file based IO Integration tests.
*/
-public abstract class AbstractFileBasedIOIT {
+public class FileBasedIOITHelper {
- protected static IOTestPipelineOptions readTestPipelineOptions() {
- PipelineOptionsFactory.register(IOTestPipelineOptions.class);
- return TestPipeline.testingPipelineOptions().as(IOTestPipelineOptions.class);
+ private FileBasedIOITHelper() {
}
- protected static String appendTimestampToPrefix(String filenamePrefix) {
- return String.format("%s_%s", filenamePrefix, new Date().getTime());
+ public static IOTestPipelineOptions readTestPipelineOptions() {
+ PipelineOptionsFactory.register(IOTestPipelineOptions.class);
+ IOTestPipelineOptions options = TestPipeline
+ .testingPipelineOptions()
+ .as(IOTestPipelineOptions.class);
+
+ return PipelineOptionsValidator.validate(IOTestPipelineOptions.class, options);
}
- protected static Compression parseCompressionType(String compressionType) {
- try {
- return Compression.valueOf(compressionType.toUpperCase());
- } catch (IllegalArgumentException ex) {
- throw new IllegalArgumentException(
- String.format("Unsupported compression type: %s", compressionType));
- }
+ public static String appendTimestampToPrefix(String filenamePrefix) {
+ return String.format("%s_%s", filenamePrefix, new Date().getTime());
}
- protected String getExpectedHashForLineCount(Long lineCount) {
+ public static String getExpectedHashForLineCount(Long lineCount) {
Map<Long, String> expectedHashes = ImmutableMap.of(
100_000L, "4c8bb3b99dcc59459b20fefba400d446",
1_000_000L, "9796db06e7a7960f974d5a91164afff1",
@@ -78,6 +75,7 @@ public abstract class AbstractFileBasedIOIT {
* Constructs text lines in files used for testing.
*/
public static class DeterministicallyConstructTestTextLineFn extends DoFn<Long, String> {
+
@ProcessElement
public void processElement(ProcessContext c) {
c.output(String.format("IO IT Test line of text. Line seed: %s", c.element()));
@@ -94,18 +92,12 @@ public abstract class AbstractFileBasedIOIT {
MatchResult match = Iterables
.getOnlyElement(FileSystems.match(Collections.singletonList(c.element())));
- Collection<ResourceId> resourceIds = toResourceIds(match);
+ Set<ResourceId> resourceIds = new HashSet<>();
+ for (MatchResult.Metadata metadataElem : match.metadata()) {
+ resourceIds.add(metadataElem.resourceId());
+ }
FileSystems.delete(resourceIds);
}
- private Collection<ResourceId> toResourceIds(MatchResult match) throws IOException {
- return FluentIterable.from(match.metadata())
- .transform(new Function<MatchResult.Metadata, ResourceId>() {
- @Override
- public ResourceId apply(MatchResult.Metadata metadata) {
- return metadata.resourceId();
- }
- }).toList();
- }
}
}
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
index 7593f85..f9fad80 100644
--- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/text/TextIOIT.java
@@ -19,12 +19,15 @@
package org.apache.beam.sdk.io.text;
import static org.apache.beam.sdk.io.Compression.AUTO;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampToPrefix;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.getExpectedHashForLineCount;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.readTestPipelineOptions;
import java.text.ParseException;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TextIO;
-import org.apache.beam.sdk.io.common.AbstractFileBasedIOIT;
+import org.apache.beam.sdk.io.common.FileBasedIOITHelper;
import org.apache.beam.sdk.io.common.HashingFn;
import org.apache.beam.sdk.io.common.IOTestPipelineOptions;
import org.apache.beam.sdk.testing.PAssert;
@@ -49,7 +52,7 @@ import org.junit.runners.JUnit4;
* -Dit.test=org.apache.beam.sdk.io.text.TextIOIT
* -DintegrationTestPipelineOptions='[
* "--numberOfRecords=100000",
- * "--filenamePrefix=FILEBASEDIOIT"
+ * "--filenamePrefix=output_file_path",
* "--compressionType=GZIP"
* ]'
* </pre>
@@ -58,7 +61,7 @@ import org.junit.runners.JUnit4;
* running this test using Beam performance testing framework.</p>
*/
@RunWith(JUnit4.class)
-public class TextIOIT extends AbstractFileBasedIOIT {
+public class TextIOIT {
private static String filenamePrefix;
private static Long numberOfTextLines;
@@ -73,7 +76,7 @@ public class TextIOIT extends AbstractFileBasedIOIT {
numberOfTextLines = options.getNumberOfRecords();
filenamePrefix = appendTimestampToPrefix(options.getFilenamePrefix());
- compressionType = parseCompressionType(options.getCompressionType());
+ compressionType = Compression.valueOf(options.getCompressionType());
}
@Test
@@ -86,7 +89,8 @@ public class TextIOIT extends AbstractFileBasedIOIT {
PCollection<String> testFilenames = pipeline
.apply("Generate sequence", GenerateSequence.from(0).to(numberOfTextLines))
- .apply("Produce text lines", ParDo.of(new DeterministicallyConstructTestTextLineFn()))
+ .apply("Produce text lines",
+ ParDo.of(new FileBasedIOITHelper.DeterministicallyConstructTestTextLineFn()))
.apply("Write content to files", write)
.getPerDestinationOutputFilenames().apply(Values.<String>create());
@@ -97,7 +101,7 @@ public class TextIOIT extends AbstractFileBasedIOIT {
String expectedHash = getExpectedHashForLineCount(numberOfTextLines);
PAssert.thatSingleton(consolidatedHashcode).isEqualTo(expectedHash);
- testFilenames.apply("Delete test files", ParDo.of(new DeleteFileFn())
+ testFilenames.apply("Delete test files", ParDo.of(new FileBasedIOITHelper.DeleteFileFn())
.withSideInputs(consolidatedHashcode.apply(View.<String>asSingleton())));
pipeline.run().waitUntilFinish();
diff --git a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
index 4589942..b887316 100644
--- a/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
+++ b/sdks/java/io/file-based-io-tests/src/test/java/org/apache/beam/sdk/io/tfrecord/TFRecordIOIT.java
@@ -19,12 +19,15 @@
package org.apache.beam.sdk.io.tfrecord;
import static org.apache.beam.sdk.io.Compression.AUTO;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.appendTimestampToPrefix;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.getExpectedHashForLineCount;
+import static org.apache.beam.sdk.io.common.FileBasedIOITHelper.readTestPipelineOptions;
import java.text.ParseException;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.GenerateSequence;
import org.apache.beam.sdk.io.TFRecordIO;
-import org.apache.beam.sdk.io.common.AbstractFileBasedIOIT;
+import org.apache.beam.sdk.io.common.FileBasedIOITHelper;
import org.apache.beam.sdk.io.common.HashingFn;
import org.apache.beam.sdk.io.common.IOTestPipelineOptions;
import org.apache.beam.sdk.testing.PAssert;
@@ -51,7 +54,7 @@ import org.junit.runners.JUnit4;
* -Dit.test=org.apache.beam.sdk.io.tfrecord.TFRecordIOIT
* -DintegrationTestPipelineOptions='[
* "--numberOfRecords=100000",
- * "--filenamePrefix=FILEBASEDIOIT"
+ * "--filenamePrefix=output_file_path",
* "--compressionType=GZIP"
* ]'
* </pre>
@@ -60,7 +63,7 @@ import org.junit.runners.JUnit4;
* running this test using Beam performance testing framework.</p>
*/
@RunWith(JUnit4.class)
-public class TFRecordIOIT extends AbstractFileBasedIOIT {
+public class TFRecordIOIT {
private static String filenamePrefix;
private static Long numberOfTextLines;
@@ -78,7 +81,7 @@ public class TFRecordIOIT extends AbstractFileBasedIOIT {
numberOfTextLines = options.getNumberOfRecords();
filenamePrefix = appendTimestampToPrefix(options.getFilenamePrefix());
- compressionType = parseCompressionType(options.getCompressionType());
+ compressionType = Compression.valueOf(options.getCompressionType());
}
private static String createFilenamePattern() {
@@ -96,7 +99,8 @@ public class TFRecordIOIT extends AbstractFileBasedIOIT {
writePipeline
.apply("Generate sequence", GenerateSequence.from(0).to(numberOfTextLines))
- .apply("Produce text lines", ParDo.of(new DeterministicallyConstructTestTextLineFn()))
+ .apply("Produce text lines",
+ ParDo.of(new FileBasedIOITHelper.DeterministicallyConstructTestTextLineFn()))
.apply("Transform strings to bytes", MapElements.via(new StringToByteArray()))
.apply("Write content to files", writeTransform);
@@ -112,7 +116,7 @@ public class TFRecordIOIT extends AbstractFileBasedIOIT {
PAssert.thatSingleton(consolidatedHashcode).isEqualTo(expectedHash);
readPipeline.apply(Create.of(filenamePattern))
- .apply("Delete test files", ParDo.of(new DeleteFileFn())
+ .apply("Delete test files", ParDo.of(new FileBasedIOITHelper.DeleteFileFn())
.withSideInputs(consolidatedHashcode.apply(View.<String>asSingleton())));
readPipeline.run().waitUntilFinish();
}
--
To stop receiving notification emails like this one, please contact
"commits@beam.apache.org" <co...@beam.apache.org>.