You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mrql.apache.org by fe...@apache.org on 2014/10/20 23:19:21 UTC
git commit: MRQL-55: Add support for Hadoop Sequence input format in
flink mode
Repository: incubator-mrql
Updated Branches:
refs/heads/master 1f7a1820b -> 192e58c30
MRQL-55: Add support for Hadoop Sequence input format in flink mode
Project: http://git-wip-us.apache.org/repos/asf/incubator-mrql/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-mrql/commit/192e58c3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-mrql/tree/192e58c3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-mrql/diff/192e58c3
Branch: refs/heads/master
Commit: 192e58c305fc03734037c38b82a5e02f02a5cc16
Parents: 1f7a182
Author: fegaras <fe...@cse.uta.edu>
Authored: Mon Oct 20 16:18:51 2014 -0500
Committer: fegaras <fe...@cse.uta.edu>
Committed: Mon Oct 20 16:18:51 2014 -0500
----------------------------------------------------------------------
bin/mrql.flink | 7 +-
conf/mrql-env.sh | 12 +-
.../java/org/apache/mrql/MapReduceAlgebra.java | 2 +-
core/src/main/java/org/apache/mrql/Plan.java | 4 +-
.../java/org/apache/mrql/FDataContainer.java | 39 -------
.../org/apache/mrql/FlinkBinaryInputFormat.java | 2 +-
.../java/org/apache/mrql/FlinkDataSource.java | 3 +-
.../java/org/apache/mrql/FlinkEvaluator.gen | 67 +++---------
.../apache/mrql/FlinkSequenceInputFormat.java | 109 +++++++++++++++++++
pom.xml | 13 ++-
10 files changed, 154 insertions(+), 104 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/bin/mrql.flink
----------------------------------------------------------------------
diff --git a/bin/mrql.flink b/bin/mrql.flink
index cd3eab4..40d7777 100755
--- a/bin/mrql.flink
+++ b/bin/mrql.flink
@@ -34,7 +34,7 @@ MRQL_JAR=`ls "$MRQL_HOME"/lib/mrql-flink-*.jar`
FULL_JAR="/tmp/${USER}_mrql_flink.jar"
CLASS_DIR="/tmp/${USER}_mrql_classes"
-export FLINK_HOME FLINK_JARS FLINK_MASTER
+export FLINK_HOME FLINK_JARS FLINK_MASTER FS_DEFAULT_NAME
if [[ ($MRQL_JAR -nt $FULL_JAR) ]]; then
rm -rf $CLASS_DIR
@@ -45,6 +45,7 @@ if [[ ($MRQL_JAR -nt $FULL_JAR) ]]; then
$JAVA_HOME/bin/jar xf $GEN_JAR
$JAVA_HOME/bin/jar xf $CORE_JAR
$JAVA_HOME/bin/jar xf $MRQL_JAR
+ $JAVA_HOME/bin/jar xf $HDFS_JAR
cd ..
$JAVA_HOME/bin/jar cf $FULL_JAR -C $CLASS_DIR .
popd > /dev/null
@@ -54,6 +55,10 @@ if [ "$1" == "-dist" ]; then
if ([ -a $FLINK_HOME/conf/.yarn-jobmanager ]); then
export FLINK_MASTER=`cat $FLINK_HOME/conf/.yarn-jobmanager`
fi
+ if ([ -a $FLINK_HOME/conf/.yarn-properties ]); then
+ . $FLINK_HOME/conf/.yarn-properties
+ export FLINK_MASTER=$jobManager
+ fi
args=" -flink $*"
args=${args// /\!}
$FLINK_HOME/bin/flink run $FULL_JAR -c org.apache.mrql.Main -v args $args
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/conf/mrql-env.sh
----------------------------------------------------------------------
diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index 808d151..902d910 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -56,12 +56,12 @@ FS_DEFAULT_NAME=hdfs://localhost:9000/
# Optional: Hama configuration. Supports versions 0.6.2, 0.6.3, and 0.6.4
-HAMA_VERSION=0.6.3
+HAMA_VERSION=0.6.4
# The Hadoop installation directory
HAMA_HOME=${HOME}/hama-${HAMA_VERSION}
# The Hama configuration (as defined in hama-site.xml)
-BSP_MASTER_ADDRESS=localhost:40000
-HAMA_ZOOKEEPER_QUORUM=localhost
+BSP_MASTER_ADDRESS=`hostname`:40000
+HAMA_ZOOKEEPER_QUORUM=`hostname`
# Optional: Spark configuration. Supports versions 1.0.0, 1.0.2, and 1.1.0
@@ -84,14 +84,16 @@ SPARK_WORKER_MEMORY=1G
SPARK_MASTER_MEMORY=512M
-# Optional: Flink configuration. Supports version 0.6-incubating
+# Optional: Flink configuration. Supports versions 0.6-incubating, 0.6.1-incubating, and 0.7.0-incubating
# Note: for yarn, set yarn.nodemanager.vmem-check-enabled to false in yarn-site.xml
FLINK_VERSION=yarn-0.6.1-incubating
# Flink installation directory
FLINK_HOME=${HOME}/flink-${FLINK_VERSION}
# (use this for a Flink snapshot):
#FLINK_HOME=${HOME}/flink-${FLINK_VERSION}/flink-dist/target/flink-${FLINK_VERSION}-bin/flink-${FLINK_VERSION}
-# Flink JobManager (on a yarn cluster, it is printed when you run yarn-session.sh)
+# Hadoop HDFS: needed for Sequence files in Flink mode
+HDFS_JAR=${HADOOP_HOME}/share/hadoop/hdfs/hadoop-hdfs-${HADOOP_VERSION}.jar
+# Flink JobManager (it is derived automatically on a yarn cluster)
if [ "$FLINK_MASTER" = "" ]; then
FLINK_MASTER=`hostname`:6123
fi
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
index 397166b..beaad27 100644
--- a/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
+++ b/core/src/main/java/org/apache/mrql/MapReduceAlgebra.java
@@ -656,7 +656,7 @@ final public class MapReduceAlgebra {
/** parse a text document using a given parser
* @param parser the parser
- * @param file the text document (local file or directory of files)
+ * @param path the text document (local file or directory of files)
* @param args the arguments to pass to the parser
* @return a lazy bag that contains the parsed data
*/
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/core/src/main/java/org/apache/mrql/Plan.java
----------------------------------------------------------------------
diff --git a/core/src/main/java/org/apache/mrql/Plan.java b/core/src/main/java/org/apache/mrql/Plan.java
index 116a822..94fd011 100644
--- a/core/src/main/java/org/apache/mrql/Plan.java
+++ b/core/src/main/java/org/apache/mrql/Plan.java
@@ -89,7 +89,9 @@ public class Plan {
if (!Config.compile_functional_arguments)
return;
Path local_path = new Path("file://"+Compiler.jar_path);
- if (Config.spark_mode)
+ if (Config.flink_mode)
+ conf.set("mrql.jar.path",Compiler.jar_path);
+ else if (Config.spark_mode)
conf.set("mrql.jar.path",local_path.toString());
else {
// distribute the jar file with the compiled arguments to all clients
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/flink/src/main/java/org/apache/mrql/FDataContainer.java
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FDataContainer.java b/flink/src/main/java/org/apache/mrql/FDataContainer.java
deleted file mode 100644
index 07eaa22..0000000
--- a/flink/src/main/java/org/apache/mrql/FDataContainer.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/**
- * 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.mrql;
-
-import java.io.IOException;
-import org.apache.flink.core.memory.DataInputView;
-import org.apache.flink.core.memory.DataOutputView;
-import org.apache.flink.core.io.IOReadableWritable;
-
-final public class FDataContainer extends MRContainer implements IOReadableWritable, Comparable<MRContainer> {
-
- FDataContainer ( MRData d ) { super(d); }
-
- @Override
- public void read ( DataInputView in ) throws IOException {
- data = super.read(in);
- }
-
- @Override
- public void write ( DataOutputView out ) throws IOException {
- super.write(out);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/flink/src/main/java/org/apache/mrql/FlinkBinaryInputFormat.java
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkBinaryInputFormat.java b/flink/src/main/java/org/apache/mrql/FlinkBinaryInputFormat.java
index 5969c1e..bd6f66d 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkBinaryInputFormat.java
+++ b/flink/src/main/java/org/apache/mrql/FlinkBinaryInputFormat.java
@@ -23,7 +23,7 @@ import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.api.common.io.FileInputFormat;
-/** Input format for sequence files */
+/** Input format for binary files */
final public class FlinkBinaryInputFormat extends FlinkMRQLFileInputFormat {
public FlinkBinaryInputFormat () {}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/flink/src/main/java/org/apache/mrql/FlinkDataSource.java
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkDataSource.java b/flink/src/main/java/org/apache/mrql/FlinkDataSource.java
index 9a5bbe7..3dfc9e9 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkDataSource.java
+++ b/flink/src/main/java/org/apache/mrql/FlinkDataSource.java
@@ -24,7 +24,8 @@ import java.io.IOException;
import java.io.Serializable;
import org.apache.hadoop.conf.Configuration;
import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.functions.RichFlatMapFunction;
+import org.apache.flink.api.common.functions.*;
+import org.apache.flink.api.java.functions.*;
import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.common.accumulators.LongCounter;
import org.apache.flink.core.memory.DataInputView;
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
index e5c92a6..d14def5 100644
--- a/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
+++ b/flink/src/main/java/org/apache/mrql/FlinkEvaluator.gen
@@ -28,23 +28,14 @@ import org.apache.flink.util.Collector;
import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.api.common.io.FileInputFormat;
-import org.apache.flink.api.common.operators.Order;
-import org.apache.flink.api.java.ExecutionEnvironment;
-import org.apache.flink.api.java.RemoteEnvironment;
+import org.apache.flink.api.common.functions.*;
+import org.apache.flink.api.common.operators.*;
+import org.apache.flink.api.java.*;
import org.apache.flink.api.java.DataSet;
-import org.apache.flink.api.java.IterativeDataSet;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.io.LocalCollectionOutputFormat;
-import org.apache.flink.api.java.functions.RichMapFunction;
-import org.apache.flink.api.java.functions.RichFlatMapFunction;
-import org.apache.flink.api.java.functions.RichReduceFunction;
-import org.apache.flink.api.java.functions.RichGroupReduceFunction;
-import org.apache.flink.api.java.functions.RichCoGroupFunction;
-import org.apache.flink.api.java.functions.RichCrossFunction;
-import org.apache.flink.api.java.functions.RichJoinFunction;
-import org.apache.flink.api.java.functions.RichFilterFunction;
-import org.apache.flink.api.java.functions.KeySelector;
-import org.apache.flink.api.java.operators.GroupReduceOperator;
+import org.apache.flink.api.java.functions.*;
+import org.apache.flink.api.java.operators.*;
/** Evaluates physical plans in Apache Flink mode */
@@ -68,6 +59,7 @@ final public class FlinkEvaluator extends Evaluator {
flink_env = ExecutionEnvironment.createLocalEnvironment(Config.nodes);
// curently, the compiler doesn't work in local mode
Config.compile_functional_arguments = false;
+ Plan.conf.set("fs.default.name","file:///");
Config.write(Plan.conf);
} else if (Config.distributed_mode) {
String master_node = System.getenv("FLINK_MASTER");
@@ -78,6 +70,7 @@ final public class FlinkEvaluator extends Evaluator {
throw new Error("Need both host name and port number for the Flink application master: "+master_node);
master_host = m[0];
master_port = Integer.parseInt(m[1]);
+ Plan.conf.set("fs.default.name",System.getenv("FS_DEFAULT_NAME"));
}
}
@@ -94,8 +87,7 @@ final public class FlinkEvaluator extends Evaluator {
Config.nodes,flink_jar.toURI().getPath(),Plan.conf.get("mrql.jar.path"));
else flink_env = ExecutionEnvironment.createRemoteEnvironment(master_host,master_port,
Config.nodes,flink_jar.toURI().getPath());
- } else if (Config.distributed_mode)
- flink_env.setDefaultLocalParallelism(Config.nodes);
+ } else flink_env.setDefaultLocalParallelism(Config.nodes);
} catch (Exception ex) {
throw new Error("Cannot initialize the Flink evaluator: "+ex);
}
@@ -112,7 +104,7 @@ final public class FlinkEvaluator extends Evaluator {
/** return the FileInputFormat for binary files */
final public Class<? extends MRQLFileInputFormat> binaryInputFormat () {
- return FlinkBinaryInputFormat.class;
+ return FlinkSequenceInputFormat.class;
}
/** return the FileInputFormat for data generator files */
@@ -142,46 +134,13 @@ final public class FlinkEvaluator extends Evaluator {
/** for dumped data to a file, return the MRQL type of the data */
@Override
public Tree get_type ( String file ) {
- try {
- Path path = new Path(absolute_path(file));
- FileSystem fs = path.getFileSystem();
- BufferedReader ftp = new BufferedReader(new InputStreamReader(fs.open(path.suffix(".type"))));
- String s[] = ftp.readLine().split("@");
- ftp.close();
- if (s.length != 2 )
- return null;
- if (s[0].equals("2"))
- throw new Error("This is a Hadoop Sequence and cannot be read as a Flink binary file: "+file);
- else if (!s[0].equals("3"))
- throw new Error("The binary file has been created in java mode and cannot be read in hadoop mode: "+file);
- else return Tree.parse(s[1]);
- } catch (Exception e) {
- return null;
- }
+ return super.get_type(absolute_path(file));
}
- /** dump MRQL data into a Flink binary file instead of a Hadoop Sequence file */
+ /** dump MRQL data into a Hadoop Sequence file */
@Override
public void dump ( String file, Tree type, MRData data ) throws Exception {
- Path path = new Path(absolute_path(file));
- FileSystem fs = path.getFileSystem();
- PrintStream ftp = new PrintStream(fs.create(path.suffix(".type"),true));
- ftp.print("3@"+type.toString()+"\n");
- ftp.close();
- if (data instanceof MR_dataset)
- data = Plan.collect(((MR_dataset)data).dataset());
- FlinkOutputFormat of = new FlinkOutputFormat();
- of.setOutputFilePath(path);
- of.setWriteMode(org.apache.flink.core.fs.FileSystem.WriteMode.OVERWRITE);
- of.open(0,2);
- if (data instanceof Bag) {
- final FData container = new FData(new MR_int(0));
- for ( MRData e: (Bag)data ) {
- container.data = e;
- of.writeRecord(container);
- }
- } else of.writeRecord(new FData(data));
- of.close();
+ super.dump(absolute_path(file),type,data);
}
/** dump MRQL data into a text CVS file */
@@ -682,7 +641,7 @@ final public class FlinkEvaluator extends Evaluator {
case BinarySource(`file,_):
final String path = absolute_path(((MR_string)evalE(file,null)).get());
new BinaryDataSource(path,Plan.conf);
- final FileInputFormat<FData> sf = new FlinkBinaryInputFormat().inputFormat(path);
+ final FileInputFormat<FData> sf = new FlinkSequenceInputFormat().inputFormat(path);
return flink_env.createInput(sf).map(new restore_global_functions());
case ParsedSource(`parser,`file,...args):
final String path = absolute_path(((MR_string)evalE(file,null)).get());
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/flink/src/main/java/org/apache/mrql/FlinkSequenceInputFormat.java
----------------------------------------------------------------------
diff --git a/flink/src/main/java/org/apache/mrql/FlinkSequenceInputFormat.java b/flink/src/main/java/org/apache/mrql/FlinkSequenceInputFormat.java
new file mode 100644
index 0000000..00d173c
--- /dev/null
+++ b/flink/src/main/java/org/apache/mrql/FlinkSequenceInputFormat.java
@@ -0,0 +1,109 @@
+/**
+ * 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.mrql;
+
+import java.io.*;
+import java.util.List;
+import java.util.ArrayList;
+import org.apache.flink.api.common.io.BinaryInputFormat;
+import org.apache.flink.core.memory.DataInputView;
+import org.apache.flink.api.common.io.FileInputFormat;
+import org.apache.flink.core.fs.FileInputSplit;
+import org.apache.flink.core.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.io.SequenceFile;
+import org.apache.hadoop.conf.Configuration;
+
+
+/** Input format for Hadoop Sequence files */
+final public class FlinkSequenceInputFormat extends FlinkMRQLFileInputFormat {
+ public FlinkSequenceInputFormat () {}
+
+ public static final class FDataInputFormat extends BinaryInputFormat<FData> {
+ private SequenceFile.Reader in;
+ private long start;
+ private long end;
+ private boolean more = true;
+ MRContainer key = new MRContainer(new MR_int(0));
+ MRContainer value = new MRContainer(new MR_int(0));
+
+ @Override
+ public void open ( FileInputSplit split ) throws IOException {
+ Path path = new Path(split.getPath().toString());
+ if (Plan.conf == null)
+ Plan.conf = new Configuration();
+ FileSystem fs = path.getFileSystem(Plan.conf);
+ in = new SequenceFile.Reader(fs,path,Plan.conf);
+ end = split.getStart()+split.getLength();
+ if (split.getStart() > in.getPosition())
+ in.sync(split.getStart()); // sync to start
+ start = in.getPosition();
+ more = start < end;
+ }
+
+ @Override
+ public FData nextRecord ( FData record ) throws IOException {
+ if (!more)
+ return null;
+ long pos = in.getPosition();
+ if (!in.next(key) || (pos >= end && in.syncSeen())) {
+ more = false;
+ return null;
+ };
+ in.getCurrentValue(value);
+ return new FData(value.data());
+ }
+
+ @Override
+ public boolean reachedEnd () {
+ return !more;
+ }
+
+ @Override
+ public void close () throws IOException {
+ super.close();
+ in.close();
+ }
+
+ @Override
+ protected List<FileStatus> getFiles () throws IOException {
+ List<FileStatus> files = super.getFiles();
+ List<FileStatus> sfiles = new ArrayList<FileStatus>();
+ for ( FileStatus file: files ) {
+ String fname = file.getPath().getName();
+ if (!fname.startsWith("_") && !fname.endsWith(".crc"))
+ sfiles.add(file);
+ };
+ return sfiles;
+ }
+
+ @Override
+ protected FData deserialize ( FData reuse, DataInputView in ) throws IOException {
+ reuse.read(in);
+ return reuse;
+ }
+ }
+
+ /** the Flink input format for this input */
+ public FileInputFormat<FData> inputFormat ( String path ) {
+ FDataInputFormat sf = new FDataInputFormat();
+ sf.setFilePath(path.toString());
+ return sf;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/192e58c3/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 7af8ca9..a0dc930 100644
--- a/pom.xml
+++ b/pom.xml
@@ -43,10 +43,21 @@
<relativePath />
</parent>
+ <repositories>
+ <!-- Needed for flink snapshots -->
+ <repository>
+ <id>apache.snapshots</id>
+ <name>Apache Development Snapshot Repository</name>
+ <url>https://repository.apache.org/content/repositories/snapshots/</url>
+ <releases><enabled>false</enabled></releases>
+ <snapshots><enabled>true</enabled></snapshots>
+ </repository>
+ </repositories>
+
<properties>
<hadoop.version>1.2.1</hadoop.version>
<yarn.version>2.2.0</yarn.version>
- <hama.version>0.6.3</hama.version>
+ <hama.version>0.6.4</hama.version>
<spark.version>1.1.0</spark.version>
<scala.version>2.10</scala.version>
<flink.version>0.6.1-incubating</flink.version>