WIP - spark program runner (dataproc)

cdapio · Mar 30, 2018 · 4ebf9b7 · 4ebf9b7
1 parent 46a1487
commit 4ebf9b7
Show file tree

Hide file tree

Showing 6 changed files with 519 additions and 25 deletions.
diff --git a/.../cdap-etl/cdap-etl-core/src/main/java/co/cask/cdap/etl/common/submit/SubmitterPlugin.java b/.../cdap-etl/cdap-etl-core/src/main/java/co/cask/cdap/etl/common/submit/SubmitterPlugin.java
@@ -61,29 +61,30 @@ public SubmitterPlugin(String stageName, Transactional transactional,
 
   @Override
   public void onFinish(final boolean succeeded) {
-    try {
-      transactional.execute(new TxRunnable() {
-        @Override
-        public void run(DatasetContext datasetContext) throws Exception {
-          T context = contextProvider.getContext(datasetContext);
+    //try {
+    //  transactional.execute(new TxRunnable() {
+    //    @Override
+    //    public void run(DatasetContext datasetContext) throws Exception {
+          T context = contextProvider.getContext(null);
           delegate.onRunFinish(succeeded, context);
-        }
-      });
-    } catch (TransactionFailureException e) {
-      LOG.warn("Error calling onRunFinish on stage {}", stageName);
-    }
+        //}
+      //});
+    //} catch (TransactionFailureException e) {
+    //  LOG.warn("Error calling onRunFinish on stage {}", stageName);
+    //}
   }
 
   @Override
-  public void prepareRun() throws TransactionFailureException {
-    transactional.execute(new TxRunnable() {
-      @Override
-      public void run(DatasetContext datasetContext) throws Exception {
-        U context = contextProvider.getContext(datasetContext);
+  public void prepareRun() throws Exception {
+
+    //transactional.execute(new TxRunnable() {
+    //  @Override
+    //  public void run(DatasetContext datasetContext) throws Exception {
+        U context = contextProvider.getContext(null);
         delegate.prepareRun(context);
         prepareAction.act(context);
-      }
-    });
+      //}
+    //});
   }
 
   /**

diff --git a/...-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/BatchSparkPipelineDriver.java b/...-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/BatchSparkPipelineDriver.java
@@ -115,7 +115,8 @@ public void run(JavaSparkExecutionContext sec) throws Exception {
     // Execution the whole pipeline in one long transaction. This is because the Spark execution
     // currently share the same contract and API as the MapReduce one.
     // The API need to expose DatasetContext, hence it needs to be executed inside a transaction
-    Transactionals.execute(sec, this, Exception.class);
+    //Transactionals.execute(sec, this, Exception.class);
+    run(datasetContext);
   }
 
   @Override

diff --git a/...tor-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/SparkBatchSinkContext.java b/...tor-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/SparkBatchSinkContext.java
@@ -53,8 +53,8 @@ public SparkBatchSinkContext(SparkBatchSinkFactory sinkFactory, JavaSparkExecuti
   @Override
   public void addOutput(Output output) {
     Output actualOutput = suffixOutput(getOutput(output));
-    Output trackableOutput = isPreviewEnabled ? actualOutput : ExternalDatasets.makeTrackable(admin, actualOutput);
-    sinkFactory.addOutput(getStageName(), trackableOutput);
+    //Output trackableOutput = isPreviewEnabled ? actualOutput : ExternalDatasets.makeTrackable(admin, actualOutput);
+    sinkFactory.addOutput(getStageName(), actualOutput);
   }
 
   @Override

diff --git a/...r-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/SparkBatchSourceContext.java b/...r-spark-core-base/src/main/java/co/cask/cdap/etl/spark/batch/SparkBatchSourceContext.java
@@ -53,7 +53,8 @@ public SparkBatchSourceContext(SparkBatchSourceFactory sourceFactory, SparkClien
 
   @Override
   public void setInput(Input input) {
-    Input trackableInput = ExternalDatasets.makeTrackable(admin, suffixInput(input));
+    //Input trackableInput = ExternalDatasets.makeTrackable(admin, suffixInput(input));
+    Input trackableInput = suffixInput(input);
     sourceFactory.addInput(getStageName(), trackableInput);
   }