You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by atsikiridis <gi...@git.apache.org> on 2014/06/24 15:01:43 UTC
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
GitHub user atsikiridis opened a pull request:
https://github.com/apache/incubator-flink/pull/37
hadoopcompatibility: Implementations of basic programming interfaces and...
... a basic driver.
* wrappers for Mapper, Reducer and Combiner (as a local Reducer) on the new Java API
* wrapper for OutputCollector
* wrappers for Partitioner, values comparator.
* a driver making it posible to run unmonitored (so far) Hadoop jobs on Flink.
* tests and variations of the WordCount driver exclusively in Hadoop.
You can find more about the project here: https://issues.apache.org/jira/browse/FLINK-838?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14031511
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/atsikiridis/incubator-flink gsoc-midterm
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/incubator-flink/pull/37.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #37
----
commit 6aaccc71910074c3130228a7855d25cfcfce3e82
Author: Artem Tsikiridis <ar...@cern.ch>
Date: 2014-05-13T19:31:13Z
hadoopcompatibility: Implementations of basic programming interfaces and a basic driver.
* wrappers for Mapper, Reducer and Combiner (as a local Reducer) on the new Java API
* wrapper for OutputCollector
* wrappers for Partitioner, values comparator.
* a driver making it posible to run unmonitored (so far) Hadoop jobs on Flink.
* tests and variations of the WordCount driver exclusively in Hadoop.
----
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15549954
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceFunction.java ---
@@ -0,0 +1,199 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.functions.GroupReduceFunction;
+import org.apache.flink.api.java.operators.translation.TupleUnwrappingIterator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.WritableTypeInfo;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * The wrapper for a Hadoop Reducer (mapred API).
+ */
+public class HadoopReduceFunction<KEYIN extends WritableComparable, VALUEIN extends Writable,
+ KEYOUT extends WritableComparable, VALUEOUT extends Writable> extends GroupReduceFunction<Tuple2<KEYIN,VALUEIN>,
+ Tuple2<KEYOUT,VALUEOUT>> implements Serializable, ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Class<KEYOUT> keyoutClass;
+ private Class<VALUEOUT> valueoutClass;
+ private JobConf jobConf;
+
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> reducer;
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> combiner;
+ private HadoopOutputCollector outputCollector;
+ private Reporter reporter;
+ private ReducerTransformingIterator iterator;
+
+ @SuppressWarnings("unchecked")
+ public HadoopReduceFunction(JobConf jobConf) {
+ this.jobConf = jobConf;
+
+ this.reducer = InstantiationUtil.instantiate(jobConf.getReducerClass());
+ final Class combinerClass = jobConf.getCombinerClass();
+ if (combinerClass != null) {
+ this.combiner = (Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>) InstantiationUtil.instantiate(combinerClass);
+ }
+
+ this.keyoutClass = (Class<KEYOUT>) jobConf.getOutputKeyClass();
+ this.valueoutClass = (Class<VALUEOUT>) jobConf.getOutputValueClass();
+ this.outputCollector = new HadoopOutputCollector(keyoutClass, valueoutClass);
+ this.iterator = new ReducerTransformingIterator();
+ this.reporter = new HadoopDummyReporter();
+ }
+
+ /**
+ * A wrapping iterator for an iterator of key-value tuples that can be used as an iterator of values.
+ */
+ private final class ReducerTransformingIterator extends TupleUnwrappingIterator<VALUEIN,KEYIN>
+ implements java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private Iterator<Tuple2<KEYIN,VALUEIN>> iterator;
+ private KEYIN key;
+ private Tuple2<KEYIN,VALUEIN> first;
+
+ @Override()
+ public void set(Iterator<Tuple2<KEYIN,VALUEIN>> iterator) {
+ this.iterator = iterator;
+ if(this.hasNext()) {
+ this.first = iterator.next();
+ this.key = this.first.f0;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if(this.first != null) {
+ return true;
+ }
+ return iterator.hasNext();
+ }
+
+ @Override
+ public VALUEIN next() {
+ if(this.first != null) {
+ final VALUEIN val = this.first.f1;
+ this.first = null;
+ return val;
+ }
+ final Tuple2<KEYIN,VALUEIN> tuple = iterator.next();
+ return tuple.f1;
+ }
+
+ private KEYIN getKey() {
+ return WritableUtils.clone(this.key, jobConf);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void reduce(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYOUT,VALUEOUT>> out)
+ throws Exception {
+ outputCollector.set(out);
+ iterator.set(values);
+ reducer.reduce(iterator.getKey(), iterator, outputCollector, reporter);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void combine(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYIN,VALUEIN>> out) throws Exception {
+ if (this.combiner == null) {
--- End diff --
No, I am suggesting to throw an exception instead of calling the reduce method.
The code should not be executed anyways if combinable is not set to true.
However, if it (for any reason) happens, it is better to throw a meaningful exception than to call the Reduce function and silently continue with the program.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15512975
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
+ final Class partitionerClass = hadoopJobConf.getPartitionerClass();
+ if (! partitionerClass.equals(HashPartitioner.class)) {
+ throw new UnsupportedOperationException("Custom partitioners are not supported yet.");
+ }
+ final UnsortedGrouping<?> grouping = mapped.groupBy(0);
+
+ final ReduceGroupOperator reduceOp = grouping.reduceGroup(new HadoopReduceFunction(hadoopJobConf));
+ final Class<? extends Reducer> combinerClass = hadoopJobConf.getCombinerClass();
+ if (combinerClass != null) {
+ reduceOp.setCombinable(true);
+ }
+ reduceOp.setParallelism(reduceParallelism);
+
+ //Wrapping the output format.
+ final OutputFormat<?> outputFormat = new HadoopOutputFormat(hadoopJobConf.getOutputFormat() ,hadoopJobConf);
+ reduceOp.output(outputFormat).setParallelism(reduceParallelism);
+
+ return new DummyFlinkRunningJob(hadoopJobConf.getJobName());
+ }
+
+ @SuppressWarnings("unchecked")
+ private HadoopInputFormat<?,?> getFlinkInputFormat(JobConf jobConf) throws IOException{
+ final InputFormat inputFormat = jobConf.getInputFormat();
+ final Class inputFormatClass = inputFormat.getClass();
+
+ final TypeInformation keyTypeInfo = TypeExtractor.createTypeInfo(InputFormat.class, inputFormatClass,
+ 0, null, null);
+ final Class keyClass = keyTypeInfo.getTypeClass();
+
+ final TypeInformation valueTypeInfo = TypeExtractor.createTypeInfo(InputFormat.class, inputFormatClass,
+ 1, null, null);
+ final Class valueClass = valueTypeInfo.getTypeClass();
+
+ return new HadoopInputFormat(inputFormat, keyClass, valueClass, jobConf);
+ }
+
+ /**
+ * The number of map tasks that can be run in parallel is the number of inputSplits.
+ * The usage of JobConf.setNumMapTasks() is deprecated.
+ * The upper bound for the number of parallel map tasks is the number of task slots.
+ */
+ private int getMapParallelism(JobConf conf) throws IOException{
+ final int mapTasks = conf.getInputFormat().getSplits(conf, 0).length;
+ return mapTasks > TASK_SLOTS ? TASK_SLOTS : mapTasks;
+ }
+
+ /**
+ * The number of reduce tasks that can be run in parallel is set by JobConf.setNumReduceTasks().
+ * The upper bound for the number of parallel reduce tasks is the number of task slots.
+ */
+ private int getReduceParallelism(JobConf conf) {
+ final int reduceTasks = conf.getNumReduceTasks();
+ if (reduceTasks < TASK_SLOTS) {
--- End diff --
should be `if (reduceTasks <= TASK_SLOTS) {`?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50402587
Hi, @fhueske I hope I haven't missed anything we discussed. If I have, please tell me :)
- Your test with the configuration above passes now. I added a similar test case.
-The parallelism is now how we discussed.
- I reworked the "different combiner" case.
-I did not include custom partioning and grouping of values here. However, I would like to present something more detailed in my tracker issue.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15930089
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopMapFunction.java ---
@@ -0,0 +1,118 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.functions.FlatMapFunction;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.WritableTypeInfo;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
+/**
+ * The wrapper for a Hadoop Mapper (mapred API). Parses a Hadoop JobConf object and initialises all operations related
+ * mappers.
+ */
+public final class HadoopMapFunction<KEYIN extends WritableComparable, VALUEIN extends Writable,
+ KEYOUT extends WritableComparable, VALUEOUT extends Writable> extends FlatMapFunction<Tuple2<KEYIN,VALUEIN>,
+ Tuple2<KEYOUT,VALUEOUT>> implements Serializable, ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private transient Mapper<KEYIN,VALUEIN,KEYOUT,VALUEOUT> mapper;
+ private transient HadoopOutputCollector<KEYOUT,VALUEOUT> outputCollector;
+ private transient Reporter reporter;
+ private transient Class<KEYOUT> keyoutClass;
+ private transient Class<VALUEOUT> valueoutClass;
+
+ private JobConf jobConf;
+
+ @SuppressWarnings("unchecked")
+ public HadoopMapFunction(final JobConf jobConf) {
+ this.jobConf = jobConf;
+ this.keyoutClass = (Class<KEYOUT>) jobConf.getMapOutputKeyClass();
+ this.valueoutClass = (Class<VALUEOUT>) jobConf.getMapOutputValueClass();
+ }
+
+ /**
+ * Wrap a hadoop map() function call and use a Flink collector to collect the result values.
+ * @param value The input value.
+ * @param out The collector for emitting result values.
+ *
+ * @throws Exception
+ */
+ @Override
+ public void flatMap(final Tuple2<KEYIN,VALUEIN> value, final Collector<Tuple2<KEYOUT,VALUEOUT>> out)
+ throws Exception {
+ outputCollector.set(out);
+ mapper.map(value.f0, value.f1, outputCollector, reporter);
+ }
+
+ @Override
+ public TypeInformation<Tuple2<KEYOUT,VALUEOUT>> getProducedType() {
+ final WritableTypeInfo<KEYOUT> keyTypeInfo = new WritableTypeInfo<KEYOUT>(keyoutClass);
+ final WritableTypeInfo<VALUEOUT> valueTypleInfo = new WritableTypeInfo<VALUEOUT>(valueoutClass);
+ return new TupleTypeInfo<Tuple2<KEYOUT,VALUEOUT>>(keyTypeInfo, valueTypleInfo);
+ }
+
+ /**
+ * Custom serialization methods.
+ * @see http://docs.oracle.com/javase/7/docs/api/java/io/Serializable.html
+ */
+ private void writeObject(final ObjectOutputStream out) throws IOException {
+ HadoopConfiguration.writeHadoopJobConf(jobConf,out);
--- End diff --
This ensures that each object gets a seperate ```JobConf``` object. This is related to https://mail-archives.apache.org/mod_mbox/incubator-flink-dev/201407.mbox/browser
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15342004
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
+ final Partitioner partitioner = InstantiationUtil.instantiate(hadoopJobConf.getPartitionerClass());
+ final int noOfReduceTasks = hadoopJobConf.getNumReduceTasks();
+ final UnsortedGrouping partitions = mapped.groupBy(new HadoopPartitioner(partitioner, noOfReduceTasks));
+
+ final ReduceGroupOperator identity = partitions.reduceGroup(new HadoopIdentityReduce()); //In order to regroup.
+
+ //Grouping
+ final RawComparator comparator = hadoopJobConf.getOutputValueGroupingComparator();
+ final UnsortedGrouping grouping = identity.groupBy(new HadoopGroupingKeySelector(comparator, mapOutputKeyClass));
+
+ //Sorting. TODO Custom sorting should be implemented. Ascending by default.
+
+ //Is a combiner specified in the jobConf?
+ final Class<? extends Reducer> combinerClass = hadoopJobConf.getCombinerClass();
+
+ //Is a Reducer specified ? No reducer means identity reducer.
+ final Class<? extends Reducer> reducerClass = hadoopJobConf.getReducerClass();
+ final Reducer reducer = InstantiationUtil.instantiate(reducerClass);
+
+ //The output types of the reducers.
+ final Class outputKeyClass = hadoopJobConf.getOutputKeyClass();
+ final Class outputValueClass = hadoopJobConf.getOutputValueClass();
+
+ final ReduceGroupOperator reduceOp;
+ if (combinerClass != null && combinerClass.equals(reducerClass)) {
+ reduceOp = grouping.reduceGroup(new HadoopReduceFunction(reducer, mapOutputKeyClass,
+ mapOutputValueClass));
+ reduceOp.setCombinable(true); //The combiner is the same class as the reducer.
+ }
+ else if(combinerClass != null) { //We have a different combiner.
--- End diff --
The case of different combiner and reducer needs to be handled differently.
You do `mapOut.groupBy(hadoopGrouper).reduceGroup(combiner).reduceGroup(reducer)`.
In this case, the combiner is called as it was a reducer, i.e., after the data was partitioned and sorted. The purpose of a combiner is to reduce the data before shipping it over the network and (completely) sorting it.
Moreover, the reduce is finally called on the fully reduced data as an AllReduce, which puts all data into a single group which will cause wrong results.
The solution here is to have a HadoopReduceFunction that can deal with a Reduce and Combine function.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48535683
Yes, move them to a `test` package.
Regarding the DOP: The `./bin/stratosphere` client allows to pass a `-p` parameter that sets the DOP for the enviroment. This way, users can change the DOP very easily.
So by default, I would not set any DOP so that you can set it when running.
If the user has set the respective hadoop property (`mapred.tasktracker.map.tasks.maximum`) we indeed should respect that. Flink allows to set the DOP individually to operators, the functionality is just not exposed via the Java API.
Can you open a JIRA for that, discuss how to implement it and then contribute the change via a PR? (I will merge this change independently of the change, so it won't be a blocker).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48532914
One more question: Why do you force the application to use a DOP of one?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15544868
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceFunction.java ---
@@ -0,0 +1,199 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.functions.GroupReduceFunction;
+import org.apache.flink.api.java.operators.translation.TupleUnwrappingIterator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.WritableTypeInfo;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * The wrapper for a Hadoop Reducer (mapred API).
+ */
+public class HadoopReduceFunction<KEYIN extends WritableComparable, VALUEIN extends Writable,
+ KEYOUT extends WritableComparable, VALUEOUT extends Writable> extends GroupReduceFunction<Tuple2<KEYIN,VALUEIN>,
+ Tuple2<KEYOUT,VALUEOUT>> implements Serializable, ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Class<KEYOUT> keyoutClass;
+ private Class<VALUEOUT> valueoutClass;
+ private JobConf jobConf;
+
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> reducer;
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> combiner;
+ private HadoopOutputCollector outputCollector;
+ private Reporter reporter;
+ private ReducerTransformingIterator iterator;
+
+ @SuppressWarnings("unchecked")
+ public HadoopReduceFunction(JobConf jobConf) {
+ this.jobConf = jobConf;
+
+ this.reducer = InstantiationUtil.instantiate(jobConf.getReducerClass());
+ final Class combinerClass = jobConf.getCombinerClass();
+ if (combinerClass != null) {
+ this.combiner = (Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>) InstantiationUtil.instantiate(combinerClass);
+ }
+
+ this.keyoutClass = (Class<KEYOUT>) jobConf.getOutputKeyClass();
+ this.valueoutClass = (Class<VALUEOUT>) jobConf.getOutputValueClass();
+ this.outputCollector = new HadoopOutputCollector(keyoutClass, valueoutClass);
+ this.iterator = new ReducerTransformingIterator();
+ this.reporter = new HadoopDummyReporter();
+ }
+
+ /**
+ * A wrapping iterator for an iterator of key-value tuples that can be used as an iterator of values.
+ */
+ private final class ReducerTransformingIterator extends TupleUnwrappingIterator<VALUEIN,KEYIN>
+ implements java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private Iterator<Tuple2<KEYIN,VALUEIN>> iterator;
+ private KEYIN key;
+ private Tuple2<KEYIN,VALUEIN> first;
+
+ @Override()
+ public void set(Iterator<Tuple2<KEYIN,VALUEIN>> iterator) {
+ this.iterator = iterator;
+ if(this.hasNext()) {
+ this.first = iterator.next();
+ this.key = this.first.f0;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if(this.first != null) {
+ return true;
+ }
+ return iterator.hasNext();
+ }
+
+ @Override
+ public VALUEIN next() {
+ if(this.first != null) {
+ final VALUEIN val = this.first.f1;
+ this.first = null;
+ return val;
+ }
+ final Tuple2<KEYIN,VALUEIN> tuple = iterator.next();
+ return tuple.f1;
+ }
+
+ private KEYIN getKey() {
+ return WritableUtils.clone(this.key, jobConf);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void reduce(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYOUT,VALUEOUT>> out)
+ throws Exception {
+ outputCollector.set(out);
+ iterator.set(values);
+ reducer.reduce(iterator.getKey(), iterator, outputCollector, reporter);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void combine(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYIN,VALUEIN>> out) throws Exception {
+ if (this.combiner == null) {
--- End diff --
The default value for ```JobConf.getCombinerClass()``` is null ( and this is not the case with Mappers and Reducers that have their respective identity functions as default. Even if the user wants the combiner to be the same as the reducer he has to explicitly say so. Otherwise, the job runs with no combine phase. Do you suggest that the combiner class should be the reducer class by default and that we should never skip reducing?
Moreover, in the client there is:
```java
final Class<? extends Reducer> combinerClass = hadoopJobConf.getCombinerClass();
if (combinerClass != null) {
reduceOp.setCombinable(true);
}
```
so if there is no combiner the function is not combinable for Flink. Isn't that correct?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis closed the pull request at:
https://github.com/apache/incubator-flink/pull/37
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50000110
I'd also update the `HadoopDummyReporter` and throw a `UnsupportedOperationException` whenever one of its methods is called.
Otherwise users might rely on the returned values.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15390287
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
--- End diff --
The issue is that Flink uses the same key for partitioning and grouping, i.e., it would also partition on `keyOfGroup` and group on `partition`. Of course, partitioning and grouping are separate operations in the execution layer, but the API gives the same key for both to the optimizer.
If neither the partitioner or sorter are custom, MapReduce will also use the same key for both operations, i.e., behave the same way as Flink.
For this PR, I'd simply throw an exception if you encounter a user-defined partitioner or comparator and support only the default case. We need to think carefully how we can add support for custom partitioners and comparators and might need to extend the API or add a hook somewhere to enable that.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15549615
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
--- End diff --
Ah, OK!
Thanks for the explanation, let's do it like that :-)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15514228
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceFunction.java ---
@@ -0,0 +1,199 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.functions.GroupReduceFunction;
+import org.apache.flink.api.java.operators.translation.TupleUnwrappingIterator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.WritableTypeInfo;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * The wrapper for a Hadoop Reducer (mapred API).
+ */
+public class HadoopReduceFunction<KEYIN extends WritableComparable, VALUEIN extends Writable,
+ KEYOUT extends WritableComparable, VALUEOUT extends Writable> extends GroupReduceFunction<Tuple2<KEYIN,VALUEIN>,
+ Tuple2<KEYOUT,VALUEOUT>> implements Serializable, ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Class<KEYOUT> keyoutClass;
+ private Class<VALUEOUT> valueoutClass;
+ private JobConf jobConf;
+
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> reducer;
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> combiner;
--- End diff --
A combiner may not change the type of its input key and value.
It should be `private Reducer<KEYIN,VALUEIN,KEYIN,VALUEIN> combiner;`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15741355
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
With a warning perhaps?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15863231
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,473 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.functions.GroupReduceFunction;
+import org.apache.flink.api.java.functions.InvalidTypesException;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+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.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+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 org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public final class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
Yeah, let's go for now with your suggestion, i.e., use max dop as number of slot if the parameter has not been initialized before. Please also add the warning.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15515319
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
--- End diff --
Hadoop supports MapOnly jobs.
Please check if a Reduce class is set in the `JobConf`. If not, finish the job after the Mapper by adding a DataSink with DOP = mapDOP.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by nirvanesque <gi...@git.apache.org>.
Github user nirvanesque commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-53571986
Hello Artem and mentors,
First of all nice greetings from INRIA, France.
Hope you had an enjoyable experience in GSOC!
Thanks to Robert (rmetzger) for forwarding me here ...
At INRIA, we are starting to adopt Stratosphere / Flink.
The top-level goal is to enhance performance in User Defined Functions (UDFs) with long workflows using multiple M-R, by using the larger set of Second Order Functions (SOFs) in Stratosphere / Flink.
We will demonstrate this improvement by implementing some Use Cases for business purposes.
For this purpose, we have chosen some customer analysis Use Cases using weblogs and related data, for 2 companies (who appeared interested to try using Stratosphere / Flink )
- a mobile phone app developer: http://www.tribeflame.com
- an anti-virus & Internet security software company: www.f-secure.com
I will be happy to share with you these Use Cases, if you are interested. Just ask me here.
At present, we are typically in the profiles of Alice-Bob-Sam, as described in [Artem's GSoC proposal](https://github.com/stratosphere/stratosphere/wiki/GSoC-2014-Project-Proposal-Draft-by-Artem-Tsikiridis).
Hadoop seems to be the starting square for our Stratosphere / Flink journey.
Same is the situation with developers in the above 2 companies :-)
Briefly,
We have installed and run some example programmes from Flink / Stratosphere (versions 0.5.2 and 0.6). We use a cluster (the grid5000 for our Hadoop & Stratosphere installations)
We have some good understanding of Hadoop and its use in Streaming and Pipes in conjunction with scripting languages (Python & R specifically)
In the first phase, we would like to run some "Hadoop-like" jobs (mainly multiple M-R workflows) on Stratosphere, preferably with extensive Java or Scala programming.
I refer to your [GSoC project map](https://github.com/stratosphere/stratosphere/wiki/%5BGSoC-14%5D-A-Hadoop-abstraction-layer-for-Stratosphere-%28Project-Map-and-Notes%29) which seems very interesting.
If we could have a Hadoop abstraction as you have mentioned, that would be ideal for our first phase.
In later phases, when we implement complex join and group operations, we would dive deeper into Stratosphere / Flink Java or Scala APIs
Hence, I would like to know, what is the current status in this direction?
What has been implemented already? In which version onwards? How to try them?
What is yet to be implemented? When - which versions?
You may also like to see [my discussion with Robert on this page](http://flink.incubator.apache.org/docs/0.6-incubating/cli.html#comment-1558297261).
I am still mining into different discussions - here as well as on JIRA.
Please do refer me to the relevant links, JIRA tickets, etc if that saves your time in re-typing large replies.
It will help us to catch up fast with the train of collective thinking in the Stratosphere / Flink roadmap, and eventually contribute to the project.
Thanks in advance,
Anirvan
PS : Apologies for using names / rechristened names (e.g. Flink / Stratosphere) as I am not sure, which name to use currently.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-54731181
@atsikiridis I think we can close this PR for now.
The support to run complete Hadoop Jobs requires a bit more work. At least the combiner should work.
I am waiting for comments on PR #108 which is required to add custom combiners to the Hadoop Job operation.
Parts of this PR (wrappers for iterators and collectors, dummy reporters, etc.) can be added in a new PR which addresses FLINK-1076.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15341292
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
--- End diff --
The code contains many rawtype warnings.
Please try to get rid of them (e.g., `DataSet<?>` instead of `DataSet`) and add a `@SuppressWarnings("rawtypes")` to the method if that's not possible.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50551442
Ahh @fhueske squashing was a complete brain stop sorry :(
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-47141587
Hey @rmetzger I think I made a fix for hadoop 2.2.0 dependencies. But I'm not sure I understand why it is still failing. Can you take a look please? :)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-49999194
I checked whether a MapFunction can be configured using a custom key in the JobConf.
The following code does not work correctly (for multiple reasons):
```{java}
public class MyMapper implements Mapper<LongWritable, Text, IntWritable, IntWritable> {
int x = 0;
@Override
public void configure(JobConf arg0) {
x = arg0.getInt("myX", 100);
System.out.println("x="+x);
}
@Override
public void map(LongWritable arg0, Text arg1, OutputCollector<IntWritable, IntWritable> arg2, Reporter arg3)
throws IOException {
System.out.println("map x="+x);
}
}
public class MyStupidHadoopJob {
public static void main(String[] args) throws Exception {
final JobConf conf = new JobConf();
conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(...));
conf.setOutputFormat(TextOutputFormat.class);
TextOutputFormat.setOutputPath(conf, new Path(...));
conf.setMapperClass(MyMapper.class);
conf.setMapOutputKeyClass(IntWritable.class);
conf.setMapOutputValueClass(IntWritable.class);
conf.setInt("myX", 50);
conf.set("mapred.textoutputformat.separator", " ");
conf.setOutputKeyClass(Text.class);
conf.setOutputValueClass(LongWritable.class);
FlinkHadoopJobClient.runJob(conf);
}
}
```
I found the following:
- The `configure()` method is called (twice for DOP=1), i.e., the `println()` is executed.
- The JobConf object passed to `configure()` seems to be incorrect. `x` is assigned 100, i.e., the key `"myX"` is not set in the passed configuration.
- When executing the mapper, the `println()` statements print "map x=0", i.e., `x` is not initialized in MapFunction instances which are called.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-63718807
I backuped the code.
Most of it went already in as part of the Hadoop function compatibility. I will continue with the full Hadoop job integration sometime later if nobody else picks it up.
I'm OK with closing this PR.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-51531949
Looks good to me.
Note, I only browsed the code and did not run it.
I'll add a comment to JIRA as well. I have a working prototype for full support of custom partitioner, sorter, and groupers (combine is not supported yet, though). However, the solution goes through the full program compilation stack and I'd like to get some feedback before I continue...
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-63622016
@fhueske What's the state of this PR? Is it supercedes by other changes, which happened in the mean time?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50000152
Ah cool :-)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48538801
Well, it's an undocumented feature and easy to miss... ;-)
Will open a JIRA to fix that.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48525426
I was just executing the main method of this class: https://github.com/rmetzger/testjob/blob/artem/src/main/java/eu/stratosphere/test/hadoop/Join.java, with these arguments `-inFormat org.apache.hadoop.mapred.TextInputFormat -joinOp outer /tmp/jointest/input /tmp/jointest/input1 /tmp/jointest/output`
I thought this is independent of the InputFormat since the error is happening when the output collector is checking the type of the elements it received?
I'll check it again.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48881999
Let me know when I can have a look at the PR again.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15545161
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
--- End diff --
Actually, the ```IdentityReducer``` is the default reducer value by the conf for a Hadoop Job. I think we should also run it by default, as the user expects sorted data with that job.
In order to go map-only ( and no sorting) the user must set ```JobConf.setNumReduceTasks(0)```. And you are right, this must be supported properly.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48538483
Oh yes, @fhueske. I completely missed it for some reason. So then, it gets trivial and I will add correct parallelism for this PR and see if the tests pass.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48536073
Ok, I'll do that. Thank you :)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15377471
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
+ final Partitioner partitioner = InstantiationUtil.instantiate(hadoopJobConf.getPartitionerClass());
+ final int noOfReduceTasks = hadoopJobConf.getNumReduceTasks();
+ final UnsortedGrouping partitions = mapped.groupBy(new HadoopPartitioner(partitioner, noOfReduceTasks));
+
+ final ReduceGroupOperator identity = partitions.reduceGroup(new HadoopIdentityReduce()); //In order to regroup.
+
+ //Grouping
+ final RawComparator comparator = hadoopJobConf.getOutputValueGroupingComparator();
+ final UnsortedGrouping grouping = identity.groupBy(new HadoopGroupingKeySelector(comparator, mapOutputKeyClass));
+
+ //Sorting. TODO Custom sorting should be implemented. Ascending by default.
+
+ //Is a combiner specified in the jobConf?
+ final Class<? extends Reducer> combinerClass = hadoopJobConf.getCombinerClass();
+
+ //Is a Reducer specified ? No reducer means identity reducer.
+ final Class<? extends Reducer> reducerClass = hadoopJobConf.getReducerClass();
+ final Reducer reducer = InstantiationUtil.instantiate(reducerClass);
+
+ //The output types of the reducers.
+ final Class outputKeyClass = hadoopJobConf.getOutputKeyClass();
+ final Class outputValueClass = hadoopJobConf.getOutputValueClass();
+
+ final ReduceGroupOperator reduceOp;
+ if (combinerClass != null && combinerClass.equals(reducerClass)) {
+ reduceOp = grouping.reduceGroup(new HadoopReduceFunction(reducer, mapOutputKeyClass,
+ mapOutputValueClass));
+ reduceOp.setCombinable(true); //The combiner is the same class as the reducer.
+ }
+ else if(combinerClass != null) { //We have a different combiner.
--- End diff --
Ok, I understand.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15741334
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
So would it be safe to assume that if the slots have not been set ( = -1) this is obviously an IDE run and slots should be equal to the max DOP ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-47066734
Thank you for the pull request. I know it was a lot of work.
It fails to build with hadoop 2.2.0 dependencies: https://travis-ci.org/apache/incubator-flink/builds/28311858
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-63813451
That would be really cool! Let me know if you need any help! :-)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48534266
Definitely, this should not stay 1. I just left it here for simplicity.
In my new branch it is the the maximum of ( ```mapred.tasktracker.map.tasks.maximum``` and ```mapred.tasktracker.reduce.tasks.maximum``` ) which are the "slots" of tasks per node. What do you think?
The problem here is that in flink it is a degree for all operators in an environment (if I'm not mistaken) so if for example, DOP = 3 it will be 3 mappers and 3 reducers. What happens if I want 3 mappers in parallel and 1 reducer ? Is there actually a way to set parallelism per operator with the current APIs ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by twalthr <gi...@git.apache.org>.
Github user twalthr commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-47162498
I don't think that I can review and test it this week. But next week would be possible.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48971633
Moreover, if I rebase I can handle the ```InputFormats``` more elegantly for the client after https://github.com/apache/incubator-flink/pull/52#event-139635478
Of course I can do seperate smaller PRs in the coming weeks.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-53575301
Hi Anirvan,
thanks a lot for getting in touch.
Robert forwarded your mail to the flink-dev mailing list and Artem and I replied there ([see Archive](http://mail-archives.apache.org/mod_mbox/flink-dev/201408.mbox/%3CCAAiGx18ZSW0Lfsapg_KPwmEO%2B-1TZXT0q_1OAbGRzT5At-b%3Dkg%40mail.gmail.com%3E))
I would suggest continue the discussion on the mailing list.
You can sign up [here](http://flink.incubator.apache.org/community.html#mailing-lists).
Best, Fabian
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48535007
It's great that you think it is ok. I will check which operations and have not been implemented on the branch and guard them as you said. By the way, I wanted to ask you. By the way, do you think that ```WordCountVariations``` file makes sense where it is? Maybe I should move it to ```test``` ? Because these are not really examples, they are just used for testing I think that maybe I have it in the wrong place...
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r14649247
--- 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 --
Why do you expect the `JobConf` again here? I think the user already passed it when creating the `StratosphereHadoopJobClient`. (`JobConf` inherits from `Configuration`).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15513139
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
--- End diff --
Add complete JavaDocs to all public methods including parameter, return, and throws descriptions.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15512850
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
When I run a job from Eclipse using a LocalExecutionEnvironment, `TASK_SLOTS` is set to -1 and execution fails.
This might be a problem with the LocalExecutionEnvironment, but should be resolved.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48284632
I'm testing the PR now.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48382325
I experienced another error
```
Exception in thread "main" java.io.IOException: An error has occurred.
at eu.stratosphere.hadoopcompatibility.mapred.StratosphereHadoopJobClient$DummyStratosphereRunningJob.waitForCompletion(StratosphereHadoopJobClient.java:238)
at eu.stratosphere.test.hadoop.Join.run(Join.java:162)
at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
at eu.stratosphere.test.hadoop.Join.main(Join.java:172)
Caused by: eu.stratosphere.nephele.client.JobExecutionException: java.io.IOException: Type mismatch in key: expected class org.apache.hadoop.io.BytesWritable, received org.apache.hadoop.io.LongWritable
at eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector.validateExpectedTypes(HadoopOutputCollector.java:78)
at eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector.collect(HadoopOutputCollector.java:48)
at eu.stratosphere.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector.collect(HadoopOutputCollector.java:1)
at org.apache.hadoop.mapred.lib.IdentityMapper.map(IdentityMapper.java:38)
at eu.stratosphere.hadoopcompatibility.mapred.HadoopMapFunction.flatMap(HadoopMapFunction.java:80)
at eu.stratosphere.hadoopcompatibility.mapred.HadoopMapFunction.flatMap(HadoopMapFunction.java:1)
at eu.stratosphere.pact.runtime.task.chaining.ChainedFlatMapDriver.collect(ChainedFlatMapDriver.java:71)
at eu.stratosphere.pact.runtime.task.DataSourceTask.invoke(DataSourceTask.java:215)
at eu.stratosphere.nephele.execution.RuntimeEnvironment.run(RuntimeEnvironment.java:260)
at java.lang.Thread.run(Thread.java:745)
at eu.stratosphere.nephele.client.JobClient.submitJobAndWait(JobClient.java:354)
at eu.stratosphere.client.LocalExecutor.executePlan(LocalExecutor.java:231)
at eu.stratosphere.api.java.LocalEnvironment.execute(LocalEnvironment.java:55)
at eu.stratosphere.hadoopcompatibility.mapred.StratosphereHadoopJobClient$DummyStrat
```
I'm not sure what I'm doing wrong.
You can find the code that I'm using for testing here: https://github.com/rmetzger/testjob/tree/artem
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-51454124
@rmetzger @fhueske what do you think now, guys ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48537249
In fact, setting the DOP of individual operators is supported with ``Operator.setParallelism(int dop)``.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-54790061
@fhueske Ok, I will try to have the pr for FLINK-1076 soon.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15514510
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/HadoopReduceFunction.java ---
@@ -0,0 +1,199 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.functions.GroupReduceFunction;
+import org.apache.flink.api.java.operators.translation.TupleUnwrappingIterator;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.ResultTypeQueryable;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.java.typeutils.WritableTypeInfo;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopConfiguration;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopDummyReporter;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopOutputCollector;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.util.Collector;
+import org.apache.flink.util.InstantiationUtil;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.Reporter;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.Iterator;
+
+/**
+ * The wrapper for a Hadoop Reducer (mapred API).
+ */
+public class HadoopReduceFunction<KEYIN extends WritableComparable, VALUEIN extends Writable,
+ KEYOUT extends WritableComparable, VALUEOUT extends Writable> extends GroupReduceFunction<Tuple2<KEYIN,VALUEIN>,
+ Tuple2<KEYOUT,VALUEOUT>> implements Serializable, ResultTypeQueryable<Tuple2<KEYOUT,VALUEOUT>> {
+
+ private static final long serialVersionUID = 1L;
+
+ private Class<KEYOUT> keyoutClass;
+ private Class<VALUEOUT> valueoutClass;
+ private JobConf jobConf;
+
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> reducer;
+ private Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> combiner;
+ private HadoopOutputCollector outputCollector;
+ private Reporter reporter;
+ private ReducerTransformingIterator iterator;
+
+ @SuppressWarnings("unchecked")
+ public HadoopReduceFunction(JobConf jobConf) {
+ this.jobConf = jobConf;
+
+ this.reducer = InstantiationUtil.instantiate(jobConf.getReducerClass());
+ final Class combinerClass = jobConf.getCombinerClass();
+ if (combinerClass != null) {
+ this.combiner = (Reducer<KEYIN, VALUEIN, KEYOUT, VALUEOUT>) InstantiationUtil.instantiate(combinerClass);
+ }
+
+ this.keyoutClass = (Class<KEYOUT>) jobConf.getOutputKeyClass();
+ this.valueoutClass = (Class<VALUEOUT>) jobConf.getOutputValueClass();
+ this.outputCollector = new HadoopOutputCollector(keyoutClass, valueoutClass);
+ this.iterator = new ReducerTransformingIterator();
+ this.reporter = new HadoopDummyReporter();
+ }
+
+ /**
+ * A wrapping iterator for an iterator of key-value tuples that can be used as an iterator of values.
+ */
+ private final class ReducerTransformingIterator extends TupleUnwrappingIterator<VALUEIN,KEYIN>
+ implements java.io.Serializable {
+
+ private static final long serialVersionUID = 1L;
+ private Iterator<Tuple2<KEYIN,VALUEIN>> iterator;
+ private KEYIN key;
+ private Tuple2<KEYIN,VALUEIN> first;
+
+ @Override()
+ public void set(Iterator<Tuple2<KEYIN,VALUEIN>> iterator) {
+ this.iterator = iterator;
+ if(this.hasNext()) {
+ this.first = iterator.next();
+ this.key = this.first.f0;
+ }
+ }
+
+ @Override
+ public boolean hasNext() {
+ if(this.first != null) {
+ return true;
+ }
+ return iterator.hasNext();
+ }
+
+ @Override
+ public VALUEIN next() {
+ if(this.first != null) {
+ final VALUEIN val = this.first.f1;
+ this.first = null;
+ return val;
+ }
+ final Tuple2<KEYIN,VALUEIN> tuple = iterator.next();
+ return tuple.f1;
+ }
+
+ private KEYIN getKey() {
+ return WritableUtils.clone(this.key, jobConf);
+ }
+
+ @Override
+ public void remove() {
+ throw new UnsupportedOperationException();
+ }
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void reduce(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYOUT,VALUEOUT>> out)
+ throws Exception {
+ outputCollector.set(out);
+ iterator.set(values);
+ reducer.reduce(iterator.getKey(), iterator, outputCollector, reporter);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void combine(Iterator<Tuple2<KEYIN,VALUEIN>> values, Collector<Tuple2<KEYIN,VALUEIN>> out) throws Exception {
+ if (this.combiner == null) {
--- End diff --
The constructor initializes `this.combiner` to the combine class registered in the `JobConf`. If the combiner is the same class as the reducer it is still registered in the `JobConf`, right?
In that case, the combine method may not be called if no combiner is registered, i.e., `this.combiner == null`. Instead, this is a invalid system behavior and an exception should be raised.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48403826
Thank you for testing it.
I believe it is not a bug. I actually tried to simulate this behavior with my last commit. If you run this on Hadoop, the same message is shown.
Can you mention how do you run the driver please?
I get the same when I set ```TextInputFormat``` and ```TextOutputFormat```, And all defaults.
So it really depends on the InputFormat and OutputFormat you choose.
This happens because we have mismatches between the types built in the input / output formats and the the input output format keys specified in the driver.
If you mention your case exactly (if it is a sequence file how is it generated?) and maybe we can check (or it is really a bug).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15515585
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
--- End diff --
I saw MapOnly jobs are currently supported by injecting an IdentityReducer. This is a very expensive way to do it, because the data is shuffled over the network and sorted. MapOnly jobs aim to avoid that.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15512807
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
+ final Class partitionerClass = hadoopJobConf.getPartitionerClass();
+ if (! partitionerClass.equals(HashPartitioner.class)) {
+ throw new UnsupportedOperationException("Custom partitioners are not supported yet.");
+ }
+ final UnsortedGrouping<?> grouping = mapped.groupBy(0);
+
+ final ReduceGroupOperator reduceOp = grouping.reduceGroup(new HadoopReduceFunction(hadoopJobConf));
+ final Class<? extends Reducer> combinerClass = hadoopJobConf.getCombinerClass();
+ if (combinerClass != null) {
+ reduceOp.setCombinable(true);
+ }
+ reduceOp.setParallelism(reduceParallelism);
+
+ //Wrapping the output format.
+ final OutputFormat<?> outputFormat = new HadoopOutputFormat(hadoopJobConf.getOutputFormat() ,hadoopJobConf);
+ reduceOp.output(outputFormat).setParallelism(reduceParallelism);
+
+ return new DummyFlinkRunningJob(hadoopJobConf.getJobName());
+ }
+
+ @SuppressWarnings("unchecked")
+ private HadoopInputFormat<?,?> getFlinkInputFormat(JobConf jobConf) throws IOException{
+ final InputFormat inputFormat = jobConf.getInputFormat();
+ final Class inputFormatClass = inputFormat.getClass();
+
+ final TypeInformation keyTypeInfo = TypeExtractor.createTypeInfo(InputFormat.class, inputFormatClass,
+ 0, null, null);
+ final Class keyClass = keyTypeInfo.getTypeClass();
+
+ final TypeInformation valueTypeInfo = TypeExtractor.createTypeInfo(InputFormat.class, inputFormatClass,
+ 1, null, null);
+ final Class valueClass = valueTypeInfo.getTypeClass();
+
+ return new HadoopInputFormat(inputFormat, keyClass, valueClass, jobConf);
+ }
+
+ /**
+ * The number of map tasks that can be run in parallel is the number of inputSplits.
+ * The usage of JobConf.setNumMapTasks() is deprecated.
+ * The upper bound for the number of parallel map tasks is the number of task slots.
+ */
+ private int getMapParallelism(JobConf conf) throws IOException{
--- End diff --
the number of map tasks should be the minimum of
- user specified value `conf.getNumMapTasks()`
- the number of splits `conf.getInputFormat().getSplits(conf, 0).length`
- the number of slots `TASK_SLOTS`
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15341686
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
--- End diff --
The partitioning / grouping code does not look right to me.
It does `mapOut.groupBy(partitioner).reduceGroup(identity).groupBy(groupingKeySelector)`
`groupBy()` does both, partitioning and grouping, in the Flink API. The partitioning of the first `groupBy()` is destroyed by the second `groupBy()`. So the only thing that will happen is that the data is shuffled and sorted twice which is very expensive. We need to find a clever way to combine the HadoopPartitioner and the HadoopGroupingKeySelector into a single `groupBy()` operation or extend the Flink API to be able to specify both independently.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-49002401
It would be much easier for us, if you could
a) update the license headers to the new ASF headers
b) rename the code from stratosphere to flink (also the packages and files and directories)
c) rebase the code to the latest master
It would also be cool if you could improve the InputFormats.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48974556
You might also be interested in my weekly report:
https://issues.apache.org/jira/browse/FLINK-838
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-49561327
This PR is now ASF compliant. The parallelism of the tasks is not fully mapped though. Hopefully I can solve this properly soon... ( @fhueske has given me some input in the tracker issue https://issues.apache.org/jira/browse/FLINK-838)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15512920
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
+ private static final Log LOG = LogFactory.getLog(FlinkHadoopJobClient.class);
+
+
+ private ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ public void init(JobConf conf) throws IOException {
+ this.hadoopConf = conf;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ final int mapParallelism = getMapParallelism(hadoopJobConf);
+ final int reduceParallelism = getReduceParallelism(hadoopJobConf);
+
+ //setting up the inputFormat for the job
+ final DataSource<?> input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+ input.setParallelism(mapParallelism);
+
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(hadoopJobConf));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
--- End diff --
Set it to `mapParallelism` to avoid the recomputation of the input splits.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15558369
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
I investigated why this is happening.
When we run tests, our tests run in a ```ContextEnvironment``` (e.g. a test environment) and for this environment the values in ```GlobalConfiguration``` have been already set.
However, in the case when we just run the user's job, since there is no test environment the local or remote environment have not been created yet.
This makes it impossible to use the task's maximum task slots since no environment has been started.
The solution would be to do perform this check for the limit of maximum tasks in parallel after this value has been initialized by its environment (after the job has actually started).
I have been working on some extensions and minor customizations of the environments (think of a ```HadoopLocalEnvironment```, for example) in order to support monitoring and counters with a plan to show them to you in the next PR.
With these extensions, this would be definitely possible. Should I merge the relevant changes in this PR? Unless you think it is not necessary and have a different approach in mind.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15377533
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
--- End diff --
I mean that the ```KeySelector.getKey()``` should return ```(partition, keyOfGroup)```
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-47146152
Hey, everything is okay now, according to this page: https://travis-ci.org/apache/incubator-flink/builds/28399204 (You can ignore the one failed build, its not your fault).
So travis gave its okay to your pull request.
Now we have to wait until somebody finds time to test the pull request.
@twalthr: you do have time for that? I think we have to write a little map-reduce example and run it on our cluster. The test should verify that everything is intuitive, well documented .. and that it runs on the cluster. We could also try to run a Mahout (map reduce) algorithm from Stratosphere.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by uce <gi...@git.apache.org>.
Github user uce commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-63789051
OK. +1 to close this. I've pinged @atsikiridis.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15515720
--- Diff: flink-addons/flink-hadoop-compatibility/src/test/java/org/apache/flink/test/hadoopcompatibility/mapred/driver/HadoopWordCountVariations.java ---
@@ -0,0 +1,316 @@
+/**
+ * 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.flink.test.hadoopcompatibility.mapred.driver;
+
+
+import org.apache.flink.hadoopcompatibility.mapred.FlinkHadoopJobClient;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.io.WritableComparator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.MapReduceBase;
+import org.apache.hadoop.mapred.Mapper;
+import org.apache.hadoop.mapred.OutputCollector;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextOutputFormat;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapred.lib.LongSumReducer;
+import org.apache.hadoop.mapred.lib.TokenCountMapper;
+
+import java.io.IOException;
+import java.util.Iterator;
+
+public class HadoopWordCountVariations {
+
+ public static class TestTokenizeMap<K> extends TokenCountMapper<K> {
+ @Override
+ public void map(K key, Text value, OutputCollector<Text, LongWritable> output, Reporter reporter)
+ throws IOException {
+ final Text strippedValue = new Text(value.toString().toLowerCase().replaceAll("\\W+", " "));
+ super.map(key, strippedValue, output, reporter);
+ }
+ }
+
+ public static class NonGenericInputFormat {
+
+ public static void main(String[] args) throws Exception {
+ final String inputPath = args[0];
+ final String outputPath = args[1];
+
+ final JobConf conf = new JobConf();
+
+ conf.setInputFormat(CustomTextInputFormat.class);
+ org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(inputPath));
+
+ conf.setOutputFormat(TextOutputFormat.class);
+ TextOutputFormat.setOutputPath(conf, new Path(outputPath));
+
+ conf.setMapperClass(TestTokenizeMap.class);
+ conf.setReducerClass(LongSumReducer.class);
+ conf.setCombinerClass((LongSumReducer.class));
+
+ conf.set("mapred.textoutputformat.separator", " ");
+ conf.setOutputKeyClass(Text.class);
+ conf.setOutputValueClass(LongWritable.class);
+
+ FlinkHadoopJobClient.runJob(conf);
+ }
+
+ public static class CustomTextInputFormat extends org.apache.hadoop.mapred.TextInputFormat {
+ }
+ }
+
+ public static class StringTokenizer {
+
+ public static void main(String[] args) throws Exception {
+ final String inputPath = args[0];
+ final String outputPath = args[1];
+
+ final JobConf conf = new JobConf();
+
+ conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
+ org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(inputPath));
+
+ conf.setOutputFormat(TextOutputFormat.class);
+ TextOutputFormat.setOutputPath(conf, new Path(outputPath));
+
+ conf.setMapperClass(TestTokenizeMap.class);
+
+
+ conf.set("mapred.textoutputformat.separator", " ");
+ conf.setOutputKeyClass(Text.class);
+ conf.setOutputValueClass(LongWritable.class);
+
+ FlinkHadoopJobClient.runJob(conf);
+ }
+ }
+
+ public static class WordCountCustomGroupingComparator {
+
+ public static void main(String[] args) throws Exception {
+ final String inputPath = args[0];
+ final String outputPath = args[1];
+
+ final JobConf conf = new JobConf();
+
+ conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
+ org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(inputPath));
+
+ conf.setOutputFormat(TextOutputFormat.class);
+ TextOutputFormat.setOutputPath(conf, new Path(outputPath));
+
+ conf.setMapperClass(TestTokenizeMap.class);
+ conf.setReducerClass(LongSumReducer.class);
+ conf.setCombinerClass((LongSumReducer.class));
+ conf.setOutputValueGroupingComparator(FirstLetterComparator.class);
+
+ conf.set("mapred.textoutputformat.separator", " ");
+ conf.setOutputKeyClass(Text.class);
+ conf.setOutputValueClass(LongWritable.class);
+
+ FlinkHadoopJobClient.runJob(conf);
+ }
+
+ //First Letters only.
+ public static class FirstLetterComparator extends WritableComparator {
+
+ public FirstLetterComparator() {
+ super(Text.class, true);
+ }
+
+ @Override
+ public int compare(WritableComparable t1, WritableComparable t2) {
+ final Text key1 = (Text) t1;
+ final IntWritable key1Char = new IntWritable(key1.charAt(0));
+ final Text key2 = (Text) t2;
+ final IntWritable key2Char = new IntWritable(key2.charAt(0));
+ return key1Char.compareTo(key2Char);
+ }
+ }
+ }
+
+ public static class WordCountCustomPartitioner {
+ public static void main(String[] args) throws Exception {
+ final String inputPath = args[0];
+ final String outputPath = args[1];
+
+ final JobConf conf = new JobConf();
+
+ conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
+ org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(inputPath));
+
+ conf.setOutputFormat(TextOutputFormat.class);
+ TextOutputFormat.setOutputPath(conf, new Path(outputPath));
+
+ conf.setMapperClass(TestTokenizeMap.class);
+ conf.setReducerClass(LongSumReducer.class);
+ conf.setCombinerClass((LongSumReducer.class));
+ conf.setPartitionerClass(MyPartitioner.class);
+ conf.setNumReduceTasks(5); //Will be ignored!
+
+ conf.set("mapred.textoutputformat.separator", " ");
+ conf.setOutputKeyClass(Text.class);
+ conf.setOutputValueClass(LongWritable.class);
+
+ FlinkHadoopJobClient.runJob(conf);
+ }
+
+
+ public static class MyPartitioner<Text, LongWritable> extends HashPartitioner<Text, LongWritable> {
+
+ @Override
+ public int getPartition(Text key, LongWritable value, int numReduceTasks) {
+ return 0;
+ }
+ }
+ }
+
+ public static class WordCountDifferentCombiner {
+
+ public static void main(String[] args) throws Exception {
+ final String inputPath = args[0];
+ final String outputPath = args[1];
+
+ final JobConf conf = new JobConf();
+
+ conf.setInputFormat(org.apache.hadoop.mapred.TextInputFormat.class);
+ org.apache.hadoop.mapred.TextInputFormat.addInputPath(conf, new Path(inputPath));
+
+ conf.setOutputFormat(TextOutputFormat.class);
+ TextOutputFormat.setOutputPath(conf, new Path(outputPath));
+
+ conf.setMapperClass(TestTokenizeMap.class);
+ conf.setCombinerClass((LongSumReducer.class));
+ conf.setReducerClass(TestReducer.class);
--- End diff --
Use a Reducer that is not derived from the combiner and has different return types to check if the combiner/reducer type handling is correct.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48971465
Hi @rmetzger , I believe I have implemented your observations along with some other minor stuff. Do you think I should rebase and change the name now or in a seperate PR? Moreover, the sorting and the Distributed cache support are ready, do you think I should include them? Then we can be sure we have all the programming interfaces in this PR. I can do that tomorrow. What do you think?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48532118
I was following these instructions: http://www.mail-archive.com/core-user@hadoop.apache.org/msg04066.html and it worked! So the problem was indeed my input format!
I think the pull request is ready to merge once the `NullPointerException` (inherited methods which are not working) issue has been resolved.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15377406
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,318 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.flink.api.java.DataSet;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.hadoopcompatibility.mapred.utils.HadoopIdentityReduce;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopGroupingKeySelector;
+import org.apache.flink.hadoopcompatibility.mapred.wrapper.HadoopPartitioner;
+import org.apache.flink.types.TypeInformation;
+import org.apache.flink.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.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.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 Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final ExecutionEnvironment environment;
+ private Configuration hadoopConf;
+
+ public FlinkHadoopJobClient() throws IOException {
+ this(new Configuration());
+ }
+
+ public FlinkHadoopJobClient(JobConf jobConf) throws IOException {
+ this(new Configuration(jobConf));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf) throws IOException{
+ this(hadoopConf, (ExecutionEnvironment.getExecutionEnvironment()));
+ }
+
+ public FlinkHadoopJobClient(Configuration hadoopConf, ExecutionEnvironment environment) throws IOException {
+ super(new JobConf(hadoopConf));
+ this.hadoopConf = hadoopConf;
+ this.environment = environment;
+ }
+
+ /**
+ * Submits a Hadoop job to Flink (as described by the JobConf) and returns after the job has been completed.
+ */
+ public static RunningJob runJob(JobConf hadoopJobConf) throws IOException{
+ final FlinkHadoopJobClient jobClient = new FlinkHadoopJobClient(hadoopJobConf);
+ final RunningJob job = jobClient.submitJob(hadoopJobConf);
+ job.waitForCompletion();
+ return job;
+ }
+
+ /**
+ * Submits a job to Flink 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{
+
+ //setting up the inputFormat for the job
+ final DataSet input = environment.createInput(getFlinkInputFormat(hadoopJobConf));
+
+ final Mapper mapper = InstantiationUtil.instantiate(hadoopJobConf.getMapperClass());
+ final Class mapOutputKeyClass = hadoopJobConf.getMapOutputKeyClass();
+ final Class mapOutputValueClass = hadoopJobConf.getMapOutputValueClass();
+ final FlatMapOperator mapped = input.flatMap(new HadoopMapFunction(mapper, mapOutputKeyClass,
+ mapOutputValueClass));
+ mapped.setParallelism(getMapParallelism(hadoopJobConf));
+
+ //Partitioning
--- End diff --
What if this unified partitioner / groupingcomparator gives us ```( (partition, keyOfGroup), (K,V) ) ``` Would that also be wrong ?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by rmetzger <gi...@git.apache.org>.
Github user rmetzger commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-48327027
Okay, I've taken this Job: https://github.com/apache/hadoop-common/blob/branch-1.2/src/examples/org/apache/hadoop/examples/Join.java and replaced the `JobClient` by the `StratosphereHadoopJobClient`.
The first thing I got is the following:
```
Exception in thread "main" java.lang.NullPointerException
at org.apache.hadoop.mapred.JobClient.getClusterStatus(JobClient.java:1298)
at org.apache.hadoop.mapred.JobClient.getClusterStatus(JobClient.java:1285)
at eu.stratosphere.test.hadoop.Join.run(Join.java:87)
at org.apache.hadoop.util.ToolRunner.run(ToolRunner.java:65)
at eu.stratosphere.test.hadoop.Join.main(Join.java:172)
```
I think you should either throw a "UnsupportedOperationException" or implement support for this. (The same applies to all the other methods that you've inherited from `JobClient` which are not working)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-63797456
@fhueske @uce ok guys. I'd love to continue working on it hopefully very soon. So I'll open a new issue on JIRA or discuss with @fhueske . :)
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50459914
Hi @atsikiridis,
please add new commits to this PR and do not squash and force push the branch. Otherwise, all previous code comments are gone and not longer available.
The `FlinkHadoopJobClient.submitJob()` method looks much better and I checked that Configuration values are correctly passed.
There are a few issues that I commented inline.
Also please add JavaDocs to all public methods of user-facing classes and check if you can get rid of some of the `rawType` warnings (use `@SuppressWarnings("rawtype")` if not possible).
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by atsikiridis <gi...@git.apache.org>.
Github user atsikiridis commented on the pull request:
https://github.com/apache/incubator-flink/pull/37#issuecomment-50000102
Hi @fhueske, concerning this last issue with the configurable interfaces it has been fixed in my branch with this distributed cache, but not here. I will merge them today. Thank you for reviewing.
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] incubator-flink pull request: hadoopcompatibility: Implementations...
Posted by fhueske <gi...@git.apache.org>.
Github user fhueske commented on a diff in the pull request:
https://github.com/apache/incubator-flink/pull/37#discussion_r15569898
--- Diff: flink-addons/flink-hadoop-compatibility/src/main/java/org/apache/flink/hadoopcompatibility/mapred/FlinkHadoopJobClient.java ---
@@ -0,0 +1,427 @@
+/**
+ * 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.flink.hadoopcompatibility.mapred;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.flink.api.common.io.OutputFormat;
+import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.operators.DataSource;
+import org.apache.flink.api.java.operators.FlatMapOperator;
+import org.apache.flink.api.java.operators.ReduceGroupOperator;
+import org.apache.flink.api.java.operators.UnsortedGrouping;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.GlobalConfiguration;
+import org.apache.flink.types.TypeInformation;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.ClusterStatus;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.JobClient;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobID;
+import org.apache.hadoop.mapred.JobQueueInfo;
+import org.apache.hadoop.mapred.JobStatus;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mapred.RunningJob;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapred.TaskCompletionEvent;
+import org.apache.hadoop.mapred.TaskReport;
+import org.apache.hadoop.mapred.lib.HashPartitioner;
+import org.apache.hadoop.mapreduce.security.token.delegation.DelegationTokenIdentifier;
+import org.apache.hadoop.security.token.Token;
+
+import java.io.IOException;
+
+/**
+ * The user's view of a Hadoop Job executed on a Flink cluster.
+ */
+public class FlinkHadoopJobClient extends JobClient {
+
+ private final static int TASK_SLOTS = GlobalConfiguration.getInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, -1);
--- End diff --
My intuition is that it would be good to have as many slots as requested tasks if a job is locally run from an IDE. If you run a program with DOP > 1 from an IDE, you want to test if it is working correctly when run in parallel or debug it. So in this case the slot mechanism should be somehow deactivated (or as many slots created as the max DOP of the job to run).
I'm not so familiar with the scheduling code. @StephanEwen can this be done somehow?
---
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 infrastructure@apache.org or file a JIRA ticket
with INFRA.
---