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 2013/08/27 14:27:58 UTC
[20/21] MRQL-2: Mavenize MRQL
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hadoop/taskcontroller.cfg
----------------------------------------------------------------------
diff --git a/conf/conf-hadoop/taskcontroller.cfg b/conf/conf-hadoop/taskcontroller.cfg
deleted file mode 100644
index 9cc88d0..0000000
--- a/conf/conf-hadoop/taskcontroller.cfg
+++ /dev/null
@@ -1,3 +0,0 @@
-mapreduce.cluster.local.dir=#configured value of mapreduce.cluster.local.dir. It can be a list of comma separated paths.
-hadoop.log.dir=#configured value of hadoop.log.dir.
-mapreduce.tasktracker.group=#configured value of mapreduce.tasktracker.group
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hama/groomservers
----------------------------------------------------------------------
diff --git a/conf/conf-hama/groomservers b/conf/conf-hama/groomservers
deleted file mode 100755
index 2fbb50c..0000000
--- a/conf/conf-hama/groomservers
+++ /dev/null
@@ -1 +0,0 @@
-localhost
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hama/hama-default.xml
----------------------------------------------------------------------
diff --git a/conf/conf-hama/hama-default.xml b/conf/conf-hama/hama-default.xml
deleted file mode 100755
index 5758670..0000000
--- a/conf/conf-hama/hama-default.xml
+++ /dev/null
@@ -1,208 +0,0 @@
-<?xml version="1.0"?>
-<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
-<!--
-/**
- * Copyright 2009 The Apache Software Foundation
- *
- * 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.
- */
--->
-<configuration>
- <property>
- <name>bsp.master.address</name>
- <value>local</value>
- <description>The address of the bsp master server. Either the
- literal string "local" or a host[:port] (where host is a name or
- IP address) for distributed mode.
- </description>
- </property>
- <property>
- <name>bsp.master.port</name>
- <value>40000</value>
- <description>The port master should bind to.</description>
- </property>
- <property>
- <name>bsp.peer.port</name>
- <value>61000</value>
- <description>The port an groom server binds to.</description>
- </property>
- <property>
- <name>bsp.groom.rpc.port</name>
- <value>50000</value>
- <description>The port an groom rpc binds to.</description>
- </property>
- <property>
- <name>bsp.http.infoserver.port</name>
- <value>40013</value>
- <description>The port where the web-interface can be seen.
- </description>
- </property>
- <property>
- <name>bsp.http.groomserver.port</name>
- <value>40015</value>
- <description>The port where the web-interface can be seen.
- </description>
- </property>
- <property>
- <name>bsp.groom.report.address</name>
- <value>127.0.0.1:0</value>
- <description>The interface and port that groom server listens on.
- Since it is only connected to by the tasks, it uses the local interface.
- EXPERT ONLY. Should only be changed if your host does not have the loopback
- interface.</description>
- </property>
- <property>
- <name>bsp.groomserver.pingperiod</name>
- <value>5000</value>
- <description>Periodicity in milliseconds that every BSP task should send its hearbeat ping latest. If a task fails to do so, the groom server would deem the task as failed.</description>
- </property>
- <property>
- <name>bsp.system.dir</name>
- <value>${hadoop.tmp.dir}/bsp/system</value>
- <description>The shared directory where BSP stores control files.
- </description>
- </property>
- <property>
- <name>bsp.local.dir</name>
- <value>${hadoop.tmp.dir}/bsp/local</value>
- <description>local directory for temporal store.</description>
- </property>
- <property>
- <name>hama.tmp.dir</name>
- <value>/tmp/hama-${user.name}</value>
- <description>Temporary directory on the local filesystem.</description>
- </property>
- <property>
- <name>bsp.disk.queue.dir</name>
- <value>${hama.tmp.dir}/messages/</value>
- <description>Temporary directory on the local message buffer on disk.</description>
- </property>
- <property>
- <name>bsp.child.java.opts</name>
- <value>-Xmx512m</value>
- <description>Java opts for the groom server child processes.
- The following symbol, if present, will be interpolated: @taskid@ is replaced
- by current TaskID. Any other occurrences of '@' will go unchanged.
- For example, to enable verbose gc logging to a file named for the taskid in
- /tmp and to set the heap maximum to be a gigabyte, pass a 'value' of:
- -Xmx1024m -verbose:gc -Xloggc:/tmp/@taskid@.gc
- The configuration variable bsp.child.ulimit can be used to control the
- maximum virtual memory of the child processes.
- </description>
- </property>
- <property>
- <name>hama.messenger.compression.class</name>
- <value>org.apache.hama.bsp.message.compress.SnappyCompressor</value>
- <description>The message compression algorithm to choose.</description>
- </property>
- <property>
- <name>bsp.local.tasks.maximum</name>
- <value>10</value>
- <description>Number of tasks that run in parallel when in local mode.</description>
- </property>
- <property>
- <name>bsp.tasks.maximum</name>
- <value>3</value>
- <description>The maximum number of BSP tasks that will be run simultaneously
- by a groom server.</description>
- </property>
- <property>
- <name>bsp.checkpoint.enabled</name>
- <value>false</value>
- <description>Enable Hama to checkpoint the messages transferred among BSP tasks during the BSP synchronization period.</description>
- </property>
- <property>
- <name>bsp.checkpoint.interval</name>
- <value>1</value>
- <description>If bsp.checkpoint.enabled is set to true, the checkpointing is initiated on the valueth synchronization process of BSP tasks.</description>
- </property>
-
-
- <!--
- Beginning of properties that are directly mapped from ZooKeeper's zoo.cfg.
- All properties with an "hama.zookeeper.property." prefix are converted for
- ZooKeeper's configuration. Hence, if you want to add an option from zoo.cfg,
- e.g. "initLimit=10" you would append the following to your configuration:
- <property>
- <name>hama.zookeeper.property.initLimit</name>
- <value>10</value>
- </property>
- -->
- <property>
- <name>hama.zookeeper.property.initLimit</name>
- <value>10</value>
- <description>Property from ZooKeeper's config zoo.cfg.
- The number of ticks that the initial synchronization phase can take.
- </description>
- </property>
- <property>
- <name>hama.zookeeper.property.syncLimit</name>
- <value>5</value>
- <description>Property from ZooKeeper's config zoo.cfg.
- The number of ticks that can pass between sending a request and getting an
- acknowledgment.
- </description>
- </property>
- <property>
- <name>hama.zookeeper.property.dataDir</name>
- <value>${hama.tmp.dir}/zookeeper</value>
- <description>Property from ZooKeeper's config zoo.cfg.
- The directory where the snapshot is stored.
- </description>
- </property>
- <property>
- <name>hama.zookeeper.property.clientPort</name>
- <value>21810</value>
- <description>Property from ZooKeeper's config zoo.cfg.
- The port at which the clients will connect.
- </description>
- </property>
- <property>
- <name>hama.zookeeper.property.tickTime</name>
- <value>3000</value>
- <description></description>
- </property>
- <property>
- <name>hama.zookeeper.property.maxClientCnxns</name>
- <value>30</value>
- <description>Property from ZooKeeper's config zoo.cfg.
- Limit on number of concurrent connections (at the socket level) that a
- single client, identified by IP address, may make to a single member of
- the ZooKeeper ensemble. Set high to avoid zk connection issues running
- standalone and pseudo-distributed.
- </description>
- </property>
-
- <property>
- <name>hama.messenger.class</name>
- <value>org.apache.hama.bsp.message.AvroMessageManagerImpl</value>
- </property>
-
- <property>
- <name>hama.zookeeper.quorum</name>
- <value>localhost</value>
- <description>Comma separated list of servers in the ZooKeeper quorum.
- For example, "host1.mydomain.com,host2.mydomain.com,host3.mydomain.com".
- By default this is set to localhost for local and pseudo-distributed modes
- of operation. For a fully-distributed setup, this should be set to a full
- list of ZooKeeper quorum servers. If HAMA_MANAGES_ZK is set in hama-env.sh
- this is the list of servers which we will start/stop ZooKeeper on.
- </description>
- </property>
-
- <!-- End of properties that are directly mapped from ZooKeeper's zoo.cfg -->
-</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hama/hama-env.sh
----------------------------------------------------------------------
diff --git a/conf/conf-hama/hama-env.sh b/conf/conf-hama/hama-env.sh
deleted file mode 100755
index 9df6311..0000000
--- a/conf/conf-hama/hama-env.sh
+++ /dev/null
@@ -1,37 +0,0 @@
-#
-#/**
-# * Copyright 2007 The Apache Software Foundation
-# *
-# * 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.
-# */
-
-# Set environment variables here.
-
-# The java implementation to use. Default is from mrql-env.sh
-# export JAVA_HOME=/root/jdk
-
-# Where log files are stored. $HAMA_HOME/logs by default.
-# export HAMA_LOG_DIR=${HAMA_HOME}/logs
-
-# The maximum amount of heap to use, in MB. Default is 1000.
-# export HAMA_HEAPSIZE=1000
-
-# Extra ssh options. Empty by default.
-# export HAMA_SSH_OPTS="-o ConnectTimeout=1 -o SendEnv=HAMA_CONF_DIR"
-
-# Tell Hama whether it should manage it's own instance of Zookeeper or not.
-# export HAMA_MANAGES_ZK=true
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hama/hama-site.xml
----------------------------------------------------------------------
diff --git a/conf/conf-hama/hama-site.xml b/conf/conf-hama/hama-site.xml
deleted file mode 100755
index 455d8e5..0000000
--- a/conf/conf-hama/hama-site.xml
+++ /dev/null
@@ -1,37 +0,0 @@
-<?xml version="1.0"?>
- <?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
- <configuration>
- <property>
- <name>bsp.master.address</name>
- <value>localhost:40000</value>
- <description>The address of the bsp master server. Either the
- literal string "local" or a host:port for distributed mode
- </description>
- </property>
-
- <property>
- <name>fs.default.name</name>
- <value>hdfs://localhost:9000</value>
- <description>
- The name of the default file system. Either the literal string
- "local" or a host:port for HDFS.
- </description>
- </property>
-
- <property>
- <name>hama.zookeeper.quorum</name>
- <value>localhost</value>
- <description>Comma separated list of servers in the ZooKeeper Quorum.
- For example, "host1.mydomain.com,host2.mydomain.com,host3.mydomain.com".
- By default this is set to localhost for local and pseudo-distributed modes
- of operation. For a fully-distributed setup, this should be set to a full
- list of ZooKeeper quorum servers. If HAMA_MANAGES_ZK is set in hama-env.sh
- this is the list of servers which we will start/stop zookeeper on.
- </description>
- </property>
-
- <property>
- <name>bsp.tasks.maximum</name>
- <value>2</value>
- </property>
- </configuration>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/conf-hama/log4j.properties
----------------------------------------------------------------------
diff --git a/conf/conf-hama/log4j.properties b/conf/conf-hama/log4j.properties
deleted file mode 100755
index 1e8fea7..0000000
--- a/conf/conf-hama/log4j.properties
+++ /dev/null
@@ -1,81 +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.
-#
-
-# Define some default values that can be overridden by system properties
-hama.root.logger=INFO,console
-hama.log.dir=.
-hama.log.file=hama.log
-
-# Define the root logger to the system property "hama.root.logger".
-log4j.rootLogger=${hama.root.logger}
-
-# Logging Threshold
-log4j.threshhold=ALL
-
-#
-# Daily Rolling File Appender
-#
-log4j.appender.DRFA=org.apache.log4j.DailyRollingFileAppender
-log4j.appender.DRFA.File=${hama.log.dir}/${hama.log.file}
-
-# Rollver at midnight
-log4j.appender.DRFA.DatePattern=.yyyy-MM-dd
-
-# 30-day backup
-#log4j.appender.DRFA.MaxBackupIndex=30
-log4j.appender.DRFA.layout=org.apache.log4j.PatternLayout
-
-# Pattern format: Date LogLevel LoggerName LogMessage
-log4j.appender.DRFA.layout.ConversionPattern=%d{ISO8601} %p %c: %m%n
-
-# Debugging Pattern format
-#log4j.appender.DRFA.layout.ConversionPattern=%d{ISO8601} %-5p %c{2} (%F:%M(%L)) - %m%n
-
-#
-# TaskLog Appender
-#
-
-#Default values
-hama.tasklog.taskid=null
-hama.tasklog.noKeepSplits=4
-hama.tasklog.totalLogFileSize=100
-hama.tasklog.purgeLogSplits=true
-hama.tasklog.logsRetainHours=12
-
-log4j.appender.TLA=org.apache.hama.bsp.TaskLogAppender
-log4j.appender.TLA.taskId=${hama.tasklog.taskid}
-log4j.appender.TLA.totalLogFileSize=${hama.tasklog.totalLogFileSize}
-
-log4j.appender.TLA.layout=org.apache.log4j.PatternLayout
-log4j.appender.TLA.layout.ConversionPattern=%d{ISO8601} %p %c: %m%n
-
-#
-# console
-# Add "console" to rootlogger above if you want to use this
-#
-log4j.appender.console=org.apache.log4j.ConsoleAppender
-log4j.appender.console.target=System.err
-log4j.appender.console.layout=org.apache.log4j.PatternLayout
-log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n
-
-# Custom Logging levels
-
-#log4j.logger.org.apache.hadoop.fs.FSNamesystem=DEBUG
-#log4j.logger.org.apache.hadoop.dfs=DEBUG
-#log4j.logger.org.apache.hama=DEBUG
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/mrql-ant-env.sh
----------------------------------------------------------------------
diff --git a/conf/mrql-ant-env.sh b/conf/mrql-ant-env.sh
new file mode 100644
index 0000000..6fa59bf
--- /dev/null
+++ b/conf/mrql-ant-env.sh
@@ -0,0 +1,42 @@
+#/bin/bash
+#--------------------------------------------------------------------------------
+#
+# 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.
+#
+#
+# Set MRQL-specific environment variables here.
+# Used by Makefile and build.xml only
+#
+#--------------------------------------------------------------------------------
+
+# Required: The MRQL version
+MRQL_VERSION=0.9.0-incubating
+
+# Required: The jflex parser library
+# You may install it as a linux package or download it from http://jflex.de/
+JFLEX_JAR=/usr/share/java/JFlex.jar
+
+# Required: The CUP parser library
+# You may install it as a linux package or download it from http://www2.cs.tum.edu/projects/cup/
+CUP_JAR=/usr/share/java/cup.jar
+
+# Required: The JLine library
+# You may download from http://jline.sourceforge.net
+JLINE_JAR=${HOME}/mrql18/lib/jline-1.0.jar
+
+MRQL_CLASSPATH=classes:${HAMA_JAR}:${SPARK_JARS}:${HADOOP_JARS}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/conf/mrql-env.sh
----------------------------------------------------------------------
diff --git a/conf/mrql-env.sh b/conf/mrql-env.sh
index 2439a31..e537ea0 100644
--- a/conf/mrql-env.sh
+++ b/conf/mrql-env.sh
@@ -23,44 +23,42 @@
#
#--------------------------------------------------------------------------------
-# Required: The java implementation to use
+
+# Required: The java installation directory
JAVA_HOME=/root/jdk
+
# Required: Hadoop configuration
HADOOP_VERSION=1.0.3
+# The Hadoop installation directory
HADOOP_HOME=${HOME}/hadoop-${HADOOP_VERSION}
-# The Hadoop configuration directory
-HADOOP_CONF=${HOME}/conf
+# The Hadoop job trackeer (as defined in hdfs-site.xml)
+MAPRED_JOB_TRACKER=localhost:9001
+# The HDFS namenode URI (as defined in hdfs-site.xml)
+FS_DEFAULT_NAME=hdfs://localhost:9000/
+
# Optional: Hama configuration
HAMA_VERSION=0.5.0
+# The Hadoop installation directory
HAMA_HOME=${HOME}/hama-${HAMA_VERSION}
-# The Hama configuration directory
-HAMA_CONF=${HOME}/conf-hama
-HAMA_JAR=${HAMA_HOME}/hama-core-${HAMA_VERSION}.jar
+# The Hama configuration as defined in hama-site.xml
+BSP_MASTER_ADDRESS=localhost:40000
+HAMA_ZOOKEEPER_QUORUM=localhost
+
# Optional: Spark configuration
SPARK_HOME=${HOME}/spark-0.7.3
+# Location of the Scala libs
SCALA_LIB=/usr/share/java
+# URI of the Spark master node
SPARK_MASTER=spark://crete:7077
-SPARK_DEFAULT_URI=hdfs://localhost:9000/
-SPARK_JARS=${SCALA_LIB}/scala-library.jar:${SCALA_LIB}/scala-compiler.jar:${SPARK_HOME}/core/target/scala-2.9.3/classes:${SPARK_HOME}/lib_managed/jars/*:${SPARK_HOME}/lib_managed/bundles/*
-
-# Required: The jflex parser library
-# You may install it as a linux package or download it from http://jflex.de/
-JFLEX_JAR=/usr/share/java/JFlex.jar
-# Required: The CUP parser library
-# You may install it as a linux package or download it from http://www2.cs.tum.edu/projects/cup/
-CUP_JAR=/usr/share/java/cup.jar
-# Hadoop libraries are from a Hadoop tarball
-HADOOP_JARS=${HADOOP_HOME}/hadoop-core-${HADOOP_VERSION}.jar:${HADOOP_HOME}/lib/commons-cli-1.2.jar
+# Claspaths
-# Hadoop libraries are from a YARN tarball
-#HADOOP_JARS=${HADOOP_HOME}/share/hadoop/common/hadoop-common-${HADOOP_VERSION}.jar:${HADOOP_HOME}/share/hadoop/mapreduce/hadoop-mapreduce-client-core-${HADOOP_VERSION}.jar:${HADOOP_HOME}/share/hadoop/hdfs/hadoop-hdfs-${HADOOP_VERSION}.jar:${HADOOP_HOME}/share/hadoop/common/lib/hadoop-annotations-${HADOOP_VERSION}.jar:${HADOOP_HOME}/share/hadoop/common/lib/commons-cli-1.2.jar
+HADOOP_JARS=${HADOOP_HOME}/hadoop-core-${HADOOP_VERSION}.jar:${HADOOP_HOME}/lib/commons-logging-1.1.1.jar:${HADOOP_HOME}/lib/log4j-1.2.15.jar:${HADOOP_HOME}/lib/commons-cli-1.2.jar
-# Hadoop libraries are from a Cloudera package
-#HADOOP_JARS=/usr/lib/hadoop-0.20/hadoop-core.jar:/usr/lib/hadoop-0.20/lib/commons-cli-1.2.jar
+HAMA_JAR=${HAMA_HOME}/hama-core-${HAMA_VERSION}.jar
-MRQL_CLASSPATH=classes:${HAMA_JAR}:${SPARK_JARS}:${HADOOP_JARS}
+SPARK_JARS=${SCALA_LIB}/scala-library.jar:${SCALA_LIB}/scala-compiler.jar:${SPARK_HOME}/core/target/scala-2.9.3/classes:${SPARK_HOME}/lib_managed/jars/*:${SPARK_HOME}/lib_managed/bundles/*
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/lib/jline-1.0.jar
----------------------------------------------------------------------
diff --git a/lib/jline-1.0.jar b/lib/jline-1.0.jar
deleted file mode 100644
index 6c949b2..0000000
Binary files a/lib/jline-1.0.jar and /dev/null differ
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 9becab9..757d232 100644
--- a/pom.xml
+++ b/pom.xml
@@ -18,58 +18,78 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
-
- <parent>
- <groupId>org.apache</groupId>
- <artifactId>apache</artifactId>
- <version>7</version>
- </parent>
<groupId>org.apache.mrql</groupId>
- <artifactId>mrql</artifactId>
- <version>0.1.0-incubating</version>
+ <artifactId>MRQL</artifactId>
+ <version>0.9.0-incubating</version>
<packaging>pom</packaging>
-
<name>MRQL</name>
- <description>The development of packaging and tests of the Apache Hadoop ecosystem</description>
+ <description>MRQL is a query processing and optimization system for large-scale, distributed data analysis, built on top of Apache Hadoop, Hama, and Spark</description>
<url>http://mrql.incubator.apache.org/</url>
<inceptionYear>2013</inceptionYear>
+ <licenses>
+ <license>
+ <name>Apache 2</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <properties>
+ <hadoop.version>1.0.3</hadoop.version>
+ <hama.version>0.6.2</hama.version>
+ <spark.version>0.7.3</spark.version>
+ </properties>
+
+ <modules>
+ <module>Gen</module>
+ <module>MapReduce</module>
+ <module>BSP</module>
+ <module>Spark</module>
+ </modules>
+
+ <dependencies>
+ <dependency>
+ <groupId>net.sf.squirrel-sql.thirdparty.non-maven</groupId>
+ <artifactId>java-cup</artifactId>
+ <version>11a</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-core</artifactId>
+ <version>${hadoop.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ <version>1.2.17</version>
+ </dependency>
+ <dependency>
+ <groupId>jline</groupId>
+ <artifactId>jline</artifactId>
+ <version>1.0</version>
+ </dependency>
+ </dependencies>
<build>
- <extensions>
- <extension>
- <groupId>org.apache.maven.wagon</groupId>
- <artifactId>wagon-ssh-external</artifactId>
- <version>1.0-beta-6</version>
- </extension>
- </extensions>
- <pluginManagement>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-pdf-plugin</artifactId>
- <version>1.1</version>
- <dependencies>
- <dependency>
- <groupId>org.apache.maven.doxia</groupId>
- <artifactId>doxia-module-confluence</artifactId>
- <version>1.1.3</version>
- </dependency>
- </dependencies>
- </plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-site-plugin</artifactId>
- <version>2.1.1</version>
- </plugin>
- </plugins>
- </pluginManagement>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>exec-maven-plugin</artifactId>
+ <version>1.2.1</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-site-plugin</artifactId>
+ <version>2.1.1</version>
+ </plugin>
+ </plugins>
</build>
<issueManagement>
<system>JIRA</system>
<url>http://issues.apache.org/jira/browse/MRQL</url>
</issueManagement>
+
<mailingLists>
<mailingList>
<name>Development</name>
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/AlgebraicOptimization.gen
----------------------------------------------------------------------
diff --git a/src/AlgebraicOptimization.gen b/src/AlgebraicOptimization.gen
deleted file mode 100644
index 0b1a100..0000000
--- a/src/AlgebraicOptimization.gen
+++ /dev/null
@@ -1,315 +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 org.apache.mrql.gen.*;
-
-/**
- * Optimize MRQL algebra expressions using normalization rules and heuristic optimizations
- */
-public class AlgebraicOptimization extends Simplification {
-
- /** Is this a group-by operation?*/
- private static boolean is_groupBy ( Tree e ) {
- match e {
- case groupBy(...): return true;
- case orderBy(...): return true;
- };
- return false;
- }
-
- /**
- * algebraic optimization (algebra to algebra)
- * @param e the algebraic form to be optimized
- * @returns the optimized form
- */
- public static Tree translate ( Tree e ) {
- match e {
- // if the group-by key is the same as the join key, fuse the group-by into the join
- case mapReduce2(lambda(`v,bag(tuple(nth(`v1,0),`mx))),`my,
- lambda(`rv,`rb),
- `groupBy(cmap(lambda(`w,`m),`X)),`Y,`o):
- if (!v1.equals(v) || !Config.groupJoinOpt || ! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- Tree nv = new_var();
- Tree nr = subst(#<nth(`rv,0)>,#<cmap(lambda(`v,bag(`mx)),groupBy(nth(`rv,0)))>,rb);
- Tree res = #<mapReduce2(lambda(`w,cmap(lambda(`nv,bag(tuple(nth(`nv,0),`nv))),`m)),
- `my,lambda(`rv,`nr),`X,`Y,`o)>;
- res = simplify_all(rename(res));
- TypeInference.type_inference(res);
- return translate(res);
- // same for the right join input
- case mapReduce2(`mx,lambda(`v,bag(tuple(nth(`v1,0),`my))),
- lambda(`rv,`rb),
- `X,`groupBy(cmap(lambda(`w,`m),`Y)),`o):
- if (!v1.equals(v) || !Config.groupJoinOpt || ! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- Tree nv = new_var();
- Tree nr = subst(#<nth(`rv,1)>,#<cmap(lambda(`v,bag(`my)),groupBy(nth(`rv,1)))>,rb);
- Tree res = #<mapReduce2(`mx,lambda(`w,cmap(lambda(`nv,bag(tuple(nth(`nv,0),`nv))),`m)),
- lambda(`rv,`nr),`X,`Y,`o)>;
- res = simplify_all(rename(res));
- TypeInference.type_inference(res);
- return translate(res);
- // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway
- case mapReduce2(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`X,`Y,`o):
- if (!alpha_equivalent(X,Y) || !Config.selfJoinOpt)
- fail;
- Tree ny = subst(vy,vx,by);
- Tree tx = null;
- Tree ty = null;
- match TypeInference.type_inference(bx) {
- case _(tuple(_,`t)): tx = t;
- };
- match TypeInference.type_inference(by) {
- case _(tuple(_,`t)): ty = t;
- };
- Tree b = subst(s,#<tuple(cmap(lambda(tuple(n,v1,v2),
- if(call(eq,n,1),bag(v1),bag())),s),
- cmap(lambda(tuple(n,v1,v2),
- if(call(eq,n,2),bag(v2),bag())),s))>,
- f);
- Tree res = #<mapReduce(lambda(`vx,call(plus,
- cmap(lambda(x,bag(tuple(nth(x,0),tuple(1,nth(x,1),typed(null,`ty))))),`bx),
- cmap(lambda(y,bag(tuple(nth(y,0),tuple(2,typed(null,`tx),nth(y,1))))),`ny))),
- lambda(tuple(k,s),`b),
- `X,`o)>;
- res = simplify_all(rename(res));
- TypeInference.type_inference(res);
- return translate(res);
- case mapReduce2(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y,`o):
- match X {
- case groupBy(...): ;
- case _: return translate(#<mapReduce2(lambda(`v,cmap(`mx,`b)),`my,`r,`X,`Y,`o)>);
- };
- fail
- case mapReduce2(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y),`o):
- match Y {
- case groupBy(...): ;
- case _: return translate(#<mapReduce2(`mx,lambda(`v,cmap(`my,`b)),`r,`X,`Y,`o)>);
- };
- fail
- case crossProduct(`mx,`my,`r,cmap(lambda(`vx,`bx),`X),cmap(lambda(`vy,`by),`Y)):
- return translate(#<crossProduct(lambda(`vx,cmap(`mx,`bx)),
- lambda(`vy,cmap(`my,`by)),`r,`X,`Y)>);
- case crossProduct(`mx,`my,`r,cmap(lambda(`v,`b),`X),`Y):
- return translate(#<crossProduct(lambda(`v,cmap(`mx,`b)),`my,`r,`X,`Y)>);
- case crossProduct(`mx,`my,`r,`X,cmap(lambda(`v,`b),`Y)):
- return translate(#<crossProduct(`mx,lambda(`v,cmap(`my,`b)),`r,`X,`Y)>);
- case cmap(`m,crossProduct(`mx,`my,lambda(`v,`b),`X,`Y)):
- return translate(#<crossProduct(`mx,`my,lambda(`v,cmap(`m,`b)),`X,`Y)>);
- case cmap(`r,`groupBy1(cmap(`m,`groupBy2(`s)))):
- if (! #[groupBy,orderBy].member(#<`groupBy1>)
- && ! #[groupBy,orderBy].member(#<`groupBy2>))
- fail;
- return #<mapReduce(`(identity()),
- `(translate(r)),
- `(translate(#<cmap(`m,`groupBy2(`s))>)),
- `((#<`groupBy1>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case cmap(`r,`groupBy(cmap(`m,`s))):
- if (! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- return #<mapReduce(`(translate(m)),
- `(translate(r)),
- `(translate(s)),
- `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case `groupBy(cmap(`m,groupBy(`s))):
- if (! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- return #<mapReduce(`(identity()),
- `(identity()),
- `(translate(#<cmap(`m,groupBy(`s))>)),
- `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case `groupBy(cmap(`m,`s)):
- if (! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- return #<mapReduce(`(translate(m)),
- `(identity()),
- `(translate(s)),
- `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case cmap(`r,`groupBy(`s)):
- if (! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- return #<mapReduce(`(identity()),
- `(translate(r)),
- `(translate(s)),
- `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case `groupBy(`s):
- if (! #[groupBy,orderBy].member(#<`groupBy>))
- fail;
- return #<mapReduce(`(identity()),
- `(identity()),
- `(translate(s)),
- `((#<`groupBy>.equals(#<orderBy>)) ? #<true> : #<false>))>;
- case cmap(`m,`s):
- return #<cmap(`(translate(m)),
- `(translate(s)))>;
- // convert self-joins to single-source mapreduce; Hadoop doesn't like self-joins anyway
- case join(lambda(`vx,`bx),lambda(`vy,`by),lambda(`s,`f),`x,`y):
- if (!x.equals(y) || !Config.selfJoinOpt)
- fail;
- Tree ny = subst(vy,vx,by);
- Tree b = subst(s,
- #<tuple(cmap(lambda(tuple(n,v),
- if(call(eq,n,1),bag(v),bag())),s),
- cmap(lambda(tuple(n,v),
- if(call(eq,n,2),bag(v),bag())),s))>,
- f);
- Tree res = #<mapReduce(lambda(`vx,bag(tuple(`bx,tuple(1,`vx)),
- tuple(`ny,tuple(2,`vx)))),
- lambda(tuple(k,s),`b),`x,false)>;
- res = simplify_all(rename(res));
- TypeInference.type_inference(res);
- return translate(res);
- case join(lambda(`vx,`bx),lambda(`vy,`by),`f,`x,`y):
- return translate(#<mapReduce2(lambda(`vx,bag(tuple(`bx,`vx))),
- lambda(`vy,bag(tuple(`by,`vy))),
- `f,`x,`y,false)>);
- case nth(`x,`n):
- match TypeInference.type_inference2(x) {
- case `S(tuple(...bl)):
- if (!is_collection(S))
- fail;
- Tree nv = new_var();
- type_env.insert(nv.toString(),bl.nth((int)n.longValue()));
- return translate(#<cmap(lambda(`nv,`S(nth(`nv,`n))),`x)>);
- };
- fail
- case project(`x,`a):
- match TypeInference.type_inference2(x) {
- case `S(record(...bl)):
- if (!is_collection(S))
- fail;
- for ( Tree b: bl )
- match b {
- case bind(`c,_):
- if (!a.equals(c))
- fail;
- Tree nv = new_var();
- type_env.insert(nv.toString(),#<record(...bl)>);
- return translate(#<cmap(lambda(`nv,`S(project(`nv,`a))),`x)>);
- };
- };
- fail
- case `f(...al):
- Trees bl = #[];
- for ( Tree a: al )
- bl = bl.append(translate(a));
- return #<`f(...bl)>;
- };
- return e;
- }
-
- /** apply algebraic optimizations multiple times until no change */
- public static Tree translate_all ( Tree e ) {
- Tree ne = translate(e);
- if (e.equals(ne))
- return e;
- else return translate_all(ne);
- }
-
- /**
- * does a form contain a bulk plan that doesn't refer to a given variable?
- * @param e the form
- * @param v the given variable
- * @return true if e contains a bulk plan that doesn't refer to v
- */
- private static boolean contains_plan ( Tree e, Tree v ) {
- match e {
- case lambda(`x,`u): return false;
- case let(...): return false;
- case Let(...): return false;
- case `f(...as):
- if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v))
- return true;
- for (Tree a: as)
- if (contains_plan(a,v))
- return true;
- return false;
- };
- return false;
- }
-
- /**
- * extract the common factors (common sub-expressions) from a form
- * that do not refer to a given variable
- * @param e the form
- * @param v the given variable
- * @return the list of common factors
- */
- private static Trees common_factors ( Tree e, Tree v ) {
- match e {
- case lambda(`x,`u): return #[];
- case let(...): return #[];
- case Let(...): return #[];
- case `f(...as):
- if (!contains_plan(e,v))
- fail;
- if (plan_names.member(#<`f>) && !free_variables(e,#[]).member(v))
- return #[`e];
- Trees bs = #[];
- for ( Tree a: as )
- bs = bs.append(common_factors(a,v));
- return bs;
- };
- return #[];
- }
-
- /**
- * if a term is used multiple times in a query, factor it out using let-expressions
- * @param e the expression to be factored-out
- * @return the factored-out expression
- */
- public static Tree common_factoring ( Tree e ) {
- match e {
- case `f(...as):
- if (!plan_names.member(#<`f>))
- fail;
- Trees bs = #[];
- Trees binds = #[];
- for ( Tree a: as )
- match a {
- case lambda(`v,`u):
- if (!contains_plan(u,v))
- fail;
- Trees gs = common_factors(u,v);
- Tree nb = u;
- for ( Tree g: gs) {
- Tree nv = new_var();
- nb = subst(g,nv,nb);
- binds = binds.append(#<bind(`nv,`g)>);
- };
- bs = bs.append(#<lambda(`v,`(common_factoring(nb)))>);
- case _: bs = bs.append(common_factoring(a));
- };
- Tree res = #<`f(...bs)>;
- for ( Tree bind: binds )
- match bind {
- case bind(`v,`x):
- res = #<let(`v,`x,`res)>;
- };
- return res;
- case `f(...as):
- Trees bs = #[];
- for ( Tree a: as )
- bs = bs.append(common_factoring(a));
- return #<`f(...bs)>;
- };
- return e;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/BSPPlan.java
----------------------------------------------------------------------
diff --git a/src/BSP/BSPPlan.java b/src/BSP/BSPPlan.java
deleted file mode 100644
index 7252628..0000000
--- a/src/BSP/BSPPlan.java
+++ /dev/null
@@ -1,501 +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 org.apache.mrql.gen.*;
-import java.io.*;
-import java.util.Arrays;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hama.bsp.*;
-import org.apache.hama.bsp.sync.SyncException;
-import org.apache.hama.HamaConfiguration;
-import org.apache.hadoop.conf.Configuration;
-
-
-/** Evaluate a BSP plan using Hama */
-final public class BSPPlan extends Plan {
-
- final static Configuration getConfiguration ( BSPJob job ) {
- return job.getConf(); // use job.getConfiguration() for Hama 0.6.0
- }
-
- /** The BSP evaluator */
- final static class BSPop extends BSP<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> {
- final static MRContainer null_key = new MRContainer(new MR_byte(0));
- // special message for sub-sync()
- final static MRData more_to_come = new MR_sync();
- final static MRData more_supersteps = new MR_more_bsp_steps();
-
- private int source_num;
- private Function superstep_fnc; // superstep function
- private MRData state; // BSP state
- private boolean orderp; // will output be ordered?
- private MRData source; // BSP input
- private Function acc_fnc; // aggregator
- private MRData acc_result; // aggregation result
- private static String[] all_peer_names; // all BSP peer names
- private static BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer>[] all_peers; // all BSP peers
- // a master peer that coordinates and collects results of partial aggregations
- private String masterTask;
- // buffer for received messages -- regularly in a vector, but can be spilled in a local file
- Bag msg_cache;
- // the cache that holds all local data in memory
- Tuple local_cache;
-
- private static BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> getPeer ( String name ) {
- for ( int i = 0; i < all_peer_names.length; i++ )
- if (all_peer_names[i].equals(name))
- return all_peers[i];
- throw new Error("Unknown peer: "+name);
- }
-
- private static void setPeer ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) {
- String name = peer.getPeerName();
- for ( int i = 0; i < all_peer_names.length; i++ )
- if (all_peer_names[i].equals(name))
- all_peers[i] = peer;
- }
-
- /** shuffle values to BSP peers based on uniform hashing on key */
- private static String shuffle ( MRData key ) {
- return all_peer_names[Math.abs(key.hashCode()) % all_peer_names.length];
- }
-
- /** to exit a BSP loop, all peers must agree to exit (this is used in BSPTranslate.bspSimplify) */
- public static MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
- return synchronize(getPeer(peerName.get()),mr_exit);
- }
-
- /** to exit a BSP loop, all peers must agree to exit (this is used in BSPTranslate.bspSimplify) */
- public static MR_bool synchronize ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer, MR_bool mr_exit ) {
- if (!Config.hadoop_mode)
- return mr_exit;
- // shortcut: if we know for sure that all peers want to exit/continue, we don't have to poll
- if (mr_exit == SystemFunctions.bsp_true_value // must be ==, not equals
- || mr_exit == SystemFunctions.bsp_false_value)
- return mr_exit;
- try {
- // this case is only used for checking the exit condition of repeat/closure
- boolean exit = mr_exit.get();
- if (all_peer_names.length <= 1)
- return (exit) ? SystemFunctions.bsp_true_value : SystemFunctions.bsp_false_value;
- if (!exit)
- // this peer is not ready to exit, so no peer should exit
- for ( String p: peer.getAllPeerNames() )
- peer.send(p,new MRContainer(more_supersteps));
- peer.sync();
- // now exit is true if no peer sent a "more_supersteps" message
- exit = peer.getNumCurrentMessages() == 0;
- peer.clear();
- return (exit) ? SystemFunctions.bsp_true_value : SystemFunctions.bsp_false_value;
- } catch (Exception ex) {
- throw new Error(ex);
- }
- }
-
- /** collect a bag from all peers by distributing the local copy s */
- public static Bag distribute ( MR_string peerName, Bag s ) {
- BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer = getPeer(peerName.get());
- if (!Config.hadoop_mode)
- return s;
- try {
- for ( MRData e: s )
- for ( String p: all_peer_names )
- peer.send(p,new MRContainer(e));
- peer.sync();
- MRContainer msg;
- Bag res = new Bag();
- while ((msg = peer.getCurrentMessage()) != null)
- if (!res.contains(msg.data()))
- res.add(msg.data());
- peer.clear();
- return res;
- } catch (Exception ex) {
- throw new Error(ex);
- }
- }
-
- private void readLocalSources ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
- throws IOException {
- MRContainer key = new MRContainer();
- MRContainer value = new MRContainer();
- while (peer.readNext(key,value)) {
- Tuple p = (Tuple)(value.data());
- ((Bag)local_cache.get(((MR_int)p.first()).get())).add(p.second());
- }
- }
-
- private void writeLocalResult ( Bag result,
- BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
- throws IOException {
- MRContainer key = new MRContainer();
- MRContainer value = new MRContainer();
- for ( MRData v: result )
- if (orderp) { // prepare for sorting
- Tuple t = (Tuple)v;
- key.set(t.get(1));
- value.set(t.get(0));
- peer.write(key,value);
- } else {
- value.set(v);
- peer.write(null_key,value);
- }
- }
-
- /** receive messages from other peers */
- private void receive_messages ( final BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
- throws IOException, SyncException, InterruptedException {
- if (Config.bsp_msg_size <= 0) { // no buffering
- msg_cache = new Bag(new BagIterator() {
- MRContainer msg;
- public boolean hasNext () {
- try {
- return (msg = peer.getCurrentMessage()) != null;
- } catch (Exception ex) {
- throw new Error(ex);
- }
- }
- public MRData next () {
- return msg.data();
- }
- });
- } else {
- boolean expect_more = false; // are we expecting more incoming messages?
- do {
- // just in case this peer did a regular-sync() before the others did a sub-sync()
- expect_more = false;
- MRContainer msg;
- // cache the received messages
- while ((msg = peer.getCurrentMessage()) != null)
- // if at least one peer sends a more_to_come message, then expect_more
- if (msg.data().equals(more_to_come))
- expect_more = true;
- else msg_cache.add(msg.data());
- if (expect_more)
- peer.sync(); // sub-sync()
- } while (expect_more);
- }
- }
-
- /** send the messages produced by a superstep to peers and then receive the replies */
- private void send_messages ( Bag msgs,
- BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
- throws IOException, SyncException, InterruptedException {
- int size = 0;
- if (Config.bsp_msg_size > 0)
- msg_cache.clear();
- for ( MRData m: msgs ) {
- Tuple t = (Tuple)m;
- // if there are too many messages to send, then sub-sync()
- if ( Config.bsp_msg_size > 0 && size++ > Config.bsp_msg_size ) {
- // tell all peers that there is more to come after sync
- for ( String p: all_peer_names )
- if (!peer.getPeerName().equals(p))
- peer.send(p,new MRContainer(more_to_come));
- peer.sync(); // sub-sync()
- size = 0;
- MRContainer msg;
- // cache the received messages
- while ((msg = peer.getCurrentMessage()) != null)
- if (!msg.data().equals(more_to_come))
- msg_cache.add(msg.data());
- };
- // suffle messages based on key; needs new MRContainer object
- peer.send(shuffle(t.get(0)),new MRContainer(t.get(1)));
- };
- peer.sync(); // regular-sync()
- receive_messages(peer);
- }
-
- @Override
- public void bsp ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer )
- throws IOException, SyncException, InterruptedException {
- final Tuple stepin = new Tuple(4);
- stepin.set(3,new MR_string(peer.getPeerName()));
- Tuple result;
- boolean skip = false;
- String tabs = "";
- int step = 0;
- boolean exit;
- readLocalSources(peer);
- setPeer(peer);
- do {
- if (!skip)
- step++;
- if (!skip && Config.trace_execution) {
- tabs = Interpreter.tabs(Interpreter.tab_count);
- System.err.println(tabs+" Superstep "+step+" ["+peer.getPeerName()+"]:");
- System.err.println(tabs+" messages ["+peer.getPeerName()+"]: "+msg_cache);
- System.err.println(tabs+" state ["+peer.getPeerName()+"]: "+state);
- for ( int i = 0; i < local_cache.size(); i++)
- if (local_cache.get(i) instanceof Bag && ((Bag)local_cache.get(i)).size() > 0)
- System.out.println(tabs+" cache ["+peer.getPeerName()+"] "+i+": "+local_cache.get(i));
- };
- stepin.set(0,local_cache);
- stepin.set(1,msg_cache);
- stepin.set(2,state);
- // evaluate one superstep
- result = (Tuple)superstep_fnc.eval(stepin);
- Bag msgs = (Bag)result.get(0);
- exit = ((MR_bool)result.get(2)).get();
- state = result.get(1);
- // shortcuts: if we know for sure that all peers want to exit/continue
- if (result.get(2) == SystemFunctions.bsp_true_value) { // must be ==, not equals
- peer.sync();
- if (Config.trace_execution)
- System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
- break;
- };
- if (result.get(2) == SystemFunctions.bsp_false_value) {
- if (Config.trace_execution)
- System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
- send_messages(msgs,peer);
- skip = false;
- continue;
- };
- // shortcut: skip is true when NONE of the peers sent any messages
- skip = (msgs == SystemFunctions.bsp_empty_bag); // must be ==, not equals
- if (skip)
- continue;
- if (Config.trace_execution)
- System.err.println(tabs+" result ["+peer.getPeerName()+"]: "+result);
- exit = synchronize(peer,(MR_bool)result.get(2)).get();
- send_messages(msgs,peer);
- } while (!exit);
- if (acc_result == null) {
- // the BSP result is a bag that needs to be dumped to the HDFS
- writeLocalResult((Bag)(local_cache.get(source_num)),peer);
- // if there more results, dump them to HDFS
- final MR_long key = new MR_long(0);
- final MRContainer key_container = new MRContainer(key);
- final MRContainer data_container = new MRContainer(new MR_int(0));
- int loc = 0;
- while ( loc < all_peer_names.length && peer.getPeerName().equals(all_peer_names[loc]) )
- loc++;
- Configuration conf = peer.getConfiguration();
- String[] out_paths = conf.get("mrql.output.paths").split(",");
- for ( int i = 1; i < out_paths.length; i++ ) {
- String[] s = out_paths[i].split(":");
- int out_num = Integer.parseInt(s[0]);
- Path path = new Path(s[1]+"/peer"+loc);
- FileSystem fs = path.getFileSystem(conf);
- SequenceFile.Writer writer
- = new SequenceFile.Writer(fs,conf,path,
- MRContainer.class,MRContainer.class);
- long count = 0;
- for ( MRData e: (Bag)(local_cache.get(out_num)) ) {
- key.set(count++);
- data_container.set(e);
- writer.append(key_container,data_container);
- };
- writer.close();
- };
- } else {
- // the BSP result is an aggregation:
- // send the partial results to the master peer
- peer.send(masterTask,new MRContainer(local_cache.get(source_num)));
- peer.sync();
- if (peer.getPeerName().equals(masterTask)) {
- // only the master peer collects the partial aggregations
- MRContainer msg;
- while ((msg = peer.getCurrentMessage()) != null)
- acc_result = acc_fnc.eval(new Tuple(acc_result,msg.data()));
- // write the final aggregation result
- peer.write(null_key,new MRContainer(acc_result));
- }
- }
- }
-
- @Override
- @SuppressWarnings("unchecked")
- public void setup ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) {
- try {
- super.setup(peer);
- Configuration conf = peer.getConfiguration();
- Config.read(conf);
- if (Plan.conf == null)
- Plan.conf = conf;
- all_peer_names = peer.getAllPeerNames();
- all_peers = new BSPPeerImpl[all_peer_names.length];
- Arrays.sort(all_peer_names); // is this necessary?
- source_num = conf.getInt("mrql.output.tag",0);
- Tree code = Tree.parse(conf.get("mrql.superstep"));
- superstep_fnc = functional_argument(conf,code);
- code = Tree.parse(conf.get("mrql.initial.state"));
- state = Interpreter.evalE(code);
- if (conf.get("mrql.zero") != null && !conf.get("mrql.zero").equals("")) {
- code = Tree.parse(conf.get("mrql.zero"));
- acc_result = Interpreter.evalE(code);
- code = Tree.parse(conf.get("mrql.accumulator"));
- acc_fnc = functional_argument(conf,code);
- } else acc_result = null;
- orderp = conf.getBoolean("mrql.orderp",false);
- masterTask = all_peer_names[peer.getNumPeers()/2];
- msg_cache = new Bag(1000);
- local_cache = new Tuple(max_input_files);
- for ( int i = 0; i < max_input_files; i++ )
- local_cache.set(i,new Bag());
- } catch (Exception e) {
- e.printStackTrace();
- throw new Error("Cannot setup the Hama BSP job: "+e);
- }
- }
-
- @Override
- public void cleanup ( BSPPeer<MRContainer,MRContainer,MRContainer,MRContainer,MRContainer> peer ) throws IOException {
- if (!Config.local_mode)
- clean();
- local_cache = null;
- super.cleanup(peer);
- }
- }
-
- /** set Hama's min split size and number of BSP tasks (doesn't work with Hama 0.6.0) */
- public static void setupSplits ( BSPJob job, DataSet ds ) throws IOException {
- long[] sizes = new long[ds.source.size()];
- if (sizes.length > Config.nodes)
- throw new Error("Cannot distribute "+sizes.length+" files over "+Config.nodes+" BSP tasks");
- for ( int i = 0; i < sizes.length; i++ )
- sizes[i] = ds.source.get(i).size(Plan.conf);
- long total_size = 0;
- for ( long size: sizes )
- total_size += size;
- long split_size = Math.max(total_size/Config.nodes,100000);
- int tasks = 0;
- do { // adjust split_size
- tasks = 0;
- for ( long size: sizes )
- tasks += (int)Math.ceil(size/(double)split_size);
- if (tasks > Config.nodes)
- split_size = (long)Math.ceil((double)split_size*1.01);
- } while (tasks > Config.nodes);
- job.setNumBspTask(tasks);
- System.err.println("*** Using "+tasks+" BSP tasks (out of a max "+Config.nodes+")."
- +" Each task will handle about "+Math.min(total_size/Config.nodes,split_size)
- +" bytes of input data.");
- job.set("bsp.min.split.size",Long.toString(split_size));
- }
-
- /** Evaluate a BSP operation that returns a DataSet
- * @param source_nums output tags
- * @param superstep the superstep function
- * @param init_state initial state
- * @param orderp do we need to order the result?
- * @param source input dataset
- * @return a new data source that contains the result
- */
- public final static MRData BSP ( int[] source_nums, // output tags
- Tree superstep, // superstep function
- Tree init_state, // initial state
- boolean orderp, // do we need to order the result?
- DataSet source // input dataset
- ) throws Exception {
- String[] newpaths = new String[source_nums.length];
- newpaths[0] = new_path(conf);
- conf.set("mrql.output.paths",source_nums[0]+":"+newpaths[0]);
- for ( int i = 1; i < source_nums.length; i++ ) {
- newpaths[i] = new_path(conf);
- Path path = new Path(newpaths[1]);
- FileSystem fs = path.getFileSystem(conf);
- fs.mkdirs(path);
- conf.set("mrql.output.paths",conf.get("mrql.output.paths")+","+source_nums[i]+":"+newpaths[i]);
- };
- conf.set("mrql.superstep",superstep.toString());
- conf.set("mrql.initial.state",init_state.toString());
- conf.set("mrql.zero","");
- conf.setInt("mrql.output.tag",source_nums[0]);
- conf.setBoolean("mrql.orderp",orderp);
- BSPJob job = new BSPJob((HamaConfiguration)conf,BSPop.class);
- setupSplits(job,source);
- job.setJobName(newpaths[0]);
- distribute_compiled_arguments(getConfiguration(job));
- job.setBspClass(BSPop.class);
- Path outpath = new Path(newpaths[0]);
- job.setOutputPath(outpath);
- job.setOutputKeyClass(MRContainer.class);
- job.setOutputValueClass(MRContainer.class);
- job.setOutputFormat(SequenceFileOutputFormat.class);
- job.setInputFormat(MultipleBSPInput.class);
- FileInputFormat.setInputPaths(job,source.merge());
- job.waitForCompletion(true);
- if (source_nums.length == 1) {
- BinaryDataSource ds = new BinaryDataSource(source_nums[0],newpaths[0],conf);
- ds.to_be_merged = orderp;
- return new MR_dataset(new DataSet(ds,0,3));
- } else {
- MRData[] s = new MRData[source_nums.length];
- for ( int i = 0; i < source_nums.length; i++ ) {
- BinaryDataSource ds = new BinaryDataSource(source_nums[i],newpaths[i],conf);
- ds.to_be_merged = orderp;
- s[i] = new MR_dataset(new DataSet(ds,0,3));
- };
- return new Tuple(s);
- }
- }
-
- /** Evaluate a BSP operation that aggregates the results
- * @param source_nums output tags
- * @param superstep the superstep function
- * @param init_state initial state
- * @param acc_fnc accumulator function
- * @param zero zero value for the accumulator
- * @param source input dataset
- * @return the aggregation result
- */
- public final static MRData BSPaggregate ( int source_num, // output tag
- Tree superstep, // superstep function
- Tree init_state, // initial state
- Tree acc_fnc, // accumulator function
- Tree zero, // zero value for the accumulator
- DataSet source // input dataset
- ) throws Exception {
- String newpath = new_path(conf);
- conf.set("mrql.superstep",superstep.toString());
- conf.set("mrql.initial.state",init_state.toString());
- conf.set("mrql.accumulator",acc_fnc.toString());
- conf.set("mrql.zero",zero.toString());
- conf.setInt("mrql.output.tag",source_num);
- conf.setBoolean("mrql.orderp",false);
- BSPJob job = new BSPJob((HamaConfiguration)conf,BSPop.class);
- setupSplits(job,source);
- job.setJobName(newpath);
- distribute_compiled_arguments(getConfiguration(job));
- job.setBspClass(BSPop.class);
- Path outpath = new Path(newpath);
- job.setOutputPath(outpath);
- job.setOutputKeyClass(MRContainer.class);
- job.setOutputValueClass(MRContainer.class);
- job.setOutputFormat(SequenceFileOutputFormat.class);
- job.setInputFormat(MultipleBSPInput.class);
- FileInputFormat.setInputPaths(job,source.merge());
- job.waitForCompletion(true);
- FileSystem fs = outpath.getFileSystem(conf);
- FileStatus[] files = fs.listStatus(outpath);
- for ( int i = 0; i < files.length; i++ ) {
- SequenceFile.Reader sreader = new SequenceFile.Reader(fs,files[i].getPath(),conf);
- MRContainer key = new MRContainer();
- MRContainer value = new MRContainer();
- sreader.next(key,value);
- sreader.close();
- if (value.data() != null)
- return value.data();
- };
- return null;
- }
- }
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/BinaryInputFormat.java
----------------------------------------------------------------------
diff --git a/src/BSP/BinaryInputFormat.java b/src/BSP/BinaryInputFormat.java
deleted file mode 100644
index 27f62ee..0000000
--- a/src/BSP/BinaryInputFormat.java
+++ /dev/null
@@ -1,60 +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 org.apache.mrql.gen.Trees;
-import java.io.*;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.bsp.*;
-import org.apache.hama.HamaConfiguration;
-
-
-/** Input format for hadoop sequence files */
-final public class BinaryInputFormat extends MRQLFileInputFormat {
- public static class BinaryInputRecordReader extends SequenceFileRecordReader<MRContainer,MRContainer> {
- final MRContainer result = new MRContainer();
- final MRData source_num_data;
- final int source_number;
-
- public BinaryInputRecordReader ( FileSplit split,
- BSPJob job,
- int source_number ) throws IOException {
- super(BSPPlan.getConfiguration(job),split);
- this.source_number = source_number;
- source_num_data = new MR_int(source_number);
- }
-
- @Override
- public synchronized boolean next ( MRContainer key, MRContainer value ) throws IOException {
- boolean b = super.next(key,result);
- value.set(new Tuple(source_num_data,result.data()));
- return b;
- }
- }
-
- public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split,
- BSPJob job ) throws IOException {
- Configuration conf = BSPPlan.getConfiguration(job);
- String path = ((FileSplit)split).getPath().toString();
- BinaryDataSource ds = (BinaryDataSource)DataSource.get(path,conf);
- return new BinaryInputRecordReader((FileSplit)split,job,ds.source_num);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/Evaluator.gen
----------------------------------------------------------------------
diff --git a/src/BSP/Evaluator.gen b/src/BSP/Evaluator.gen
deleted file mode 100644
index 11a166d..0000000
--- a/src/BSP/Evaluator.gen
+++ /dev/null
@@ -1,215 +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 org.apache.mrql.gen.*;
-import java.io.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.HamaConfiguration;
-
-
-/** Evaluate a physical plan in BSP mode using Hama */
-final public class Evaluator extends Interpreter {
-
- /** initialize the BSP evaluator */
- final static void init ( Configuration conf ) {
- Config.bsp_mode = true;
- if (Config.local_mode) {
- conf.set("bsp.master.address","local");
- conf.set("mapred.job.tracker","local");
- conf.set("hama.zookeeper.quorum","localhost");
- conf.setInt("bsp.local.tasks.maximum",Config.nodes);
- conf.set("fs.default.name","file:///");
- }
- }
-
- final static void initialize_query () {
- }
-
- /** shutdown the BSP evaluator */
- final static void shutdown ( Configuration conf ) {
- }
-
- final static Configuration new_configuration () {
- return new HamaConfiguration();
- }
-
- public static MR_bool synchronize ( MR_string peerName, MR_bool mr_exit ) {
- return BSPPlan.BSPop.synchronize(peerName,mr_exit);
- }
-
- public static Bag distribute ( MR_string peerName, Bag s ) {
- return BSPPlan.BSPop.distribute(peerName,s);
- }
-
- /** Evaluate MRQL physical operators using BSP
- * @param e the physical plan
- * @param env contains bindings fro variables to values (MRData)
- * @return a DataSet (stored in HDFS)
- */
- final static MRData bsp ( Tree plan, Environment env ) throws Exception {
- match plan {
- case BSP(tuple(...ns),`superstep,`init_state,`o,...S):
- int[] os = new int[ns.length()];
- for ( int i = 0; i < os.length; i++ )
- os[i] = (int)((LongLeaf)ns.nth(i)).value();
- DataSet ds = eval(S.head(),env);
- for ( Tree s: S.tail() )
- ds.merge(eval(s,env));
- return BSPPlan.BSP(os,
- closure(superstep,env),
- init_state,
- o.equals(#<true>),
- ds);
- case BSP(`n,`superstep,`init_state,`o,...S):
- DataSet ds = eval(S.head(),env);
- for ( Tree s: S.tail() )
- ds.merge(eval(s,env));
- return BSPPlan.BSP(new int[]{(int)((LongLeaf)n).value()},
- closure(superstep,env),
- init_state,
- o.equals(#<true>),
- ds);
- }
- throw new Error("Cannot perform the BSP plan: "+plan);
- }
-
- /** The Aggregate physical operator
- * @param acc_fnc the accumulator function from (T,T) to T
- * @param zero the zero element of type T
- * @param plan the plan that constructs the dataset that contains the bag of values {T}
- * @param env contains bindings fro variables to values (MRData)
- * @returns the aggregation result of type T
- */
- final static MRData aggregate ( Tree acc_fnc,
- Tree zero,
- Tree plan,
- Environment env ) throws Exception {
- match plan {
- case BSP(`n,`superstep,`init_state,`o,...S):
- DataSet ds = eval(S.head(),env,"-");
- for ( Tree s: S.tail() )
- ds.merge(eval(s,env,"-"));
- return BSPPlan.BSPaggregate((int)((LongLeaf)n).value(),
- closure(superstep,env),
- init_state,
- closure(acc_fnc,env),
- zero,
- ds);
- }
- throw new Error("Cannot perform the aggregation: "+plan);
- }
-
- final static Tuple loop ( Tree e, Environment env ) {
- throw new Error("The BSP Loop was supposed to be translated to a BSP task");
- }
-
- final static DataSet eval ( final Tree e,
- final Environment env,
- final String counter ) {
- return eval(e,env);
- }
-
- /** Evaluate MRQL BSP physical operators using Hama (returns a DataSet)
- * and print tracing info
- * @param e the physical plan
- * @param env contains bindings fro variables to values (MRData)
- * @return a DataSet (stored in HDFS)
- */
- final static DataSet eval ( final Tree e, final Environment env ) {
- if (Config.trace_execution) {
- tab_count += 3;
- System.out.println(tabs(tab_count)+print_query(e));
- };
- DataSet res = evalD(e,env);
- if (Config.trace_execution)
- try {
- System.out.println(tabs(tab_count)
- +"-> "+Plan.collect(res,false));
- tab_count -= 3;
- } catch (Exception ex) {
- throw new Error("Cannot collect the operator output: "+e);
- };
- return res;
- }
-
- /** Evaluate MRQL BSP physical operators using Hama (returns a DataSet)
- * @param e the physical plan
- * @param env contains bindings fro variables to values (MRData)
- * @return a DataSet (stored in HDFS)
- */
- final static DataSet evalD ( final Tree e, final Environment env ) {
- try {
- match e {
- case BSPSource(`n,BinarySource(`file,_)):
- return Plan.binarySource((int)((LongLeaf)n).value(),file.stringValue());
- case BSPSource(`n,ParsedSource(`parser,`file,...args)):
- Class<? extends Parser> p = DataSource.parserDirectory.get(parser.toString());
- if (p == null)
- throw new Error("Unknown parser: "+parser);
- return Plan.parsedSource((int)((LongLeaf)n).value(),
- p,((MR_string)evalE(file,env)).get(),args);
- case BSPSource(`n,Generator(`min,`max,`size)):
- return Plan.generator((int)((LongLeaf)n).value(),
- ((MR_long)evalE(min,env)).get(),
- ((MR_long)evalE(max,env)).get(),
- ((MR_long)evalE(size,env)).get());
- case Merge(`x,`y):
- return Plan.merge(eval(x,env),eval(y,env));
- case Dump(`s):
- return Plan.fileCache((Bag)evalE(s,env));
- case apply(`f,`arg):
- if (!f.is_variable())
- return ((MR_dataset)evalF(f,env).eval(evalE(arg))).dataset();
- MRData fnc = variable_lookup(f.toString(),global_env);
- if (fnc == null)
- throw new Error("Unknown function: "+f);
- MRData t = evalE(arg,env);
- if (!(t instanceof Tuple))
- throw new Error("Expected a tuple in function application: "+t);
- return ((MR_dataset)((Lambda)fnc).lambda().eval(t)).dataset();
- case `v:
- if (!v.is_variable())
- fail;
- MRData x = variable_lookup(v.toString(),env);
- if (x != null)
- if (x instanceof MR_dataset)
- return ((MR_dataset)x).dataset();
- x = variable_lookup(v.toString(),global_env);
- if (x != null)
- if (x instanceof MR_dataset)
- return ((MR_dataset)x).dataset();
- throw new Error("Variable "+v+" is not bound");
- };
- MRData d = bsp(e,env);
- if (d instanceof MR_dataset)
- return ((MR_dataset)d).dataset();
- throw new Error("Cannot evaluate the BSP plan: "+e);
- } catch (Error msg) {
- if (!Config.trace)
- throw new Error(msg.getMessage());
- System.err.println(msg.getMessage());
- throw new Error("Evaluation error in: "+print_query(e));
- } catch (Exception ex) {
- System.err.println(ex.getMessage());
- ex.printStackTrace();
- throw new Error("Evaluation error in: "+print_query(e));
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/GeneratorInputFormat.java
----------------------------------------------------------------------
diff --git a/src/BSP/GeneratorInputFormat.java b/src/BSP/GeneratorInputFormat.java
deleted file mode 100644
index cb13a83..0000000
--- a/src/BSP/GeneratorInputFormat.java
+++ /dev/null
@@ -1,89 +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 org.apache.mrql.gen.Trees;
-import java.io.*;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.bsp.*;
-
-
-/** the FileInputFormat for data generators: it creates HDFS files, where each file contains
- * an (offset,size) pair that generates the range of values [offset,offset+size] */
-final public class GeneratorInputFormat extends MRQLFileInputFormat {
- public static class GeneratorRecordReader implements RecordReader<MRContainer,MRContainer> {
- final long offset;
- final long size;
- final int source_number;
- final MRData source_num_data;
- long index;
- SequenceFile.Reader reader;
-
- public GeneratorRecordReader ( FileSplit split,
- int source_number,
- BSPJob job ) throws IOException {
- Configuration conf = BSPPlan.getConfiguration(job);
- Path path = split.getPath();
- FileSystem fs = path.getFileSystem(conf);
- reader = new SequenceFile.Reader(path.getFileSystem(conf),path,conf);
- MRContainer key = new MRContainer();
- MRContainer value = new MRContainer();
- reader.next(key,value);
- offset = ((MR_long)((Tuple)(value.data())).first()).get();
- size = ((MR_long)((Tuple)(value.data())).second()).get();
- this.source_number = source_number;
- source_num_data = new MR_int(source_number);
- index = -1;
- }
-
- public MRContainer createKey () {
- return new MRContainer(null);
- }
-
- public MRContainer createValue () {
- return new MRContainer(null);
- }
-
- public boolean next ( MRContainer key, MRContainer value ) throws IOException {
- index++;
- value.set(new Tuple(source_num_data,new MR_long(offset+index)));
- key.set(new MR_long(index));
- return index < size;
- }
-
- public long getPos () throws IOException { return index; }
-
- public void close () throws IOException { reader.close(); }
-
- public float getProgress () throws IOException {
- return index / (float)size;
- }
-
- public void initialize ( InputSplit split, TaskAttemptContext context ) throws IOException { }
- }
-
- public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split, BSPJob job ) throws IOException {
- Configuration conf = BSPPlan.getConfiguration(job);
- String path = ((FileSplit)split).getPath().toString();
- GeneratorDataSource ds = (GeneratorDataSource)DataSource.get(path,conf);
- return new GeneratorRecordReader((FileSplit)split,ds.source_num,job);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/MRQLFileInputFormat.java
----------------------------------------------------------------------
diff --git a/src/BSP/MRQLFileInputFormat.java b/src/BSP/MRQLFileInputFormat.java
deleted file mode 100644
index c7a7abf..0000000
--- a/src/BSP/MRQLFileInputFormat.java
+++ /dev/null
@@ -1,107 +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.*;
-import java.util.Iterator;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hama.bsp.*;
-import org.apache.hama.HamaConfiguration;
-
-
-/** A superclass for all MRQL FileInputFormats */
-abstract public class MRQLFileInputFormat extends FileInputFormat<MRContainer,MRContainer> {
- public MRQLFileInputFormat () {}
-
- /** record reader for map-reduce */
- abstract public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split, BSPJob job ) throws IOException;
-
- /** materialize the input file into a memory Bag */
- public Bag materialize ( final Path file ) throws Exception {
- final BSPJob job = new BSPJob((HamaConfiguration)Plan.conf,MRQLFileInputFormat.class);
- job.setInputPath(file);
- final InputSplit[] splits = getSplits(job,1);
- final RecordReader<MRContainer,MRContainer> rd = getRecordReader(splits[0],job);
- return new Bag(new BagIterator () {
- RecordReader<MRContainer,MRContainer> reader = rd;
- MRContainer key = reader.createKey();
- MRContainer value = reader.createKey();
- int i = 0;
- public boolean hasNext () {
- try {
- if (reader.next(key,value))
- return true;
- do {
- if (++i >= splits.length)
- return false;
- reader.close();
- reader = getRecordReader(splits[i],job);
- } while (!reader.next(key,value));
- return true;
- } catch (IOException e) {
- throw new Error("Cannot collect values from an intermediate result");
- }
- }
- public MRData next () {
- return value.data();
- }
- });
- }
-
- /** materialize the entire dataset into a Bag
- * @param x the DataSet in HDFS to collect values from
- * @param strip true if you want to stripout the source id (used in BSP sources)
- * @return the Bag that contains the collected values
- */
- public final static Bag collect ( final DataSet x, boolean strip ) throws Exception {
- Bag res = new Bag();
- for ( DataSource s: x.source )
- if (s.to_be_merged)
- res = res.union(Plan.merge(s));
- else {
- Path path = new Path(s.path);
- final FileSystem fs = path.getFileSystem(Plan.conf);
- final FileStatus[] ds
- = fs.listStatus(path,
- new PathFilter () {
- public boolean accept ( Path path ) {
- return !path.getName().startsWith("_");
- }
- });
- Bag b = new Bag();
- for ( FileStatus st: ds )
- b = b.union(s.inputFormat.newInstance().materialize(st.getPath()));
- if (strip) {
- // remove source_num
- final Iterator<MRData> iter = b.iterator();
- b = new Bag(new BagIterator() {
- public boolean hasNext () {
- return iter.hasNext();
- }
- public MRData next () {
- return ((Tuple)iter.next()).get(1);
- }
- });
- };
- res = res.union(b);
- };
- return res;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/MultipleBSPInput.java
----------------------------------------------------------------------
diff --git a/src/BSP/MultipleBSPInput.java b/src/BSP/MultipleBSPInput.java
deleted file mode 100644
index 67f08f9..0000000
--- a/src/BSP/MultipleBSPInput.java
+++ /dev/null
@@ -1,49 +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 org.apache.mrql.gen.Trees;
-import java.io.*;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.bsp.*;
-
-
-/** A FileInputFormat for multiple files, where each file may be associated with
- * a different FileInputFormat */
-final public class MultipleBSPInput extends MRQLFileInputFormat {
- public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split, BSPJob job ) throws IOException {
- String path = ((FileSplit)split).getPath().toString();
- Configuration conf = BSPPlan.getConfiguration(job);
- DataSource ds = DataSource.get(path,conf);
- Plan.conf = conf;
- if (ds instanceof ParsedDataSource)
- return new ParsedInputFormat.ParsedRecordReader((FileSplit)split,
- job,
- ((ParsedDataSource)ds).parser,
- ds.source_num,
- (Trees)((ParsedDataSource)ds).args);
- else if (ds instanceof BinaryDataSource)
- return new BinaryInputFormat.BinaryInputRecordReader((FileSplit)split,job,ds.source_num);
- else if (ds instanceof GeneratorDataSource)
- return new GeneratorInputFormat.GeneratorRecordReader((FileSplit)split,ds.source_num,job);
- else throw new Error("Unknown data source: "+ds+" for path "+path);
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-mrql/blob/f5e66d20/src/BSP/ParsedInputFormat.java
----------------------------------------------------------------------
diff --git a/src/BSP/ParsedInputFormat.java b/src/BSP/ParsedInputFormat.java
deleted file mode 100644
index d7bc3f8..0000000
--- a/src/BSP/ParsedInputFormat.java
+++ /dev/null
@@ -1,102 +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 org.apache.mrql.gen.Trees;
-import java.io.*;
-import java.util.Iterator;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hama.bsp.*;
-
-
-/** A FileInputFormat for text files (CVS, XML, JSON, ...) */
-final public class ParsedInputFormat extends MRQLFileInputFormat {
- public static class ParsedRecordReader implements RecordReader<MRContainer,MRContainer> {
- final FSDataInputStream fsin;
- final long start;
- final long end;
- final int source_number;
- final MRData source_num_data;
- Iterator<MRData> result;
- Parser parser;
-
- public ParsedRecordReader ( FileSplit split,
- BSPJob job,
- Class<? extends Parser> parser_class,
- int source_number,
- Trees args ) throws IOException {
- Configuration conf = BSPPlan.getConfiguration(job);
- start = split.getStart();
- end = start + split.getLength();
- Path file = split.getPath();
- FileSystem fs = file.getFileSystem(conf);
- fsin = fs.open(split.getPath());
- try {
- parser = parser_class.newInstance();
- } catch (Exception ex) {
- throw new Error("Unrecognized parser:"+parser_class);
- };
- this.source_number = source_number;
- source_num_data = new MR_int(source_number);
- parser.initialize(args);
- parser.open(fsin,start,end);
- result = null;
- }
-
- public MRContainer createKey () {
- return new MRContainer();
- }
-
- public MRContainer createValue () {
- return new MRContainer();
- }
-
- public synchronized boolean next ( MRContainer key, MRContainer value ) throws IOException {
- while (result == null || !result.hasNext()) {
- String s = parser.slice();
- if (s == null)
- return false;
- result = parser.parse(s).iterator();
- };
- value.set(new Tuple(source_num_data,(MRData)result.next()));
- key.set(new MR_long(fsin.getPos()));
- return true;
- }
-
- public synchronized long getPos () throws IOException { return fsin.getPos(); }
-
- public synchronized void close () throws IOException { fsin.close(); }
-
- public float getProgress () throws IOException {
- if (end == start)
- return 0.0f;
- else return Math.min(1.0f, (getPos() - start) / (float)(end - start));
- }
- }
-
- public RecordReader<MRContainer,MRContainer>
- getRecordReader ( InputSplit split,
- BSPJob job ) throws IOException {
- Configuration conf = BSPPlan.getConfiguration(job);
- String path = ((FileSplit)split).getPath().toString();
- ParsedDataSource ds = (ParsedDataSource)DataSource.get(path,conf);
- return new ParsedRecordReader((FileSplit)split,job,ds.parser,ds.source_num,(Trees)ds.args);
- }
-}