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/10/26 20:21:31 UTC
[2/3] beam git commit: Adds ParseResult.failure()
Adds ParseResult.failure()
Also groups the tests into fewer methods providing the same coverage.
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/c76c518b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/c76c518b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/c76c518b
Branch: refs/heads/master
Commit: c76c518b62eee22334ff77fd15ae4eeae089e828
Parents: ba93dd3
Author: Eugene Kirpichov <ki...@google.com>
Authored: Wed Oct 25 16:57:36 2017 -0700
Committer: Eugene Kirpichov <ki...@google.com>
Committed: Thu Oct 26 13:20:39 2017 -0700
----------------------------------------------------------------------
.../beam/sdk/testing/SuccessOrFailure.java | 21 +-
.../beam/sdk/util/SerializableThrowable.java | 49 +++++
sdks/java/io/tika/pom.xml | 6 +
.../apache/beam/sdk/io/tika/ParseResult.java | 104 +++++++---
.../org/apache/beam/sdk/io/tika/TikaIO.java | 192 +++++++++--------
.../beam/sdk/io/tika/ParseResultTest.java | 54 ++++-
.../org/apache/beam/sdk/io/tika/TikaIOTest.java | 208 +++++++------------
.../src/test/resources/apache-beam-tika-pdf.zip | Bin 11685 -> 0 bytes
.../src/test/resources/apache-beam-tika.pdf | Bin 12392 -> 0 bytes
.../src/test/resources/apache-beam-tika1.odt | Bin 12540 -> 0 bytes
.../src/test/resources/apache-beam-tika2.odt | Bin 11412 -> 0 bytes
.../resources/valid/apache-beam-tika-pdf.zip | Bin 0 -> 11685 bytes
.../test/resources/valid/apache-beam-tika.odt | Bin 0 -> 12540 bytes
13 files changed, 351 insertions(+), 283 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java
index a63bbcc..bac4eb3 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/testing/SuccessOrFailure.java
@@ -18,34 +18,17 @@
package org.apache.beam.sdk.testing;
import com.google.common.base.MoreObjects;
-import java.io.IOException;
-import java.io.ObjectInputStream;
import java.io.Serializable;
import javax.annotation.Nullable;
import org.apache.beam.sdk.coders.DefaultCoder;
import org.apache.beam.sdk.coders.SerializableCoder;
+import org.apache.beam.sdk.util.SerializableThrowable;
/**
* Output of {@link PAssert}. Passed to a conclude function to act upon.
*/
@DefaultCoder(SerializableCoder.class)
public final class SuccessOrFailure implements Serializable {
- private static final class SerializableThrowable implements Serializable {
- @Nullable private final Throwable throwable;
- @Nullable private final StackTraceElement[] stackTrace;
-
- private SerializableThrowable(@Nullable Throwable t) {
- this.throwable = t;
- this.stackTrace = (t == null) ? null : t.getStackTrace();
- }
-
- private void readObject(ObjectInputStream is) throws IOException, ClassNotFoundException {
- is.defaultReadObject();
- if (throwable != null) {
- throwable.setStackTrace(stackTrace);
- }
- }
- }
private final boolean isSuccess;
@Nullable
@@ -68,7 +51,7 @@ public final class SuccessOrFailure implements Serializable {
@Nullable
public AssertionError assertionError() {
- return site == null ? null : site.wrap(throwable.throwable);
+ return site == null ? null : site.wrap(throwable.getThrowable());
}
public static SuccessOrFailure success() {
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/SerializableThrowable.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/SerializableThrowable.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/SerializableThrowable.java
new file mode 100644
index 0000000..4951958
--- /dev/null
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/SerializableThrowable.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.util;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.Serializable;
+import javax.annotation.Nullable;
+
+/**
+ * A wrapper around {@link Throwable} that preserves the stack trace on serialization, unlike
+ * regular {@link Throwable}.
+ */
+public final class SerializableThrowable implements Serializable {
+ @Nullable private final Throwable throwable;
+ @Nullable private final StackTraceElement[] stackTrace;
+
+ public SerializableThrowable(@Nullable Throwable t) {
+ this.throwable = t;
+ this.stackTrace = (t == null) ? null : t.getStackTrace();
+ }
+
+ @Nullable
+ public Throwable getThrowable() {
+ return throwable;
+ }
+
+ private void readObject(ObjectInputStream is) throws IOException, ClassNotFoundException {
+ is.defaultReadObject();
+ if (throwable != null) {
+ throwable.setStackTrace(stackTrace);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/pom.xml b/sdks/java/io/tika/pom.xml
index d7f7e42..5e48099 100644
--- a/sdks/java/io/tika/pom.xml
+++ b/sdks/java/io/tika/pom.xml
@@ -67,6 +67,12 @@
</dependency>
<dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
<groupId>org.apache.beam</groupId>
<artifactId>beam-sdks-java-core</artifactId>
<classifier>tests</classifier>
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/ParseResult.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/ParseResult.java b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/ParseResult.java
index 0a77491..f78d603 100644
--- a/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/ParseResult.java
+++ b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/ParseResult.java
@@ -17,61 +17,95 @@
*/
package org.apache.beam.sdk.io.tika;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Throwables;
import java.io.Serializable;
import java.util.Arrays;
-
+import java.util.Objects;
+import javax.annotation.Nullable;
+import org.apache.beam.sdk.util.SerializableThrowable;
import org.apache.tika.metadata.Metadata;
/**
- * Tika parse result containing the file location, metadata
- * and content converted to String.
+ * The result of parsing a single file with Tika: contains the file's location, metadata, extracted
+ * text, and optionally an error. If there is an error, the metadata and extracted text may be
+ * partial (i.e. not represent the entire file).
*/
-@SuppressWarnings("serial")
public class ParseResult implements Serializable {
private final String fileLocation;
private final String content;
private final Metadata metadata;
private final String[] metadataNames;
+ @Nullable private final SerializableThrowable error;
+
+ public static ParseResult success(String fileLocation, String content, Metadata metadata) {
+ return new ParseResult(fileLocation, content, metadata, null);
+ }
- public ParseResult(String fileLocation, String content) {
- this(fileLocation, content, new Metadata());
+ public static ParseResult success(String fileLocation, String content) {
+ return new ParseResult(fileLocation, content, new Metadata(), null);
}
- public ParseResult(String fileLocation, String content, Metadata metadata) {
+ public static ParseResult failure(
+ String fileLocation, String partialContent, Metadata partialMetadata, Throwable error) {
+ return new ParseResult(fileLocation, partialContent, partialMetadata, error);
+ }
+
+ private ParseResult(String fileLocation, String content, Metadata metadata, Throwable error) {
+ checkArgument(fileLocation != null, "fileLocation can not be null");
+ checkArgument(content != null, "content can not be null");
+ checkArgument(metadata != null, "metadata can not be null");
this.fileLocation = fileLocation;
this.content = content;
this.metadata = metadata;
this.metadataNames = metadata.names();
+ this.error = (error == null) ? null : new SerializableThrowable(error);
+ }
+
+ /** Returns the absolute path to the input file. */
+ public String getFileLocation() {
+ return fileLocation;
+ }
+
+ /** Returns whether this file was parsed successfully. */
+ public boolean isSuccess() {
+ return error == null;
+ }
+
+ /** Returns the parse error, if the file was parsed unsuccessfully. */
+ public Throwable getError() {
+ checkState(error != null, "This is a successful ParseResult");
+ return error.getThrowable();
}
/**
- * Gets a file content.
+ * Same as {@link #getError}, but returns the complete stack trace of the error as a {@link
+ * String}.
*/
+ public String getErrorAsString() {
+ return Throwables.getStackTraceAsString(getError());
+ }
+
+ /** Returns the extracted text. May be partial, if this parse result contains a failure. */
public String getContent() {
return content;
}
- /**
- * Gets a file metadata.
- */
+ /** Returns the extracted metadata. May be partial, if this parse result contains a failure. */
public Metadata getMetadata() {
return metadata;
}
- /**
- * Gets a file location.
- */
- public String getFileLocation() {
- return fileLocation;
- }
-
@Override
public int hashCode() {
- int hashCode = 1;
- hashCode = 31 * hashCode + fileLocation.hashCode();
- hashCode = 31 * hashCode + content.hashCode();
- hashCode = 31 * hashCode + getMetadataHashCode();
- return hashCode;
+ return Objects.hash(
+ getFileLocation(),
+ getContent(),
+ getMetadataHashCode(),
+ isSuccess() ? "" : Throwables.getStackTraceAsString(getError()));
}
@Override
@@ -80,14 +114,16 @@ public class ParseResult implements Serializable {
return false;
}
- ParseResult pr = (ParseResult) obj;
- return this.fileLocation.equals(pr.fileLocation)
- && this.content.equals(pr.content)
- && this.metadata.equals(pr.metadata);
+ ParseResult other = (ParseResult) obj;
+ return Objects.equals(getFileLocation(), other.getFileLocation())
+ && Objects.equals(getContent(), other.getContent())
+ && Objects.equals(getMetadata(), other.getMetadata())
+ && (isSuccess()
+ ? other.isSuccess()
+ : (!other.isSuccess() && Objects.equals(getErrorAsString(), other.getErrorAsString())));
}
- //TODO:
- // Remove this function and use metadata.hashCode() once Apache Tika 1.17 gets released.
+ // TODO: Remove this function and use metadata.hashCode() once Apache Tika 1.17 gets released.
private int getMetadataHashCode() {
int hashCode = 0;
for (String name : metadataNames) {
@@ -95,4 +131,14 @@ public class ParseResult implements Serializable {
}
return hashCode;
}
+
+ @Override
+ public String toString() {
+ return MoreObjects.toStringHelper(this)
+ .add("fileLocation", fileLocation)
+ .add("content", "<" + content.length() + " chars>")
+ .add("metadata", metadata)
+ .add("error", getError() == null ? null : Throwables.getStackTraceAsString(getError()))
+ .toString();
+ }
}
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/TikaIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/TikaIO.java b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/TikaIO.java
index 32353e1..26f116d 100644
--- a/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/TikaIO.java
+++ b/sdks/java/io/tika/src/main/java/org/apache/beam/sdk/io/tika/TikaIO.java
@@ -17,15 +17,13 @@
*/
package org.apache.beam.sdk.io.tika;
+import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkNotNull;
import com.google.auto.value.AutoValue;
-
import java.io.InputStream;
import java.nio.channels.Channels;
-
import javax.annotation.Nullable;
-
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.FileIO;
@@ -49,68 +47,60 @@ import org.apache.tika.parser.Parser;
import org.apache.tika.sax.ToTextContentHandler;
import org.xml.sax.ContentHandler;
-
-
-
/**
- * A collection of {@link PTransform} transforms for parsing arbitrary files using Apache Tika.
- * Files in many well known text, binary or scientific formats can be processed.
+ * Transforms for parsing arbitrary files using <a href="https://tika.apache.org/">Apache Tika</a>.
+ *
+ * <p>Tika is able to extract text and metadata from files in many well known text, binary and
+ * scientific formats.
*
- * <p>{@link TikaIO.Parse} and {@link TikaIO.ParseAll} parse the files and return
- * a {@link PCollection} containing one {@link ParseResult} per each file.
+ * <p>The entry points are {@link #parse} and {@link #parseFiles}. They parse a set of files and
+ * return a {@link PCollection} containing one {@link ParseResult} per each file. {@link #parse}
+ * implements the common case of parsing all files matching a single filepattern, while {@link
+ * #parseFiles} should be used for all use cases requiring more control, in combination with {@link
+ * FileIO#match} and {@link FileIO#readMatches} (see their respective documentation).
*
- * <p>Combine {@link TikaIO.ParseAll} with {@link FileIO.Match}
- * and {@link FileIO.ReadMatches} to match, read and parse the files.
+ * <p>{@link #parse} does not automatically uncompress compressed files: they are passed to Tika
+ * as-is.
*
- * <p>Example:
+ * <p>It's possible that some files will partially or completely fail to parse. In that case, the
+ * respective {@link ParseResult} will be marked unsuccessful (see {@link ParseResult#isSuccess})
+ * and will contain the error, available via {@link ParseResult#getError}.
+ *
+ * <p>Example: using {@link #parse} to parse all PDF files in a directory on GCS.
*
* <pre>{@code
* Pipeline p = ...;
*
- * // A simple parse of a local PDF file (only runs locally):
* PCollection<ParseResult> results =
- * p.apply(FileIO.match().filepattern("/local/path/to/file.pdf"))
- * .apply(FileIO.readMatches())
- * .apply(TikaIO.parseFiles());
+ * p.apply(TikaIO.parse().filepattern("gs://my-bucket/files/*.pdf"));
* }</pre>
*
- * <p>Use {@link TikaIO.Parse} to match, read and parse the files in simple cases.
- *
- * <p>Example:
+ * <p>Example: using {@link #parseFiles} in combination with {@link FileIO} to continuously parse
+ * new PDF files arriving into the directory.
*
* <pre>{@code
* Pipeline p = ...;
*
- * // A simple parse of a local PDF file (only runs locally):
* PCollection<ParseResult> results =
- * p.apply(TikaIO.parseAll().filepattern("/local/path/to/file.pdf"));
+ * p.apply(FileIO.match().filepattern("gs://my-bucket/files/*.pdf")
+ * .continuously(...))
+ * .apply(FileIO.readMatches())
+ * .apply(TikaIO.parseFiles());
* }</pre>
- *
- * <b>Warning:</b> the API of this IO is likely to change in the next release.
*/
@Experimental(Experimental.Kind.SOURCE_SINK)
public class TikaIO {
-
- /**
- * A {@link PTransform} that matches and parses the files
- * and returns a bounded {@link PCollection} of {@link ParseResult}.
- */
+ /** Parses files matching a given filepattern. */
public static Parse parse() {
- return new AutoValue_TikaIO_Parse.Builder()
- .build();
+ return new AutoValue_TikaIO_Parse.Builder().build();
}
- /**
- * A {@link PTransform} that accepts a {@link PCollection} of {@link ReadableFile}
- * and returns a {@link PCollection} of {@link ParseResult}.
- */
- public static ParseAll parseAll() {
- return new AutoValue_TikaIO_ParseAll.Builder()
- .build();
+ /** Parses files in a {@link PCollection} of {@link ReadableFile}. */
+ public static ParseFiles parseFiles() {
+ return new AutoValue_TikaIO_ParseFiles.Builder().build();
}
/** Implementation of {@link #parse}. */
- @SuppressWarnings("serial")
@AutoValue
public abstract static class Parse extends PTransform<PBegin, PCollection<ParseResult>> {
@Nullable
@@ -139,8 +129,7 @@ public class TikaIO {
public void populateDisplayData(DisplayData.Builder builder) {
super.populateDisplayData(builder);
- builder
- .addIfNotNull(
+ builder.addIfNotNull(
DisplayData.item("filePattern", getFilepattern()).withLabel("File Pattern"));
}
@@ -149,65 +138,68 @@ public class TikaIO {
return input
.apply(FileIO.match().filepattern(getFilepattern()))
.apply(FileIO.readMatches().withCompression(Compression.UNCOMPRESSED))
- .apply(parseAll());
+ .apply(parseFiles());
}
}
- /** Implementation of {@link #parseAll}. */
- @SuppressWarnings("serial")
+ /** Implementation of {@link #parseFiles}. */
@AutoValue
- public abstract static class ParseAll extends
- PTransform<PCollection<ReadableFile>, PCollection<ParseResult>> {
+ public abstract static class ParseFiles
+ extends PTransform<PCollection<ReadableFile>, PCollection<ParseResult>> {
+
+ @Nullable
+ abstract ValueProvider<String> getTikaConfigPath();
- @Nullable abstract ValueProvider<String> getTikaConfigPath();
- @Nullable abstract Metadata getInputMetadata();
+ @Nullable
+ abstract String getContentTypeHint();
+
+ @Nullable
+ abstract Metadata getInputMetadata();
abstract Builder toBuilder();
@AutoValue.Builder
abstract static class Builder {
abstract Builder setTikaConfigPath(ValueProvider<String> tikaConfigPath);
+
+ abstract Builder setContentTypeHint(String contentTypeHint);
+
abstract Builder setInputMetadata(Metadata metadata);
- abstract ParseAll build();
+ abstract ParseFiles build();
}
/**
- * Returns a new transform which will use the custom TikaConfig.
+ * Uses the given <a
+ * href="https://tika.apache.org/1.16/configuring.html#Using_a_Tika_Configuration_XML_file">Tika
+ * Configuration XML file</a>.
*/
- public ParseAll withTikaConfigPath(String tikaConfigPath) {
- checkNotNull(tikaConfigPath, "TikaConfigPath cannot be empty.");
+ public ParseFiles withTikaConfigPath(String tikaConfigPath) {
+ checkArgument(tikaConfigPath != null, "tikaConfigPath can not be null.");
return withTikaConfigPath(StaticValueProvider.of(tikaConfigPath));
}
- /** Same as {@code with(tikaConfigPath)}, but accepting a {@link ValueProvider}. */
- public ParseAll withTikaConfigPath(ValueProvider<String> tikaConfigPath) {
- checkNotNull(tikaConfigPath, "TikaConfigPath cannot be empty.");
- return toBuilder()
- .setTikaConfigPath(tikaConfigPath)
- .build();
+ /** Like {@code with(tikaConfigPath)}. */
+ public ParseFiles withTikaConfigPath(ValueProvider<String> tikaConfigPath) {
+ checkArgument(tikaConfigPath != null, "tikaConfigPath can not be null.");
+ return toBuilder().setTikaConfigPath(tikaConfigPath).build();
}
/**
- * Returns a new transform which will use the provided content type hint
- * to make the file parser detection more efficient.
+ * Sets a content type hint to make the file parser detection more efficient. Overrides the
+ * content type hint in {@link #withInputMetadata}, if any.
*/
- public ParseAll withContentTypeHint(String contentType) {
- checkNotNull(contentType, "ContentType cannot be empty.");
- Metadata metadata = new Metadata();
- metadata.add(Metadata.CONTENT_TYPE, contentType);
- return withInputMetadata(metadata);
+ public ParseFiles withContentTypeHint(String contentTypeHint) {
+ checkNotNull(contentTypeHint, "contentTypeHint can not be null.");
+ return toBuilder().setContentTypeHint(contentTypeHint).build();
}
- /**
- * Returns a new transform which will use the provided input metadata
- * for parsing the files.
- */
- public ParseAll withInputMetadata(Metadata metadata) {
+ /** Sets the input metadata for {@link Parser#parse}. */
+ public ParseFiles withInputMetadata(Metadata metadata) {
Metadata inputMetadata = this.getInputMetadata();
if (inputMetadata != null) {
for (String name : metadata.names()) {
- inputMetadata.set(name, metadata.get(name));
+ inputMetadata.set(name, metadata.get(name));
}
} else {
inputMetadata = metadata;
@@ -225,27 +217,25 @@ public class TikaIO {
super.populateDisplayData(builder);
if (getTikaConfigPath() != null) {
- String tikaConfigPathDisplay = getTikaConfigPath().isAccessible()
- ? getTikaConfigPath().get() : getTikaConfigPath().toString();
- builder.add(DisplayData.item("tikaConfigPath", tikaConfigPathDisplay)
- .withLabel("TikaConfig Path"));
+ builder.add(
+ DisplayData.item("tikaConfigPath", getTikaConfigPath()).withLabel("TikaConfig Path"));
}
Metadata metadata = getInputMetadata();
if (metadata != null) {
//TODO: use metadata.toString() only without a trim() once Apache Tika 1.17 gets released
- builder
- .add(DisplayData.item("inputMetadata", metadata.toString().trim())
- .withLabel("Input Metadata"));
+ builder.add(
+ DisplayData.item("inputMetadata", metadata.toString().trim())
+ .withLabel("Input Metadata"));
}
+ builder.addIfNotNull(
+ DisplayData.item("contentTypeHint", getContentTypeHint()).withLabel("Content type hint"));
}
private static class ParseToStringFn extends DoFn<ReadableFile, ParseResult> {
+ private final ParseFiles spec;
+ private transient TikaConfig tikaConfig;
- private static final long serialVersionUID = 6837207505313720989L;
- private final TikaIO.ParseAll spec;
- private TikaConfig tikaConfig;
-
- ParseToStringFn(TikaIO.ParseAll spec) {
+ ParseToStringFn(ParseFiles spec) {
this.spec = spec;
}
@@ -254,8 +244,7 @@ public class TikaIO {
if (spec.getTikaConfigPath() != null) {
ResourceId configResource =
FileSystems.matchSingleFileSpec(spec.getTikaConfigPath().get()).resourceId();
- tikaConfig = new TikaConfig(
- Channels.newInputStream(FileSystems.open(configResource)));
+ tikaConfig = new TikaConfig(Channels.newInputStream(FileSystems.open(configResource)));
}
}
@@ -264,21 +253,30 @@ public class TikaIO {
ReadableFile file = c.element();
InputStream stream = Channels.newInputStream(file.open());
try (InputStream tikaStream = TikaInputStream.get(stream)) {
+ Parser parser =
+ tikaConfig == null ? new AutoDetectParser() : new AutoDetectParser(tikaConfig);
- final Parser parser = tikaConfig == null
- ? new AutoDetectParser() : new AutoDetectParser(tikaConfig);
-
- final ParseContext context = new ParseContext();
+ ParseContext context = new ParseContext();
context.set(Parser.class, parser);
- Metadata tikaMetadata = spec.getInputMetadata() != null
- ? spec.getInputMetadata() : new org.apache.tika.metadata.Metadata();
-
+ Metadata tikaMetadata =
+ spec.getInputMetadata() != null
+ ? spec.getInputMetadata()
+ : new org.apache.tika.metadata.Metadata();
+ if (spec.getContentTypeHint() != null) {
+ tikaMetadata.set(Metadata.CONTENT_TYPE, spec.getContentTypeHint());
+ }
+
+ String location = file.getMetadata().resourceId().toString();
+ ParseResult res;
ContentHandler tikaHandler = new ToTextContentHandler();
- parser.parse(tikaStream, tikaHandler, tikaMetadata, context);
-
- c.output(new ParseResult(file.getMetadata().resourceId().toString(),
- tikaHandler.toString(),
- tikaMetadata));
+ try {
+ parser.parse(tikaStream, tikaHandler, tikaMetadata, context);
+ res = ParseResult.success(location, tikaHandler.toString(), tikaMetadata);
+ } catch (Exception e) {
+ res = ParseResult.failure(location, tikaHandler.toString(), tikaMetadata, e);
+ }
+
+ c.output(res);
}
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/ParseResultTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/ParseResultTest.java b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/ParseResultTest.java
index fd86152..95bcee0 100644
--- a/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/ParseResultTest.java
+++ b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/ParseResultTest.java
@@ -17,20 +17,60 @@
*/
package org.apache.beam.sdk.io.tika;
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
+
import org.apache.tika.metadata.Metadata;
import org.junit.Test;
-/**
- * Tests ParseResult.
- */
+/** Tests {@link ParseResult}. */
public class ParseResultTest {
@Test
public void testEqualsAndHashCode() {
- ParseResult p1 = new ParseResult("a.txt", "hello", getMetadata());
- ParseResult p2 = new ParseResult("a.txt", "hello", getMetadata());
- assertEquals(p1, p2);
- assertEquals(p1.hashCode(), p2.hashCode());
+ ParseResult successBase = ParseResult.success("a.txt", "hello", getMetadata());
+ ParseResult successSame = ParseResult.success("a.txt", "hello", getMetadata());
+ ParseResult successDifferentName = ParseResult.success("b.txt", "hello", getMetadata());
+ ParseResult successDifferentContent = ParseResult.success("a.txt", "goodbye", getMetadata());
+ ParseResult successDifferentMetadata = ParseResult.success("a.txt", "hello", new Metadata());
+
+ RuntimeException oops = new RuntimeException("oops");
+ ParseResult failureBase = ParseResult.failure("a.txt", "", new Metadata(), oops);
+ ParseResult failureSame = ParseResult.failure("a.txt", "", new Metadata(), oops);
+ ParseResult failureDifferentName = ParseResult.failure("b.txt", "", new Metadata(), oops);
+ ParseResult failureDifferentContent =
+ ParseResult.failure("b.txt", "partial", new Metadata(), oops);
+ ParseResult failureDifferentMetadata = ParseResult.failure("b.txt", "", getMetadata(), oops);
+ ParseResult failureDifferentError =
+ ParseResult.failure("a.txt", "", new Metadata(), new RuntimeException("eek"));
+
+ assertEquals(successBase, successSame);
+ assertEquals(successBase.hashCode(), successSame.hashCode());
+
+ assertThat(successDifferentName, not(equalTo(successBase)));
+ assertThat(successDifferentContent, not(equalTo(successBase)));
+ assertThat(successDifferentMetadata, not(equalTo(successBase)));
+
+ assertThat(successDifferentName.hashCode(), not(equalTo(successBase.hashCode())));
+ assertThat(successDifferentContent.hashCode(), not(equalTo(successBase.hashCode())));
+ assertThat(successDifferentMetadata.hashCode(), not(equalTo(successBase.hashCode())));
+
+ assertThat(failureBase, not(equalTo(successBase)));
+ assertThat(successBase, not(equalTo(failureBase)));
+
+ assertEquals(failureBase, failureSame);
+ assertEquals(failureBase.hashCode(), failureSame.hashCode());
+
+ assertThat(failureDifferentName, not(equalTo(failureBase)));
+ assertThat(failureDifferentError, not(equalTo(failureBase)));
+ assertThat(failureDifferentContent, not(equalTo(failureBase)));
+ assertThat(failureDifferentMetadata, not(equalTo(failureBase)));
+
+ assertThat(failureDifferentName.hashCode(), not(equalTo(failureBase.hashCode())));
+ assertThat(failureDifferentError.hashCode(), not(equalTo(failureBase.hashCode())));
+ assertThat(failureDifferentContent.hashCode(), not(equalTo(failureBase.hashCode())));
+ assertThat(failureDifferentMetadata.hashCode(), not(equalTo(failureBase.hashCode())));
}
static Metadata getMetadata() {
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/TikaIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/TikaIOTest.java b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/TikaIOTest.java
index a985b0a..1c95e9f 100644
--- a/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/TikaIOTest.java
+++ b/sdks/java/io/tika/src/test/java/org/apache/beam/sdk/io/tika/TikaIOTest.java
@@ -18,124 +18,109 @@
package org.apache.beam.sdk.io.tika;
import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;
-import static org.hamcrest.Matchers.isA;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
import java.io.IOException;
-
+import java.io.Serializable;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.List;
import org.apache.beam.sdk.io.Compression;
import org.apache.beam.sdk.io.FileIO;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.values.PCollection;
import org.apache.tika.exception.TikaException;
import org.apache.tika.metadata.Metadata;
import org.junit.Rule;
import org.junit.Test;
-import org.junit.rules.ExpectedException;
-
-/**
- * Tests TikaInput.
- */
-public class TikaIOTest {
- private static final String PDF_FILE =
- "\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n"
- + "Combining\n\nApache Beam\n\nand\n\nApache Tika\n\ncan help to ingest\n\n"
- + "the content from the files\n\nin most known formats.\n\n\n";
+/** Tests for {@link TikaIO}. */
+public class TikaIOTest implements Serializable {
private static final String PDF_ZIP_FILE =
"\n\n\n\n\n\n\n\napache-beam-tika.pdf\n\n\nCombining\n\n\nApache Beam\n\n\n"
- + "and\n\n\nApache Tika\n\n\ncan help to ingest\n\n\nthe content from the files\n\n\n"
- + "in most known formats.\n\n\n\n\n\n\n";
+ + "and\n\n\nApache Tika\n\n\ncan help to ingest\n\n\nthe content from the files\n\n\n"
+ + "in most known formats.\n\n\n\n\n\n\n";
private static final String ODT_FILE =
"\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n"
- + "Combining\nApache Beam\nand\nApache Tika\ncan help to ingest\nthe content from the"
- + " files\nin most known formats.\n";
-
- private static final String ODT_FILE2 =
- "\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n"
- + "Open Office\nPDF\nExcel\nText\nScientific\nand other formats\nare supported.\n";
+ + "Combining\nApache Beam\nand\nApache Tika\ncan help to ingest\nthe content from the"
+ + " files\nin most known formats.\n";
@Rule
- public TestPipeline p = TestPipeline.create();
- @Rule
- public ExpectedException thrown = ExpectedException.none();
-
- @Test
- public void testParsePdfFile() throws IOException {
-
- String resourcePath = getClass().getResource("/apache-beam-tika.pdf").getPath();
-
- doTestParse(resourcePath, new ParseResult(resourcePath, PDF_FILE));
- }
-
- private void doTestParse(String resourcePath, ParseResult... expectedResults)
- throws IOException {
- PCollection<ParseResult> output =
- p.apply("ParseAll", TikaIO.parse().filepattern(resourcePath))
- .apply(ParDo.of(new FilterMetadataFn()));
- PAssert.that(output).containsInAnyOrder(expectedResults);
- p.run();
- }
-
- @Test
- public void testParseAllPdfFile() throws IOException {
-
- String resourcePath = getClass().getResource("/apache-beam-tika.pdf").getPath();
-
- doTestParseAll(resourcePath, new ParseResult(resourcePath, PDF_FILE));
- }
-
- @Test
- public void testParseAllZipPdfFile() throws IOException {
-
- String resourcePath = getClass().getResource("/apache-beam-tika-pdf.zip").getPath();
-
- doTestParseAll(resourcePath, new ParseResult(resourcePath, PDF_ZIP_FILE));
- }
+ public transient TestPipeline p = TestPipeline.create();
@Test
- public void testParseAllOdtFile() throws IOException {
-
- String resourcePath = getClass().getResource("/apache-beam-tika1.odt").getPath();
-
- doTestParseAll(resourcePath, new ParseResult(resourcePath, ODT_FILE, getOdtMetadata()));
+ public void testParseAndParseFiles() throws IOException {
+ Path root =
+ Paths.get(getClass().getResource("/valid/apache-beam-tika.odt").getPath()).getParent();
+
+ List<ParseResult> expected =
+ Arrays.asList(
+ ParseResult.success(
+ root.resolve("apache-beam-tika.odt").toString(), ODT_FILE, getOdtMetadata()),
+ ParseResult.success(root.resolve("apache-beam-tika-pdf.zip").toString(), PDF_ZIP_FILE));
+
+ PCollection<ParseResult> parse =
+ p.apply("Parse", TikaIO.parse().filepattern(root.resolve("*").toString()))
+ .apply("FilterParse", ParDo.of(new FilterMetadataFn()));
+ PAssert.that(parse).containsInAnyOrder(expected);
+
+ PCollection<ParseResult> parseFiles =
+ p.apply("ParseFiles", FileIO.match().filepattern(root.resolve("*").toString()))
+ .apply(FileIO.readMatches().withCompression(Compression.UNCOMPRESSED))
+ .apply(TikaIO.parseFiles())
+ .apply("FilterParseFiles", ParDo.of(new FilterMetadataFn()));
+ PAssert.that(parseFiles).containsInAnyOrder(expected);
+ p.run();
}
- @Test
- public void testParseAllOdtFiles() throws IOException {
- String resourcePath1 = getClass().getResource("/apache-beam-tika1.odt").getPath();
- String resourcePath2 = getClass().getResource("/apache-beam-tika2.odt").getPath();
- String resourcePath = resourcePath1.replace("apache-beam-tika1", "*");
-
- doTestParseAll(resourcePath, new ParseResult(resourcePath1, ODT_FILE, getOdtMetadata()),
- new ParseResult(resourcePath2, ODT_FILE2));
+ private static Metadata getOdtMetadata() {
+ Metadata m = new Metadata();
+ m.set("Author", "BeamTikaUser");
+ return m;
}
- private void doTestParseAll(String resourcePath, ParseResult... expectedResults)
- throws IOException {
- PCollection<ParseResult> output =
- p.apply("ParseFiles", FileIO.match().filepattern(resourcePath))
- .apply(FileIO.readMatches().withCompression(Compression.UNCOMPRESSED))
- .apply(TikaIO.parseAll())
- .apply(ParDo.of(new FilterMetadataFn()));
- PAssert.that(output).containsInAnyOrder(expectedResults);
- p.run();
+ private static class FilterMetadataFn extends DoFn<ParseResult, ParseResult> {
+ @ProcessElement
+ public void processElement(ProcessContext c) {
+ ParseResult result = c.element();
+ Metadata m = new Metadata();
+ // Files contain many metadata properties. This function drops all but the "Author"
+ // property manually added to "apache-beam-tika.odt" resource only to make
+ // the tests simpler
+ if (result.getFileLocation().endsWith("valid/apache-beam-tika.odt")) {
+ m.set("Author", result.getMetadata().get("Author"));
+ }
+ ParseResult newResult = ParseResult.success(result.getFileLocation(), result.getContent(), m);
+ c.output(newResult);
+ }
}
@Test
- public void testParseAllDamagedPdfFile() throws IOException {
- thrown.expectCause(isA(TikaException.class));
- String resourcePath = getClass().getResource("/damaged.pdf").getPath();
-
- p.apply("ParseInvalidPdfFile", FileIO.match().filepattern(resourcePath))
- .apply(FileIO.readMatches())
- .apply(TikaIO.parseAll());
+ public void testParseDamagedPdfFile() throws IOException {
+ String path = getClass().getResource("/damaged.pdf").getPath();
+ PCollection<ParseResult> res = p.apply("ParseInvalidPdfFile", TikaIO.parse().filepattern(path));
+
+ PAssert.thatSingleton(res)
+ .satisfies(
+ new SerializableFunction<ParseResult, Void>() {
+ @Override
+ public Void apply(ParseResult input) {
+ assertEquals(path, input.getFileLocation());
+ assertFalse(input.isSuccess());
+ assertTrue(input.getError() instanceof TikaException);
+ return null;
+ }
+ });
p.run();
}
@@ -150,54 +135,15 @@ public class TikaIOTest {
}
@Test
- public void testParseAllDisplayData() {
- TikaIO.ParseAll parseAll = TikaIO.parseAll()
- .withTikaConfigPath("tikaconfigpath")
- .withContentTypeHint("application/pdf");
-
- DisplayData displayData = DisplayData.from(parseAll);
-
- assertThat(displayData, hasDisplayItem("tikaConfigPath", "tikaconfigpath"));
- assertThat(displayData, hasDisplayItem("inputMetadata",
- "Content-Type=application/pdf"));
- assertEquals(2, displayData.items().size());
- }
-
- @Test
- public void testParseAllDisplayDataWithCustomOptions() {
- TikaIO.ParseAll parseAll = TikaIO.parseAll()
- .withTikaConfigPath("/tikaConfigPath")
- .withContentTypeHint("application/pdf");
+ public void testParseFilesDisplayData() {
+ TikaIO.ParseFiles parseFiles =
+ TikaIO.parseFiles()
+ .withTikaConfigPath("/tikaConfigPath")
+ .withContentTypeHint("application/pdf");
- DisplayData displayData = DisplayData.from(parseAll);
+ DisplayData displayData = DisplayData.from(parseFiles);
assertThat(displayData, hasDisplayItem("tikaConfigPath", "/tikaConfigPath"));
- assertThat(displayData, hasDisplayItem("inputMetadata",
- "Content-Type=application/pdf"));
- assertEquals(2, displayData.items().size());
- }
-
- static class FilterMetadataFn extends DoFn<ParseResult, ParseResult> {
- private static final long serialVersionUID = 6338014219600516621L;
-
- @ProcessElement
- public void processElement(ProcessContext c) {
- ParseResult result = c.element();
- Metadata m = new Metadata();
- // Files contain many metadata properties. This function drops all but the "Author"
- // property manually added to "apache-beam-tika1.odt" resource only to make
- // the tests simpler
- if (result.getFileLocation().endsWith("apache-beam-tika1.odt")) {
- m.set("Author", result.getMetadata().get("Author"));
- }
- ParseResult newResult = new ParseResult(result.getFileLocation(), result.getContent(), m);
- c.output(newResult);
- }
- }
-
- static Metadata getOdtMetadata() {
- Metadata m = new Metadata();
- m.set("Author", "BeamTikaUser");
- return m;
+ assertThat(displayData, hasDisplayItem("contentTypeHint", "application/pdf"));
}
}
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/apache-beam-tika-pdf.zip
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/apache-beam-tika-pdf.zip b/sdks/java/io/tika/src/test/resources/apache-beam-tika-pdf.zip
deleted file mode 100644
index 4c0e0ef..0000000
Binary files a/sdks/java/io/tika/src/test/resources/apache-beam-tika-pdf.zip and /dev/null differ
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/apache-beam-tika.pdf
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/apache-beam-tika.pdf b/sdks/java/io/tika/src/test/resources/apache-beam-tika.pdf
deleted file mode 100644
index d3c7f14..0000000
Binary files a/sdks/java/io/tika/src/test/resources/apache-beam-tika.pdf and /dev/null differ
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/apache-beam-tika1.odt
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/apache-beam-tika1.odt b/sdks/java/io/tika/src/test/resources/apache-beam-tika1.odt
deleted file mode 100644
index 87c5577..0000000
Binary files a/sdks/java/io/tika/src/test/resources/apache-beam-tika1.odt and /dev/null differ
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/apache-beam-tika2.odt
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/apache-beam-tika2.odt b/sdks/java/io/tika/src/test/resources/apache-beam-tika2.odt
deleted file mode 100644
index a0ff320..0000000
Binary files a/sdks/java/io/tika/src/test/resources/apache-beam-tika2.odt and /dev/null differ
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika-pdf.zip
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika-pdf.zip b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika-pdf.zip
new file mode 100644
index 0000000..4c0e0ef
Binary files /dev/null and b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika-pdf.zip differ
http://git-wip-us.apache.org/repos/asf/beam/blob/c76c518b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika.odt
----------------------------------------------------------------------
diff --git a/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika.odt b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika.odt
new file mode 100644
index 0000000..87c5577
Binary files /dev/null and b/sdks/java/io/tika/src/test/resources/valid/apache-beam-tika.odt differ