transform_service.py 文件源码

def when_i_am_elected_leader(self, event):
        """Callback when this host gets elected leader."""

        # set running state
        self.previously_running = True

        self.LOG.info("Monasca Transform service running on %s "
                      "has been elected leader" % str(self.my_host_name))

        if CONF.service.spark_python_files:
            pyfiles = (" --py-files %s"
                       % CONF.service.spark_python_files)
        else:
            pyfiles = ''

        event_logging_dest = ''
        if (CONF.service.spark_event_logging_enabled and
                CONF.service.spark_event_logging_dest):
            event_logging_dest = (
                "--conf spark.eventLog.dir="
                "file://%s" %
                CONF.service.spark_event_logging_dest)

        # Build the command to start the Spark driver
        spark_cmd = "".join((
            "export SPARK_HOME=",
            CONF.service.spark_home,
            " && ",
            "spark-submit --master ",
            CONF.service.spark_master_list,
            " --conf spark.eventLog.enabled=",
            CONF.service.spark_event_logging_enabled,
            event_logging_dest,
            " --jars " + CONF.service.spark_jars_list,
            pyfiles,
            " " + CONF.service.spark_driver))

        # Start the Spark driver
        # (specify shell=True in order to
        #  correctly handle wildcards in the spark_cmd)
        subprocess.call(spark_cmd, shell=True)