src/toil_lib/spark.py
def start(self, job):
        """
        Start spark and hdfs worker containers

        :param job: The underlying job.
        """

        # start spark and our datanode
        self.sparkContainerID = dockerCheckOutput(job=job,
                                                  defer=STOP,
                                                  workDir=os.getcwd(),
                                                  tool="quay.io/ucsc_cgl/apache-spark-worker:1.5.2",
                                                  dockerParameters=["--net=host",
                                                                    "-d",
                                                                    "-v", "/mnt/ephemeral/:/ephemeral/:rw",
                                                                    "-e",
                                                                    "\"SPARK_MASTER_IP=" + self.masterIP + ":" + _SPARK_MASTER_PORT + "\"",
                                                                    "-e", "SPARK_LOCAL_DIRS=/ephemeral/spark/local",
                                                                    "-e", "SPARK_WORKER_DIR=/ephemeral/spark/work"],
                                                  parameters=[self.masterIP + ":" + _SPARK_MASTER_PORT])[:-1]
        self.__start_datanode(job)
        
        # fake do/while to check if HDFS is up
        hdfs_down = True
        retries = 0
        while hdfs_down and (retries < 5):

            _log.info("Sleeping 30 seconds before checking HDFS startup.")
            time.sleep(30)
            clusterID = ""
            try:
                clusterID = subprocess.check_output(["docker",
                                                     "exec",
                                                     self.hdfsContainerID,
                                                     "grep",
                                                     "clusterID",
                                                     "-R",
                                                     "/opt/apache-hadoop/logs"])
            except:
                # grep returns a non-zero exit code if the pattern is not found
                # we expect to not find the pattern, so a non-zero code is OK
                pass

            if "Incompatible" in clusterID:
                _log.warning("Hadoop Datanode failed to start with: %s", clusterID)
                _log.warning("Retrying container startup, retry #%d.", retries)
                retries += 1

                _log.warning("Removing ephemeral hdfs directory.")
                subprocess.check_call(["docker",
                                       "exec",
                                       self.hdfsContainerID,
                                       "rm",
                                       "-rf",
                                       "/ephemeral/hdfs"])

                _log.warning("Killing container %s.", self.hdfsContainerID)
                subprocess.check_call(["docker",
                                       "kill",
                                       self.hdfsContainerID])

                # todo: this is copied code. clean up!
                _log.info("Restarting datanode.")
                self.__start_datanode(job)

            else:
                _log.info("HDFS datanode started up OK!")
                hdfs_down = False

        if retries >= 5:
            raise RuntimeError("Failed %d times trying to start HDFS datanode." % retries)

        return
def start(self, job):
        # start spark and our datanode
        self.sparkContainerID = dockerCheckOutput(job=job,
                                                  defer=STOP,
                                                  workDir=os.getcwd(),
                                                  tool="quay.io/ucsc_cgl/apache-spark-worker:1.5.2",
                                                  dockerParameters=["--net=host",
                                                                    "-d",
                                                                    "-v", "/mnt/ephemeral/:/ephemeral/:rw",
                                                                    "-e",
                                                                    "\"SPARK_MASTER_IP=" + self.masterIP + ":" + _SPARK_MASTER_PORT + "\"",
                                                                    "-e", "SPARK_LOCAL_DIRS=/ephemeral/spark/local",
                                                                    "-e", "SPARK_WORKER_DIR=/ephemeral/spark/work"],
                                                  parameters=[self.masterIP + ":" + _SPARK_MASTER_PORT])[:-1]
        self.__start_datanode(job)
        
        # fake do/while to check if HDFS is up
        hdfs_down = True
        retries = 0
        while hdfs_down and (retries < 5):

            _log.info("Sleeping 30 seconds before checking HDFS startup.")
            time.sleep(30)
            clusterID = ""
            try:
                clusterID = subprocess.check_output(["docker",
                                                     "exec",
                                                     self.hdfsContainerID,
                                                     "grep",
                                                     "clusterID",
                                                     "-R",
                                                     "/opt/apache-hadoop/logs"])
            except:
                # grep returns a non-zero exit code if the pattern is not found
                # we expect to not find the pattern, so a non-zero code is OK
                pass

            if "Incompatible" in clusterID:
                _log.warning("Hadoop Datanode failed to start with: %s", clusterID)
                _log.warning("Retrying container startup, retry #%d.", retries)
                retries += 1

                _log.warning("Removing ephemeral hdfs directory.")
                subprocess.check_call(["docker",
                                       "exec",
                                       self.hdfsContainerID,
                                       "rm",
                                       "-rf",
                                       "/ephemeral/hdfs"])

                _log.warning("Killing container %s.", self.hdfsContainerID)
                subprocess.check_call(["docker",
                                       "kill",
                                       self.hdfsContainerID])

                # todo: this is copied code. clean up!
                _log.info("Restarting datanode.")
                self.__start_datanode(job)

            else:
                _log.info("HDFS datanode started up OK!")
                hdfs_down = False

        if retries >= 5:
            raise RuntimeError("Failed %d times trying to start HDFS datanode." % retries)

        return
def start ( self , job ) : # start spark and our datanode self . sparkContainerID = dockerCheckOutput ( job = job , defer = STOP , workDir = os . getcwd ( ) , tool = "quay.io/ucsc_cgl/apache-spark-worker:1.5.2" , dockerParameters = [ "--net=host" , "-d" , "-v" , "/mnt/ephemeral/:/ephemeral/:rw" , "-e" , "\"SPARK_MASTER_IP=" + self . masterIP + ":" + _SPARK_MASTER_PORT + "\"" , "-e" , "SPARK_LOCAL_DIRS=/ephemeral/spark/local" , "-e" , "SPARK_WORKER_DIR=/ephemeral/spark/work" ] , parameters = [ self . masterIP + ":" + _SPARK_MASTER_PORT ] ) [ : - 1 ] self . __start_datanode ( job ) # fake do/while to check if HDFS is up hdfs_down = True retries = 0 while hdfs_down and ( retries < 5 ) : _log . info ( "Sleeping 30 seconds before checking HDFS startup." ) time . sleep ( 30 ) clusterID = "" try : clusterID = subprocess . check_output ( [ "docker" , "exec" , self . hdfsContainerID , "grep" , "clusterID" , "-R" , "/opt/apache-hadoop/logs" ] ) except : # grep returns a non-zero exit code if the pattern is not found # we expect to not find the pattern, so a non-zero code is OK pass if "Incompatible" in clusterID : _log . warning ( "Hadoop Datanode failed to start with: %s" , clusterID ) _log . warning ( "Retrying container startup, retry #%d." , retries ) retries += 1 _log . warning ( "Removing ephemeral hdfs directory." ) subprocess . check_call ( [ "docker" , "exec" , self . hdfsContainerID , "rm" , "-rf" , "/ephemeral/hdfs" ] ) _log . warning ( "Killing container %s." , self . hdfsContainerID ) subprocess . check_call ( [ "docker" , "kill" , self . hdfsContainerID ] ) # todo: this is copied code. clean up! _log . info ( "Restarting datanode." ) self . __start_datanode ( job ) else : _log . info ( "HDFS datanode started up OK!" ) hdfs_down = False if retries >= 5 : raise RuntimeError ( "Failed %d times trying to start HDFS datanode." % retries ) return
Start spark and hdfs worker containers

        :param job: The underlying job.
Start spark and hdfs worker containers