Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r14682926
--- 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 --
Actually here I am just following the ```JobClient``` Interface. And it has
this static method ```runJob``` which is just a ```submitJob``` and
```waitForCompletion``` in one line. It is perhaps redundant in that case to
pass the ```JobConf``` both to the constructor and the moethod but it will get
useful once we have more methods for the JobClient.
What is really missing is a constructor with no arguments, which I should
add.
Sorry If I completely misunderstood this :P
---
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.
---
