diff --git a/bigtop-manager-dao/src/main/java/org/apache/bigtop/manager/dao/query/ComponentQuery.java b/bigtop-manager-dao/src/main/java/org/apache/bigtop/manager/dao/query/ComponentQuery.java
index 5fda2fba..6fc5c529 100644
--- a/bigtop-manager-dao/src/main/java/org/apache/bigtop/manager/dao/query/ComponentQuery.java
+++ b/bigtop-manager-dao/src/main/java/org/apache/bigtop/manager/dao/query/ComponentQuery.java
@@ -39,6 +39,8 @@ public class ComponentQuery {
 
     private String hostname;
 
+    private List<String> hostnames;
+
     private Long serviceId;
 
     private List<String> serviceNames;
diff --git a/bigtop-manager-dao/src/main/resources/mapper/mysql/ComponentMapper.xml b/bigtop-manager-dao/src/main/resources/mapper/mysql/ComponentMapper.xml
index 964295af..8a7f1fa0 100644
--- a/bigtop-manager-dao/src/main/resources/mapper/mysql/ComponentMapper.xml
+++ b/bigtop-manager-dao/src/main/resources/mapper/mysql/ComponentMapper.xml
@@ -59,6 +59,12 @@
             <if test="query.hostname != null and query.hostname != ''">
                 and h.hostname = #{query.hostname}
             </if>
+            <if test="query.hostnames != null and query.hostnames.size() > 0">
+                and h.hostname in
+                <foreach collection="query.hostnames" item="hostname" index="index" open="(" close=")" separator=",">
+                    #{hostname}
+                </foreach>
+            </if>
             <if test="query.serviceId != null">
                 and comp.service_id = #{query.serviceId}
             </if>
diff --git a/bigtop-manager-dao/src/main/resources/mapper/postgresql/ComponentMapper.xml b/bigtop-manager-dao/src/main/resources/mapper/postgresql/ComponentMapper.xml
index 964295af..8a7f1fa0 100644
--- a/bigtop-manager-dao/src/main/resources/mapper/postgresql/ComponentMapper.xml
+++ b/bigtop-manager-dao/src/main/resources/mapper/postgresql/ComponentMapper.xml
@@ -59,6 +59,12 @@
             <if test="query.hostname != null and query.hostname != ''">
                 and h.hostname = #{query.hostname}
             </if>
+            <if test="query.hostnames != null and query.hostnames.size() > 0">
+                and h.hostname in
+                <foreach collection="query.hostnames" item="hostname" index="index" open="(" close=")" separator=",">
+                    #{hostname}
+                </foreach>
+            </if>
             <if test="query.serviceId != null">
                 and comp.service_id = #{query.serviceId}
             </if>
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostAddJobFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostAddJobFactory.java
index ecf9b911..98474d8d 100644
--- a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostAddJobFactory.java
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostAddJobFactory.java
@@ -20,7 +20,6 @@
 
 import org.apache.bigtop.manager.common.enums.Command;
 import org.apache.bigtop.manager.server.command.CommandIdentifier;
-import org.apache.bigtop.manager.server.command.factory.cluster.AbstractClusterJobFactory;
 import org.apache.bigtop.manager.server.command.job.HostAddJob;
 import org.apache.bigtop.manager.server.command.job.Job;
 import org.apache.bigtop.manager.server.command.job.JobContext;
@@ -35,7 +34,7 @@
 @Slf4j
 @Component
 @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
-public class HostAddJobFactory extends AbstractClusterJobFactory {
+public class HostAddJobFactory extends AbstractHostJobFactory {
 
     @Override
     public CommandIdentifier getCommandIdentifier() {
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostRestartJobFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostRestartJobFactory.java
new file mode 100644
index 00000000..c0e25341
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostRestartJobFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.factory.host;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.HostRestartJob;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class HostRestartJobFactory extends AbstractHostJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.HOST, Command.RESTART);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new HostRestartJob(jobContext);
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStartJobFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStartJobFactory.java
new file mode 100644
index 00000000..844f6518
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStartJobFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.factory.host;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.HostStartJob;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class HostStartJobFactory extends AbstractHostJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.HOST, Command.START);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new HostStartJob(jobContext);
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStopJobFactory.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStopJobFactory.java
new file mode 100644
index 00000000..efa6e81f
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/factory/host/HostStopJobFactory.java
@@ -0,0 +1,48 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.factory.host;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.server.command.CommandIdentifier;
+import org.apache.bigtop.manager.server.command.job.HostStopJob;
+import org.apache.bigtop.manager.server.command.job.Job;
+import org.apache.bigtop.manager.server.command.job.JobContext;
+import org.apache.bigtop.manager.server.enums.CommandLevel;
+
+import org.springframework.beans.factory.config.ConfigurableBeanFactory;
+import org.springframework.context.annotation.Scope;
+import org.springframework.stereotype.Component;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
+public class HostStopJobFactory extends AbstractHostJobFactory {
+
+    @Override
+    public CommandIdentifier getCommandIdentifier() {
+        return new CommandIdentifier(CommandLevel.HOST, Command.STOP);
+    }
+
+    @Override
+    public Job createJob(JobContext jobContext) {
+        return new HostStopJob(jobContext);
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/AbstractHostJob.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/AbstractHostJob.java
new file mode 100644
index 00000000..952f6daa
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/AbstractHostJob.java
@@ -0,0 +1,137 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.job;
+
+import org.apache.bigtop.manager.common.enums.Command;
+import org.apache.bigtop.manager.dao.po.ComponentPO;
+import org.apache.bigtop.manager.dao.query.ComponentQuery;
+import org.apache.bigtop.manager.dao.repository.ComponentDao;
+import org.apache.bigtop.manager.server.command.stage.ComponentStartStage;
+import org.apache.bigtop.manager.server.command.stage.ComponentStopStage;
+import org.apache.bigtop.manager.server.command.stage.StageContext;
+import org.apache.bigtop.manager.server.holder.SpringContextHolder;
+import org.apache.bigtop.manager.server.model.dto.ComponentDTO;
+import org.apache.bigtop.manager.server.model.dto.ServiceDTO;
+import org.apache.bigtop.manager.server.utils.StackDAGUtils;
+import org.apache.bigtop.manager.server.utils.StackUtils;
+
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public abstract class AbstractHostJob extends AbstractJob {
+
+    protected ComponentDao componentDao;
+
+    public AbstractHostJob(JobContext jobContext) {
+        super(jobContext);
+    }
+
+    @Override
+    protected void injectBeans() {
+        super.injectBeans();
+
+        this.componentDao = SpringContextHolder.getBean(ComponentDao.class);
+    }
+
+    @Override
+    protected void beforeCreateStages() {
+        super.beforeCreateStages();
+    }
+
+    protected StageContext createStageContext(String componentName, List<String> hostnames) {
+        StageContext stageContext = StageContext.fromCommandDTO(jobContext.getCommandDTO());
+
+        ServiceDTO serviceDTO = StackUtils.getServiceDTOByComponentName(componentName);
+        ComponentDTO componentDTO = StackUtils.getComponentDTO(componentName);
+
+        stageContext.setHostnames(hostnames);
+        stageContext.setServiceDTO(serviceDTO);
+        stageContext.setComponentDTO(componentDTO);
+
+        return stageContext;
+    }
+
+    protected void createStartStages() {
+        List<ComponentPO> componentPOList = getComponentPOList();
+        List<String> todoList = StackDAGUtils.getTodoList(getComponentNames(componentPOList), Command.START);
+
+        for (String componentCommand : todoList) {
+            String[] split = componentCommand.split("-");
+            String componentName = split[0];
+
+            if (StackUtils.isClientComponent(componentName)) {
+                continue;
+            }
+
+            List<String> hostnames = getHostnames();
+            if (CollectionUtils.isEmpty(hostnames)) {
+                continue;
+            }
+
+            StageContext stageContext = createStageContext(componentName, hostnames);
+            stages.add(new ComponentStartStage(stageContext));
+        }
+    }
+
+    protected void createStopStages() {
+        List<ComponentPO> componentPOList = getComponentPOList();
+        List<String> todoList = StackDAGUtils.getTodoList(getComponentNames(componentPOList), Command.STOP);
+
+        for (String componentCommand : todoList) {
+            String[] split = componentCommand.split("-");
+            String componentName = split[0];
+
+            if (StackUtils.isClientComponent(componentName)) {
+                continue;
+            }
+
+            List<String> hostnames = getHostnames();
+            if (CollectionUtils.isEmpty(hostnames)) {
+                continue;
+            }
+
+            StageContext stageContext = createStageContext(componentName, hostnames);
+            stages.add(new ComponentStopStage(stageContext));
+        }
+    }
+
+    private List<ComponentPO> getComponentPOList() {
+        ComponentQuery query = ComponentQuery.builder()
+                .clusterId(clusterPO.getId())
+                .hostnames(getHostnames())
+                .build();
+        return componentDao.findByQuery(query);
+    }
+
+    private List<String> getComponentNames(List<ComponentPO> componentPOList) {
+        if (componentPOList == null) {
+            return new ArrayList<>();
+        } else {
+            return componentPOList.stream().map(ComponentPO::getName).distinct().toList();
+        }
+    }
+
+    private List<String> getHostnames() {
+        return jobContext.getCommandDTO().getHostCommands().stream()
+                .flatMap(hostCommandDTO -> hostCommandDTO.getHostnames().stream())
+                .toList();
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostRestartJob.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostRestartJob.java
new file mode 100644
index 00000000..926b9d53
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostRestartJob.java
@@ -0,0 +1,38 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.job;
+
+public class HostRestartJob extends AbstractHostJob {
+
+    public HostRestartJob(JobContext jobContext) {
+        super(jobContext);
+    }
+
+    @Override
+    protected void createStages() {
+        super.createStopStages();
+
+        super.createStartStages();
+    }
+
+    @Override
+    public String getName() {
+        return "Restart host";
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStartJob.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStartJob.java
new file mode 100644
index 00000000..b49d4d2a
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStartJob.java
@@ -0,0 +1,36 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.job;
+
+public class HostStartJob extends AbstractHostJob {
+
+    public HostStartJob(JobContext jobContext) {
+        super(jobContext);
+    }
+
+    @Override
+    protected void createStages() {
+        super.createStartStages();
+    }
+
+    @Override
+    public String getName() {
+        return "Start host";
+    }
+}
diff --git a/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStopJob.java b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStopJob.java
new file mode 100644
index 00000000..aa4d605e
--- /dev/null
+++ b/bigtop-manager-server/src/main/java/org/apache/bigtop/manager/server/command/job/HostStopJob.java
@@ -0,0 +1,36 @@
+/*
+ * 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
+ *
+ *    https://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.bigtop.manager.server.command.job;
+
+public class HostStopJob extends AbstractHostJob {
+
+    public HostStopJob(JobContext jobContext) {
+        super(jobContext);
+    }
+
+    @Override
+    protected void createStages() {
+        super.createStopStages();
+    }
+
+    @Override
+    public String getName() {
+        return "Stop host";
+    }
+}