Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r14735398
--- Diff:
stratosphere-addons/hadoop-compatibility/src/main/java/eu/stratosphere/hadoopcompatibility/mapred/StratosphereHadoopJobClient.java
---
@@ -0,0 +1,299 @@
+/***********************************************************************************************************************
+ * Copyright (C) 2010-2014 by the Stratosphere project
(http://stratosphere.eu)
+ *
+ * Licensed 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 eu.stratosphere.hadoopcompatibility.mapred;
+
+import eu.stratosphere.api.common.operators.Order;
+import eu.stratosphere.api.java.DataSet;
+import eu.stratosphere.api.java.ExecutionEnvironment;
+import eu.stratosphere.api.java.operators.ReduceGroupOperator;
+import eu.stratosphere.api.java.operators.SortedGrouping;
+import eu.stratosphere.api.java.operators.UnsortedGrouping;
+import
eu.stratosphere.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopGrouper;
+import
eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import
eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import eu.stratosphere.util.InstantiationUtil;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.Partitioner;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Stratosphere cluster.
+ */
+public class StratosphereHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+
+ public StratosphereHadoopJobClient(Configuration hadoopConf) {
+ this(hadoopConf,
(ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public StratosphereHadoopJobClient(Configuration hadoopConf,
ExecutionEnvironment environment) {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ this.environment.setDegreeOfParallelism(1); //TODO make
configurable.
+ }
+
+ /**
+ * Submits a Hadoop job to Stratoshere (as described by the JobConf)
and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws
IOException{
+ final StratosphereHadoopJobClient jobClient = new
StratosphereHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+
+ }
+
+ /**
+ * Submits a job to Stratosphere and returns a RunningJob instance
which can be scheduled and monitored
+ * without blocking by default. Use waitForCompletion() to block until
the job is finished.
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public RunningJob submitJob(JobConf hadoopJobConf) throws IOException{
--- End diff --
No, you got the question right. I'm still a bit unhappy with the
configuration-object situation because the configuration that the user is
passing into the constructor is never used. But lets keep it this way.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
