You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@gobblin.apache.org by hu...@apache.org on 2018/02/13 02:20:02 UTC
incubator-gobblin git commit: [GOBBLIN-407] fix job output path for
full snapshot
Repository: incubator-gobblin
Updated Branches:
refs/heads/master f89505702 -> 0791ee992
[GOBBLIN-407] fix job output path for full snapshot
Closes #2284 from arjun4084346/fixOutputDirPath
Project: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/commit/0791ee99
Tree: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/tree/0791ee99
Diff: http://git-wip-us.apache.org/repos/asf/incubator-gobblin/diff/0791ee99
Branch: refs/heads/master
Commit: 0791ee992e533957654ee7dedb33358b312acf26
Parents: f895057
Author: Arjun <ab...@linkedin.com>
Authored: Mon Feb 12 18:20:33 2018 -0800
Committer: Hung Tran <hu...@linkedin.com>
Committed: Mon Feb 12 18:20:33 2018 -0800
----------------------------------------------------------------------
.../gobblin/configuration/WorkUnitState.java | 4 ++++
.../apache/gobblin/source/workunit/Extract.java | 2 ++
.../gobblin/source/workunit/WorkUnit.java | 21 +++++++++++++++++++-
.../org/apache/gobblin/util/WriterUtils.java | 5 ++---
4 files changed, 28 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/0791ee99/gobblin-api/src/main/java/org/apache/gobblin/configuration/WorkUnitState.java
----------------------------------------------------------------------
diff --git a/gobblin-api/src/main/java/org/apache/gobblin/configuration/WorkUnitState.java b/gobblin-api/src/main/java/org/apache/gobblin/configuration/WorkUnitState.java
index 0b40399..45bf807 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/configuration/WorkUnitState.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/configuration/WorkUnitState.java
@@ -62,6 +62,10 @@ public class WorkUnitState extends State {
private static final Gson GSON = new Gson();
+ public String getOutputFilePath() {
+ return this.workUnit.getOutputFilePath();
+ }
+
/**
* Runtime state of the {@link WorkUnit}.
*
http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/0791ee99/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/Extract.java
----------------------------------------------------------------------
diff --git a/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/Extract.java b/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/Extract.java
index 5161427..7a3735b 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/Extract.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/Extract.java
@@ -137,7 +137,9 @@ public class Extract extends State {
* Get the writer output file path corresponding to this {@link Extract}.
*
* @return writer output file path corresponding to this {@link Extract}
+ * @deprecated As {@code this.getIsFull} is deprecated.
*/
+ @Deprecated
public String getOutputFilePath() {
return this.getNamespace().replaceAll("\\.", "/") + "/" + this.getTable() + "/" + this.getExtractId() + "_"
+ (this.getIsFull() ? "full" : "append");
http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/0791ee99/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/WorkUnit.java
----------------------------------------------------------------------
diff --git a/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/WorkUnit.java b/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/WorkUnit.java
index 38aabcb..7d3f5d3 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/WorkUnit.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/source/workunit/WorkUnit.java
@@ -311,9 +311,14 @@ public class WorkUnit extends State {
@Override
public String getProp(String key) {
+ return getProp(key, null);
+ }
+
+ @Override
+ public String getProp(String key, String def) {
String value = super.getProp(key);
if (value == null) {
- value = this.extract.getProp(key);
+ value = this.extract.getProp(key, def);
}
return value;
}
@@ -359,4 +364,18 @@ public class WorkUnit extends State {
result = prime * result + ((this.extract == null) ? 0 : this.extract.hashCode());
return result;
}
+
+ public String getOutputFilePath() {
+ // Search for the properties in the workunit.
+ // This search for the property first in State and then in the Extract of this workunit.
+ String namespace = getProp(ConfigurationKeys.EXTRACT_NAMESPACE_NAME_KEY, "");
+ String table = getProp(ConfigurationKeys.EXTRACT_TABLE_NAME_KEY, "");
+ String extractId = getProp(ConfigurationKeys.EXTRACT_EXTRACT_ID_KEY, "");
+ // getPropAsBoolean and other similar methods are not overridden in WorkUnit class
+ // Thus, to enable searching in WorkUnit's Extract, we use getProp, and not getPropAsBoolean
+ boolean isFull = Boolean.parseBoolean(getProp(ConfigurationKeys.EXTRACT_IS_FULL_KEY));
+
+ return namespace.replaceAll("\\.", "/") + "/" + table + "/" + extractId + "_"
+ + (isFull ? "full" : "append");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-gobblin/blob/0791ee99/gobblin-utility/src/main/java/org/apache/gobblin/util/WriterUtils.java
----------------------------------------------------------------------
diff --git a/gobblin-utility/src/main/java/org/apache/gobblin/util/WriterUtils.java b/gobblin-utility/src/main/java/org/apache/gobblin/util/WriterUtils.java
index f5658ea..9a628d0 100644
--- a/gobblin-utility/src/main/java/org/apache/gobblin/util/WriterUtils.java
+++ b/gobblin-utility/src/main/java/org/apache/gobblin/util/WriterUtils.java
@@ -209,12 +209,11 @@ public class WriterUtils {
public static Path getDefaultWriterFilePath(State state, int numBranches, int branchId) {
if (state instanceof WorkUnitState) {
WorkUnitState workUnitState = (WorkUnitState) state;
- return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getExtract().getOutputFilePath(),
+ return new Path(ForkOperatorUtils.getPathForBranch(workUnitState, workUnitState.getOutputFilePath(),
numBranches, branchId));
-
} else if (state instanceof WorkUnit) {
WorkUnit workUnit = (WorkUnit) state;
- return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getExtract().getOutputFilePath(),
+ return new Path(ForkOperatorUtils.getPathForBranch(workUnit, workUnit.getOutputFilePath(),
numBranches, branchId));
}