Add support for Hadoop 3.x and Flink Yarn sessions

README.md

@@ -35,6 +35,8 @@ Peel offers the following features for your experiments.
 | HDFS             | 2.7.2          | `hdfs-2.7.2`      |
 | HDFS             | 2.7.3          | `hdfs-2.7.3`      |
 | HDFS             | 2.8.0          | `hdfs-2.8.0`      |
+| HDFS             | 3.1.1          | `hdfs-3.1.1`      |
+| Yarn             | 3.1.1          | `yarn-3.1.1`      |
 | Flink            | 0.8.0          | `flink-0.8.0`     |
 | Flink            | 0.8.1          | `flink-0.8.1`     |
 | Flink            | 0.9.0          | `flink-0.9.0`     |
@@ -56,6 +58,10 @@ Peel offers the following features for your experiments.
 | Flink            | 1.3.1          | `flink-1.3.1`     |
 | Flink            | 1.3.2          | `flink-1.3.2`     |
 | Flink            | 1.4.0          | `flink-1.4.0`     |
+| Flink Standalone Cluster | 1.7.0          | `flink-1.7.0`     |
+| Flink Standalone Cluster | 1.7.2          | `flink-1.7.2`     |
+| Flink Yarn Session | 1.7.0          | `flink-yarn-1.7.0`     |
+| Flink Yarn Session | 1.7.2          | `flink-yarn-1.7.2`     |
 | MapReduce        | 1.2.1          | `mapred-1.2.1`    |
 | MapReduce        | 2.4.1          | `mapred-2.4.1`    |
 | Spark            | 1.3.1          | `spark-1.3.1`     |

peel-core/src/main/scala/org/peelframework/core/config/Model.scala

@@ -26,6 +26,7 @@ import scala.collection.mutable.ListBuffer
 trait Model {
 
   case class Pair(name: String, value: Any) {}
+  case class Section(name: String, entries: util.List[Pair]) {}
 
 }
 
@@ -76,6 +77,54 @@ object Model {
     }
   }
 
+  /** A model for INI type configuration files (e.g. etc/hadoop/container-executor.xml).
+   *
+   * Consists of a single entry `sections` which is constructed by collecting
+   * all direct children under the specified `prefix` path. Each child
+   * represents a section in the INI file and collects the (key, value) pairs
+   * below it.
+   *
+   * (key, value) pairs that should not appear in a section have to be
+   * listed under the child with a special name "_root_" (without quotes).
+   *
+   * See https://en.wikipedia.org/wiki/INI_file for details.
+   *
+   * @param c The HOCON config to use when constructing the model.
+   * @param prefix The prefix path which has to be rendered.
+   */
+  class INI(val c: Config, val prefix: String) extends Model {
+
+    val sections = {
+      val sectionBuffer = ListBuffer[Section]()
+
+      def sanitize(s: String) =
+        s.stripPrefix(s"$prefix.") // remove prefix
+
+      def fixRoot(s: String) = if (s == "_root_") null else s
+
+      def collectPairs(c: Config, name: String): Unit = {
+        val buffer = ListBuffer[Pair]()
+
+        for (e <- c.entrySet().asScala) {
+          val key = sanitize(e.getKey)
+            .replace("\"_root_\"", "_root_")
+            .stripPrefix(s"$name.")
+          buffer += Pair(key, c.getString(e.getKey))
+        }
+
+        sectionBuffer += Section(fixRoot(name), buffer.toList.asJava)
+      }
+
+      for (e <- c.getObject(prefix).entrySet().asScala) {
+        val name = sanitize(e.getKey)
+        collectPairs(c.withOnlyPath(s"$prefix.$name"), name)
+      }
+
+      sectionBuffer.toList.asJava
+    }
+
+  }
+
   /** A model for environment files (e.g., etc/hadoop/hadoop-env.sh).
     *
     * The children of the specified `prefix` path in the given `c` config are converted as (key, value) pairs in a
@@ -99,7 +148,7 @@ object Model {
     }
   }
 
-  /** A model for (key, value) based yaml files (e.g. conf/flink-conf.yaml).
+  /** A model for (key, value) based files (e.g. conf/flink-conf.yaml or etc/hadoop/capacity-manager.xml).
     *
     * Consists of multiple (key, value) entries which are constructed by collecting all values under
     * the specified `prefix` path. Intermediate paths are thereby flattened, i.e.
@@ -118,7 +167,7 @@ object Model {
     * @param c The HOCON config to use when constructing the model.
     * @param prefix The prefix path which has to be rendered.
     */
-  class Yaml(val c: Config, val prefix: String) extends util.HashMap[String, Object] with Model {
+  class KeyValue(val c: Config, val prefix: String) extends util.HashMap[String, Object] with Model {
 
     // constructor
     {
@@ -139,6 +188,11 @@ object Model {
     }
   }
 
+  /** Alias for Model.KeyValue.
+   *
+   */
+  class Yaml(override val c: Config, override val prefix: String) extends KeyValue(c, prefix) { }
+
   /** A model for list based hosts files (e.g. etc/hadoop/slaves).
     *
     * Consists of a single entry `hosts` which is constructed by extracting the string list value under the

peel-core/src/main/scala/org/peelframework/core/graph/DependencyGraph.scala

@@ -173,7 +173,7 @@ class DependencyGraph[T: ClassTag] {
       throw new Exception("Cannot reverse empty Graph!")
   }
 
-  /** Collects descendants in a depth-first manner starting from the given set.
+  /** Collects descendants in a breadth-first manner starting from the given set.
    *
    * @param toVisit A set of nodes that are yet to be visited.
    * @param visited A list of already visited nodes.
@@ -191,7 +191,7 @@ class DependencyGraph[T: ClassTag] {
         case x: Any => !visited.contains(x)
       }
 
-      collect(children ++ toVisit.tail, next :: visited, excluded)
+      collect(toVisit.tail ++ children, next :: visited, excluded)
     }
   }
 

peel-extensions/src/main/resources/reference.flink-1.7.0.conf

@@ -0,0 +1,19 @@
+# include common flink configuration
+include "reference.flink.conf"
+
+system {
+  flink {
+    path {
+      archive.url = "http://archive.apache.org/dist/flink/flink-1.7.0/flink-1.7.0-bin-hadoop28-scala_2.12.tgz"
+      archive.md5 = "b66459379703adb25c64fbe20bf8f4b1"
+      archive.src = ${app.path.downloads}"/flink-1.7.0-bin-hadoop28-scala_2.12.tgz"
+      home = ${system.flink.path.archive.dst}"/flink-1.7.0"
+    }
+    config {
+      # flink.yaml entries
+      yaml {
+        env.pid.dir = "/tmp/flink-1.7.0-pid"
+      }
+    }
+  }
+}

peel-extensions/src/main/resources/reference.flink-1.7.2.conf

@@ -0,0 +1,19 @@
+# include common flink configuration
+include "reference.flink.conf"
+
+system {
+  flink {
+    path {
+      archive.url = "http://archive.apache.org/dist/flink/flink-1.7.2/flink-1.7.2-bin-hadoop28-scala_2.12.tgz"
+      archive.md5 = "b9469d4d166520ec767bac0d82c165a7"
+      archive.src = ${app.path.downloads}"/flink-1.7.2-bin-hadoop28-scala_2.12.tgz"
+      home = ${system.flink.path.archive.dst}"/flink-1.7.2"
+    }
+    config {
+      # flink.yaml entries
+      yaml {
+        env.pid.dir = "/tmp/flink-1.7.2-pid"
+      }
+    }
+  }
+}

peel-extensions/src/main/resources/reference.flink-yarn-1.7.0.conf

@@ -0,0 +1,2 @@
+# include common flink configuration
+include "reference.flink-1.7.0.conf"

peel-extensions/src/main/resources/reference.flink-yarn-1.7.2.conf

@@ -0,0 +1,2 @@
+# include common flink configuration
+include "reference.flink-1.7.2.conf"

peel-extensions/src/main/resources/reference.hadoop-3.x.conf

@@ -0,0 +1,58 @@
+system {
+    hadoop-3 {
+        user = ${system.default.user}
+        group = ${system.default.group}
+        path {
+            isShared = ${system.default.path.isShared}
+            archive.dst = ${app.path.systems}
+            config = ${system.hadoop-3.path.home}"/etc/hadoop"
+            log = ${system.hadoop-3.path.home}"/logs"
+            input = ${system.hadoop-3.config.core.fs.default.name}"/tmp/input"
+            output = ${system.hadoop-3.config.core.fs.default.name}"/tmp/output"
+        }
+        format = true
+        startup {
+            max.attempts = ${system.default.startup.max.attempts}
+            polling {
+                counter = ${system.default.startup.polling.counter}
+                interval = ${system.default.startup.polling.interval}
+            }
+        }
+        config {
+            # put list of masters
+            masters = ${system.default.config.masters}
+            # put list of workers
+            workers = ${system.default.config.slaves}
+            # unfortunately, the slaves config key is hard-coded in Java code
+            slaves = ${system.default.config.slaves}
+            # hadoop-env.sh entries
+            env {
+                JAVA_HOME = ${system.default.config.java.home}
+                HADOOP_INSTALL = ${system.hadoop-3.path.home}
+                # directory where process IDs are stored
+                HADOOP_PID_DIR = "/tmp/hadoop-3/pid"
+                # avoids loading wrong native library in the default case
+                # override with /lib/native if lib exists
+                HADOOP_COMMON_LIB_NATIVE_DIR = "$HADOOP_INSTALL/lib/native"
+            }
+            # core-site.xml entries
+            core {
+                fs.default.name = "hdfs://localhost:9000"
+            }
+            # hdfs-site.xml entries
+            hdfs {
+                dfs.replication = 1
+                dfs.namenode.name.dir = "file:///tmp/hdfs-3/name"
+                dfs.datanode.data.dir = "file:///tmp/hdfs-3/data"
+            }
+            # mapred-site.xml entries
+            mapred { }
+            # yarn-site.xml entries
+            yarn { }
+            # capacity-scheduler.xml entries
+            capacity-scheduler { }
+            # container-executor.cfg entries
+            container-executor { }
+        }
+    }
+}

peel-extensions/src/main/resources/reference.hdfs-3.1.1.conf

@@ -0,0 +1,20 @@
+# include common hadoop-3.x configuration
+include "reference.hadoop-3.x.conf"
+
+system {
+    hadoop-3 {
+        path {
+            archive.url = "http://archive.apache.org/dist/hadoop/core/hadoop-3.1.1/hadoop-3.1.1.tar.gz"
+            archive.md5 = "0b6ab06b59ae75f433de387783f19011"
+            archive.src = ${app.path.downloads}"/hadoop-3.1.1.tar.gz"
+            home = ${system.hadoop-3.path.archive.dst}"/hadoop-3.1.1"
+        }
+        config {
+            # hadoop-env.sh entries
+            env {
+                # directory where process IDs are stored
+                HADOOP_PID_DIR = "/tmp/hadoop-3.1.1-pid"
+            }
+        }
+    }
+}

peel-extensions/src/main/resources/reference.yarn-3.1.1.conf

@@ -0,0 +1,5 @@
+# include common yarn-3.x configuration
+include "reference.yarn-3.x.conf"
+
+# pick up hdfs-3.1.1 configuration containing archive data
+include "reference.hdfs-3.1.1.conf"

peel-extensions/src/main/resources/reference.yarn-3.x.conf

@@ -0,0 +1,2 @@
+# include common hadoop-3.x configuration
+include "reference.hadoop-3.x.conf"