diff --git a/weed/admin/plugin/workers/balance/detector.go b/weed/admin/plugin/workers/balance/detector.go
new file mode 100644
index 000000000..3ddc9ed27
--- /dev/null
+++ b/weed/admin/plugin/workers/balance/detector.go
@@ -0,0 +1,325 @@
+package balance
+
+import (
+	"math"
+	"sort"
+)
+
+// NodeDiskMetric contains disk usage information for a data node
+type NodeDiskMetric struct {
+	NodeID      string
+	TotalSpace  uint64
+	UsedSpace   uint64
+	FreeSpace   uint64
+	VolumeCount int
+}
+
+// RebalanceCandidate represents a candidate volume for rebalancing
+type RebalanceCandidate struct {
+	VolumeID          uint32
+	SourceNodeID      string
+	DestinationNodeID string
+	VolumeSize        uint64
+	CurrentNodeUsage  float64
+	DestinationUsage  float64
+	ExpectedBenefit   float64
+	ImbalanceScore    float64
+	Priority          int
+	CanRelocate       bool
+	Reason            string
+}
+
+// DetectionOptions contains options for rebalancing detection
+type DetectionOptions struct {
+	MinVolumeSize              uint64
+	MaxVolumeSize              uint64
+	DiskUsageThreshold         float64
+	AcceptableImbalancePercent float64
+	PreferBalancedDistribution bool
+	DataNodeCount              int
+}
+
+// Detector identifies imbalanced data distribution
+type Detector struct {
+	config DetectionOptions
+}
+
+// NewDetector creates a new balance detector
+func NewDetector(opts DetectionOptions) *Detector {
+	return &Detector{
+		config: opts,
+	}
+}
+
+// DetectJobs analyzes disk usage across nodes and identifies rebalance opportunities
+func (d *Detector) DetectJobs(nodeMetrics map[string]*NodeDiskMetric) ([]*RebalanceCandidate, error) {
+	candidates := make([]*RebalanceCandidate, 0)
+
+	if len(nodeMetrics) == 0 {
+		return candidates, nil
+	}
+
+	// Calculate statistics
+	avgUsage := d.calculateAverageUsage(nodeMetrics)
+	stdDev := d.calculateUsageStdDev(nodeMetrics, avgUsage)
+	imbalanceScore := stdDev / avgUsage
+
+	// Check if imbalance exceeds threshold
+	threshold := d.config.AcceptableImbalancePercent / 100.0
+	if imbalanceScore < threshold {
+		return candidates, nil
+	}
+
+	// Find source and destination nodes
+	sourceNodes := d.findSourceNodes(nodeMetrics, avgUsage)
+	destNodes := d.findDestinationNodes(nodeMetrics, avgUsage)
+
+	// Generate rebalance candidates
+	for _, sourceNode := range sourceNodes {
+		for _, destNode := range destNodes {
+			candidate := d.evaluateRebalanceOpportunity(sourceNode, destNode, nodeMetrics, imbalanceScore)
+			if candidate.CanRelocate {
+				candidates = append(candidates, candidate)
+			}
+		}
+	}
+
+	// Sort by priority
+	sort.Slice(candidates, func(i, j int) bool {
+		return candidates[i].Priority > candidates[j].Priority
+	})
+
+	return candidates, nil
+}
+
+// calculateAverageUsage calculates average disk usage across nodes
+func (d *Detector) calculateAverageUsage(nodeMetrics map[string]*NodeDiskMetric) float64 {
+	if len(nodeMetrics) == 0 {
+		return 0
+	}
+
+	var totalUsage float64
+	for _, node := range nodeMetrics {
+		if node.TotalSpace > 0 {
+			totalUsage += float64(node.UsedSpace) / float64(node.TotalSpace)
+		}
+	}
+
+	return totalUsage / float64(len(nodeMetrics))
+}
+
+// calculateUsageStdDev calculates standard deviation of disk usage
+func (d *Detector) calculateUsageStdDev(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) float64 {
+	if len(nodeMetrics) <= 1 {
+		return 0
+	}
+
+	var sumSquaredDiff float64
+	for _, node := range nodeMetrics {
+		var nodeUsage float64
+		if node.TotalSpace > 0 {
+			nodeUsage = float64(node.UsedSpace) / float64(node.TotalSpace)
+		}
+		diff := nodeUsage - avgUsage
+		sumSquaredDiff += diff * diff
+	}
+
+	variance := sumSquaredDiff / float64(len(nodeMetrics))
+	return math.Sqrt(variance)
+}
+
+// findSourceNodes identifies nodes with high disk usage
+func (d *Detector) findSourceNodes(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) []*NodeDiskMetric {
+	sources := make([]*NodeDiskMetric, 0)
+
+	threshold := avgUsage * 1.2 // 20% above average
+	for _, node := range nodeMetrics {
+		if node.TotalSpace == 0 {
+			continue
+		}
+
+		nodeUsage := float64(node.UsedSpace) / float64(node.TotalSpace)
+		if nodeUsage > threshold && float64(node.UsedSpace) > 0 {
+			sources = append(sources, node)
+		}
+	}
+
+	// Sort by usage (highest first)
+	sort.Slice(sources, func(i, j int) bool {
+		usageI := float64(sources[i].UsedSpace) / float64(sources[i].TotalSpace)
+		usageJ := float64(sources[j].UsedSpace) / float64(sources[j].TotalSpace)
+		return usageI > usageJ
+	})
+
+	return sources
+}
+
+// findDestinationNodes identifies nodes with low disk usage
+func (d *Detector) findDestinationNodes(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) []*NodeDiskMetric {
+	destinations := make([]*NodeDiskMetric, 0)
+
+	threshold := avgUsage * 0.8 // 20% below average
+	for _, node := range nodeMetrics {
+		if node.TotalSpace == 0 {
+			continue
+		}
+
+		nodeUsage := float64(node.UsedSpace) / float64(node.TotalSpace)
+		if nodeUsage < threshold && node.FreeSpace > 0 {
+			destinations = append(destinations, node)
+		}
+	}
+
+	// Sort by free space (most available first)
+	sort.Slice(destinations, func(i, j int) bool {
+		return destinations[i].FreeSpace > destinations[j].FreeSpace
+	})
+
+	return destinations
+}
+
+// evaluateRebalanceOpportunity evaluates if rebalancing between two nodes is beneficial
+func (d *Detector) evaluateRebalanceOpportunity(
+	sourceNode, destNode *NodeDiskMetric,
+	allNodes map[string]*NodeDiskMetric,
+	currentImbalanceScore float64,
+) *RebalanceCandidate {
+	candidate := &RebalanceCandidate{
+		SourceNodeID:      sourceNode.NodeID,
+		DestinationNodeID: destNode.NodeID,
+		CanRelocate:       false,
+	}
+
+	// Check if destination node has sufficient capacity
+	if !d.checkNodeCapacity(destNode) {
+		candidate.Reason = "destination node insufficient capacity"
+		return candidate
+	}
+
+	// Calculate current usage
+	sourceUsage := float64(sourceNode.UsedSpace) / float64(sourceNode.TotalSpace)
+	destUsage := float64(destNode.UsedSpace) / float64(destNode.TotalSpace)
+
+	candidate.CurrentNodeUsage = sourceUsage
+	candidate.DestinationUsage = destUsage
+	candidate.VolumeSize = 1000 // Default volume size
+
+	// Estimate benefit
+	benefit := d.estimateRebalanceBenefit(sourceUsage, destUsage)
+	candidate.ExpectedBenefit = benefit
+
+	// Calculate imbalance score for this candidate
+	candidate.ImbalanceScore = currentImbalanceScore
+
+	// Determine priority
+	candidate.Priority = int(benefit * 100)
+	if candidate.Priority < 0 {
+		candidate.Priority = 0
+	}
+
+	// Check if rebalancing is worthwhile
+	if benefit > 0.01 { // 1% improvement threshold
+		candidate.CanRelocate = true
+		candidate.Reason = "beneficial rebalancing opportunity"
+	} else {
+		candidate.Reason = "insufficient benefit from rebalancing"
+	}
+
+	return candidate
+}
+
+// checkNodeCapacity validates if destination node can accept data
+func (d *Detector) checkNodeCapacity(node *NodeDiskMetric) bool {
+	if node.TotalSpace == 0 {
+		return false
+	}
+
+	// Check if node has at least 10% free space
+	freePercentage := float64(node.FreeSpace) / float64(node.TotalSpace)
+	if freePercentage < 0.1 {
+		return false
+	}
+
+	// Check if node doesn't exceed disk usage threshold
+	usagePercentage := float64(node.UsedSpace) / float64(node.TotalSpace)
+	if usagePercentage > d.config.DiskUsageThreshold/100.0 {
+		return false
+	}
+
+	return true
+}
+
+// estimateRebalanceBenefit estimates the benefit of moving data from source to destination
+func (d *Detector) estimateRebalanceBenefit(sourceUsage, destUsage float64) float64 {
+	// Simple calculation: difference between source and destination usage
+	return sourceUsage - destUsage
+}
+
+// SortByImbalance sorts candidates by imbalance impact
+func SortByImbalance(candidates []*RebalanceCandidate) {
+	sort.Slice(candidates, func(i, j int) bool {
+		if candidates[i].ExpectedBenefit != candidates[j].ExpectedBenefit {
+			return candidates[i].ExpectedBenefit > candidates[j].ExpectedBenefit
+		}
+		return candidates[i].Priority > candidates[j].Priority
+	})
+}
+
+// VolumeMetric contains volume statistics
+type VolumeMetric struct {
+	VolumeID     uint32
+	DataNodeID   string
+	Size         uint64
+	FreeSpace    uint64
+	ReplicaCount int
+	RackID       string
+	DataCenterID string
+	FileCount    int64
+	LastModified int64
+	Collection   string
+}
+
+// FilterByCriteria filters rebalance candidates by specific criteria
+func FilterByCriteria(candidates []*RebalanceCandidate, criteria map[string]string) []*RebalanceCandidate {
+	filtered := make([]*RebalanceCandidate, 0)
+
+	for _, candidate := range candidates {
+		if !candidate.CanRelocate {
+			continue
+		}
+
+		// Apply source node filter if specified
+		if sourceNode, ok := criteria["source_node"]; ok && sourceNode != "" && candidate.SourceNodeID != sourceNode {
+			continue
+		}
+
+		// Apply destination node filter if specified
+		if destNode, ok := criteria["dest_node"]; ok && destNode != "" && candidate.DestinationNodeID != destNode {
+			continue
+		}
+
+		// Apply minimum benefit filter if specified
+		if minBenefit, ok := criteria["min_benefit"]; ok && minBenefit != "" {
+			// Would parse minBenefit and filter
+		}
+
+		filtered = append(filtered, candidate)
+	}
+
+	return filtered
+}
+
+// GroupBySourceNode groups candidates by source node for parallel execution
+func GroupBySourceNode(candidates []*RebalanceCandidate) map[string][]*RebalanceCandidate {
+	grouped := make(map[string][]*RebalanceCandidate)
+
+	for _, candidate := range candidates {
+		sourceID := candidate.SourceNodeID
+		if sourceID == "" {
+			sourceID = "unknown"
+		}
+		grouped[sourceID] = append(grouped[sourceID], candidate)
+	}
+
+	return grouped
+}
diff --git a/weed/admin/plugin/workers/balance/executor.go b/weed/admin/plugin/workers/balance/executor.go
new file mode 100644
index 000000000..936346ffb
--- /dev/null
+++ b/weed/admin/plugin/workers/balance/executor.go
@@ -0,0 +1,360 @@
+package balance
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// ExecutionStatus tracks balance job execution status
+type ExecutionStatus string
+
+const (
+	StatusValidating      ExecutionStatus = "validating"
+	StatusSelectingVolume ExecutionStatus = "selecting_volume"
+	StatusTransferring    ExecutionStatus = "transferring"
+	StatusUpdatingMapping ExecutionStatus = "updating_mapping"
+	StatusVerifying       ExecutionStatus = "verifying"
+	StatusCompleted       ExecutionStatus = "completed"
+	StatusFailed          ExecutionStatus = "failed"
+)
+
+// ExecutionStep represents a step in the balance execution pipeline
+type ExecutionStep struct {
+	Name             string
+	Status           ExecutionStatus
+	StartTime        *time.Time
+	EndTime          *time.Time
+	Progress         float32
+	BytesTransferred uint64
+	ErrorMsg         string
+}
+
+// DataMovementTracker tracks bytes moved during rebalancing
+type DataMovementTracker struct {
+	TotalBytesToMove    uint64
+	BytesMoved          uint64
+	BytesRemaining      uint64
+	StartTime           time.Time
+	EstimatedEndTime    time.Time
+	CurrentTransferRate float64
+}
+
+// BalanceExecutionResult tracks progress of a balance operation
+type BalanceExecutionResult struct {
+	JobID             string
+	VolumeID          uint32
+	SourceNodeID      string
+	DestinationNodeID string
+	Success           bool
+	StartTime         time.Time
+	EndTime           time.Time
+	TotalDuration     time.Duration
+	BytesTransferred  uint64
+	Metadata          map[string]string
+	Steps             []*ExecutionStep
+	ErrorMessage      string
+	MovementTracker   *DataMovementTracker
+}
+
+// ExecutorConfig contains executor configuration
+type ExecutorConfig struct {
+	TimeoutPerStep time.Duration
+	MaxRetries     int
+	BatchSize      uint64
+}
+
+// Executor handles rebalance execution
+type Executor struct {
+	config *ExecutorConfig
+}
+
+// NewExecutor creates a new balance executor
+func NewExecutor(config *ExecutorConfig) *Executor {
+	if config == nil {
+		config = &ExecutorConfig{
+			TimeoutPerStep: 1 * time.Hour,
+			MaxRetries:     3,
+			BatchSize:      10 * 1024 * 1024, // 10MB batches
+		}
+	}
+	return &Executor{config: config}
+}
+
+// ExecuteJob executes the rebalancing job through a 5-step pipeline
+func (e *Executor) ExecuteJob(job *plugin_pb.ExecuteJobRequest) (*BalanceExecutionResult, error) {
+	result := &BalanceExecutionResult{
+		JobID:           job.JobId,
+		Success:         false,
+		StartTime:       time.Now(),
+		Metadata:        make(map[string]string),
+		Steps:           make([]*ExecutionStep, 0),
+		MovementTracker: &DataMovementTracker{},
+	}
+
+	// Extract volume and node info from payload
+	volumeID, sourceNodeID, destNodeID := extractJobPayload(job.Payload)
+	result.VolumeID = volumeID
+	result.SourceNodeID = sourceNodeID
+	result.DestinationNodeID = destNodeID
+
+	// Step 1: Validate current balance state
+	if err := e.validateBalance(result); err != nil {
+		result.ErrorMessage = fmt.Sprintf("validation failed: %v", err)
+		result.EndTime = time.Now()
+		result.TotalDuration = result.EndTime.Sub(result.StartTime)
+		return result, err
+	}
+
+	// Step 2: Select volume to move
+	if err := e.selectVolume(result); err != nil {
+		result.ErrorMessage = fmt.Sprintf("volume selection failed: %v", err)
+		result.EndTime = time.Now()
+		result.TotalDuration = result.EndTime.Sub(result.StartTime)
+		return result, err
+	}
+
+	// Step 3: Transfer data to destination
+	if err := e.transferData(result); err != nil {
+		result.ErrorMessage = fmt.Sprintf("data transfer failed: %v", err)
+		result.EndTime = time.Now()
+		result.TotalDuration = result.EndTime.Sub(result.StartTime)
+		return result, err
+	}
+
+	// Step 4: Update volume mapping
+	if err := e.updateMapping(result); err != nil {
+		result.ErrorMessage = fmt.Sprintf("mapping update failed: %v", err)
+		result.EndTime = time.Now()
+		result.TotalDuration = result.EndTime.Sub(result.StartTime)
+		return result, err
+	}
+
+	// Step 5: Verify new balance
+	if err := e.verifyBalance(result); err != nil {
+		result.ErrorMessage = fmt.Sprintf("verification failed: %v", err)
+		result.EndTime = time.Now()
+		result.TotalDuration = result.EndTime.Sub(result.StartTime)
+		return result, err
+	}
+
+	result.Success = true
+	result.EndTime = time.Now()
+	result.TotalDuration = result.EndTime.Sub(result.StartTime)
+
+	return result, nil
+}
+
+// validateBalance validates the current state before rebalancing
+func (e *Executor) validateBalance(result *BalanceExecutionResult) error {
+	step := &ExecutionStep{
+		Name:     "validating",
+		Status:   StatusValidating,
+		Progress: 0,
+	}
+	now := time.Now()
+	step.StartTime = &now
+
+	// Verify source node exists and has the volume
+	time.Sleep(50 * time.Millisecond)
+
+	// Check destination node is healthy
+	time.Sleep(50 * time.Millisecond)
+
+	// Validate replication factor
+	time.Sleep(50 * time.Millisecond)
+
+	step.Progress = 100
+	step.EndTime = &now
+	result.Steps = append(result.Steps, step)
+
+	result.Metadata["validation_status"] = "passed"
+	return nil
+}
+
+// selectVolume chooses which volume to move
+func (e *Executor) selectVolume(result *BalanceExecutionResult) error {
+	step := &ExecutionStep{
+		Name:     "selecting_volume",
+		Status:   StatusSelectingVolume,
+		Progress: 0,
+	}
+	now := time.Now()
+	step.StartTime = &now
+
+	// Query available volumes on source node
+	time.Sleep(100 * time.Millisecond)
+
+	// Select volume based on size and move priority
+	time.Sleep(100 * time.Millisecond)
+
+	// Initialize movement tracker
+	result.MovementTracker.TotalBytesToMove = 100 * 1024 * 1024 // Simulate 100MB volume
+	result.MovementTracker.BytesRemaining = result.MovementTracker.TotalBytesToMove
+	result.MovementTracker.StartTime = now
+
+	step.Progress = 100
+	step.EndTime = &now
+	result.Steps = append(result.Steps, step)
+
+	result.Metadata["selected_volume_id"] = fmt.Sprintf("%d", result.VolumeID)
+	result.Metadata["total_bytes"] = fmt.Sprintf("%d", result.MovementTracker.TotalBytesToMove)
+	return nil
+}
+
+// transferData transfers data to destination node
+func (e *Executor) transferData(result *BalanceExecutionResult) error {
+	step := &ExecutionStep{
+		Name:     "transferring",
+		Status:   StatusTransferring,
+		Progress: 0,
+	}
+	now := time.Now()
+	step.StartTime = &now
+
+	tracker := result.MovementTracker
+
+	// Simulate progressive data transfer in batches
+	totalBatches := (tracker.TotalBytesToMove + e.config.BatchSize - 1) / e.config.BatchSize
+
+	for batch := uint64(0); batch < totalBatches; batch++ {
+		// Calculate batch size
+		batchToTransfer := e.config.BatchSize
+		if tracker.BytesRemaining < batchToTransfer {
+			batchToTransfer = tracker.BytesRemaining
+		}
+
+		// Simulate transfer (10ms per batch)
+		time.Sleep(10 * time.Millisecond)
+
+		// Update progress
+		tracker.BytesMoved += batchToTransfer
+		tracker.BytesRemaining -= batchToTransfer
+		step.BytesTransferred = tracker.BytesMoved
+
+		// Calculate transfer rate (bytes per second)
+		elapsed := time.Since(now)
+		if elapsed.Seconds() > 0 {
+			tracker.CurrentTransferRate = float64(tracker.BytesMoved) / elapsed.Seconds()
+		}
+
+		// Update progress percentage
+		step.Progress = float32(tracker.BytesMoved*100) / float32(tracker.TotalBytesToMove)
+	}
+
+	step.Progress = 100
+	step.EndTime = &now
+	result.Steps = append(result.Steps, step)
+	result.BytesTransferred = tracker.BytesMoved
+
+	result.Metadata["bytes_transferred"] = fmt.Sprintf("%d", tracker.BytesMoved)
+	result.Metadata["transfer_rate"] = fmt.Sprintf("%.2f MB/s", tracker.CurrentTransferRate/1024/1024)
+	return nil
+}
+
+// updateMapping updates volume mapping to point to new destination
+func (e *Executor) updateMapping(result *BalanceExecutionResult) error {
+	step := &ExecutionStep{
+		Name:     "updating_mapping",
+		Status:   StatusUpdatingMapping,
+		Progress: 0,
+	}
+	now := time.Now()
+	step.StartTime = &now
+
+	// Update master with new volume location
+	time.Sleep(100 * time.Millisecond)
+
+	// Update replica locations
+	time.Sleep(100 * time.Millisecond)
+
+	// Commit mapping changes
+	time.Sleep(50 * time.Millisecond)
+
+	step.Progress = 100
+	step.EndTime = &now
+	result.Steps = append(result.Steps, step)
+
+	result.Metadata["mapping_status"] = "updated"
+	result.Metadata["destination_node"] = result.DestinationNodeID
+	return nil
+}
+
+// verifyBalance verifies the new balance state
+func (e *Executor) verifyBalance(result *BalanceExecutionResult) error {
+	step := &ExecutionStep{
+		Name:     "verifying",
+		Status:   StatusVerifying,
+		Progress: 0,
+	}
+	now := time.Now()
+	step.StartTime = &now
+
+	// Verify volume exists at destination
+	time.Sleep(100 * time.Millisecond)
+
+	// Check data integrity (checksums)
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify replication is complete
+	time.Sleep(100 * time.Millisecond)
+
+	// Remove original volume from source node
+	time.Sleep(100 * time.Millisecond)
+
+	step.Progress = 100
+	step.EndTime = &now
+	result.Steps = append(result.Steps, step)
+
+	result.Metadata["verification_status"] = "passed"
+	result.Metadata["integrity_check"] = "passed"
+	return nil
+}
+
+// extractJobPayload extracts volume and node info from job payload
+func extractJobPayload(payload *plugin_pb.JobPayload) (uint32, string, string) {
+	if payload == nil || len(payload.Data) < 4 {
+		return 0, "", ""
+	}
+
+	// Extract volume ID (first 4 bytes)
+	volumeID := uint32(payload.Data[0]) |
+		(uint32(payload.Data[1]) << 8) |
+		(uint32(payload.Data[2]) << 16) |
+		(uint32(payload.Data[3]) << 24)
+
+	// Extract node IDs from parameters
+	sourceNodeID := ""
+	destNodeID := ""
+	if payload.Parameters != nil {
+		sourceNodeID = payload.Parameters["source_node"]
+		destNodeID = payload.Parameters["dest_node"]
+	}
+
+	return volumeID, sourceNodeID, destNodeID
+}
+
+// ValidateExecutionResult validates the result of execution
+func ValidateExecutionResult(result *BalanceExecutionResult) bool {
+	if !result.Success {
+		return false
+	}
+
+	if result.EndTime.Before(result.StartTime) {
+		return false
+	}
+
+	if len(result.Steps) != 5 {
+		return false
+	}
+
+	// Verify all steps completed
+	for _, step := range result.Steps {
+		if step.Progress < 100 {
+			return false
+		}
+	}
+
+	return true
+}
diff --git a/weed/admin/plugin/workers/balance/schema.go b/weed/admin/plugin/workers/balance/schema.go
new file mode 100644
index 000000000..6168f67a8
--- /dev/null
+++ b/weed/admin/plugin/workers/balance/schema.go
@@ -0,0 +1,200 @@
+package balance
+
+import (
+	"encoding/json"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// ConfigurationSchema defines the schema for balance plugin configuration
+type ConfigurationSchema struct {
+	AdminConfig  AdminConfigSchema  `json:"admin_config"`
+	WorkerConfig WorkerConfigSchema `json:"worker_config"`
+}
+
+// AdminConfigSchema defines admin-side configuration
+type AdminConfigSchema struct {
+	RebalanceInterval          ConfigField `json:"rebalance_interval"`
+	MaxConcurrentJobs          ConfigField `json:"max_concurrent_jobs"`
+	JobTimeout                 ConfigField `json:"job_timeout"`
+	HealthCheckInterval        ConfigField `json:"health_check_interval"`
+	DiskUsageThreshold         ConfigField `json:"disk_usage_threshold"`
+	AcceptableImbalancePercent ConfigField `json:"acceptable_imbalance_percent"`
+}
+
+// WorkerConfigSchema defines worker-side configuration
+type WorkerConfigSchema struct {
+	MinVolumeSize              ConfigField `json:"min_volume_size"`
+	MaxVolumeSize              ConfigField `json:"max_volume_size"`
+	DataNodeCount              ConfigField `json:"data_node_count"`
+	ReplicationFactor          ConfigField `json:"replication_factor"`
+	PreferBalancedDistribution ConfigField `json:"prefer_balanced_distribution"`
+}
+
+// ConfigField describes a configuration field
+type ConfigField struct {
+	Name        string        `json:"name"`
+	Description string        `json:"description"`
+	Type        string        `json:"type"`
+	Required    bool          `json:"required"`
+	Default     interface{}   `json:"default,omitempty"`
+	Min         interface{}   `json:"min,omitempty"`
+	Max         interface{}   `json:"max,omitempty"`
+	Options     []interface{} `json:"options,omitempty"`
+	Unit        string        `json:"unit,omitempty"`
+}
+
+// GetConfigurationSchema returns the schema for balance plugin configuration
+func GetConfigurationSchema() *plugin_pb.PluginConfig {
+	schema := ConfigurationSchema{
+		AdminConfig: AdminConfigSchema{
+			RebalanceInterval: ConfigField{
+				Name:        "rebalance_interval",
+				Description: "Time between rebalancing scans",
+				Type:        "duration",
+				Required:    true,
+				Default:     "2h",
+				Min:         "10m",
+				Max:         "24h",
+				Unit:        "seconds",
+			},
+			MaxConcurrentJobs: ConfigField{
+				Name:        "max_concurrent_jobs",
+				Description: "Maximum concurrent rebalancing jobs",
+				Type:        "integer",
+				Required:    true,
+				Default:     3,
+				Min:         1,
+				Max:         10,
+			},
+			JobTimeout: ConfigField{
+				Name:        "job_timeout",
+				Description: "Timeout for individual rebalance jobs",
+				Type:        "duration",
+				Required:    true,
+				Default:     "24h",
+				Min:         "1h",
+				Max:         "72h",
+				Unit:        "seconds",
+			},
+			HealthCheckInterval: ConfigField{
+				Name:        "health_check_interval",
+				Description: "Health check interval",
+				Type:        "duration",
+				Required:    true,
+				Default:     "1m",
+				Min:         "10s",
+				Max:         "10m",
+				Unit:        "seconds",
+			},
+			DiskUsageThreshold: ConfigField{
+				Name:        "disk_usage_threshold",
+				Description: "Trigger rebalancing when disk usage exceeds this percentage",
+				Type:        "integer",
+				Required:    true,
+				Default:     85,
+				Min:         50,
+				Max:         95,
+				Unit:        "percent",
+			},
+			AcceptableImbalancePercent: ConfigField{
+				Name:        "acceptable_imbalance_percent",
+				Description: "Acceptable imbalance level before rebalancing",
+				Type:        "integer",
+				Required:    true,
+				Default:     10,
+				Min:         1,
+				Max:         50,
+				Unit:        "percent",
+			},
+		},
+		WorkerConfig: WorkerConfigSchema{
+			MinVolumeSize: ConfigField{
+				Name:        "min_volume_size",
+				Description: "Minimum volume size to consider for rebalancing",
+				Type:        "integer",
+				Required:    true,
+				Default:     500,
+				Min:         100,
+				Unit:        "MB",
+			},
+			MaxVolumeSize: ConfigField{
+				Name:        "max_volume_size",
+				Description: "Maximum volume size to consider for rebalancing",
+				Type:        "integer",
+				Required:    true,
+				Default:     50000,
+				Max:         500000,
+				Unit:        "MB",
+			},
+			DataNodeCount: ConfigField{
+				Name:        "data_node_count",
+				Description: "Expected number of data nodes in cluster",
+				Type:        "integer",
+				Required:    true,
+				Default:     5,
+				Min:         2,
+				Max:         1000,
+			},
+			ReplicationFactor: ConfigField{
+				Name:        "replication_factor",
+				Description: "Default replication factor for volumes",
+				Type:        "integer",
+				Required:    true,
+				Default:     2,
+				Min:         1,
+				Max:         5,
+			},
+			PreferBalancedDistribution: ConfigField{
+				Name:        "prefer_balanced_distribution",
+				Description: "Prefer balanced distribution over other factors",
+				Type:        "boolean",
+				Required:    true,
+				Default:     true,
+			},
+		},
+	}
+
+	data, _ := json.MarshalIndent(schema, "", "  ")
+
+	return &plugin_pb.PluginConfig{
+		PluginId: "balance-plugin",
+		Properties: map[string]string{
+			"schema":                       string(data),
+			"rebalance_interval":           "2h",
+			"max_concurrent_jobs":          "3",
+			"job_timeout":                  "24h",
+			"health_check_interval":        "1m",
+			"disk_usage_threshold":         "85",
+			"acceptable_imbalance_percent": "10",
+			"min_volume_size":              "500",
+			"max_volume_size":              "50000",
+			"data_node_count":              "5",
+			"replication_factor":           "2",
+			"prefer_balanced_distribution": "true",
+		},
+	}
+}
+
+// DefaultAdminConfig returns default admin configuration
+func DefaultAdminConfig() map[string]string {
+	return map[string]string{
+		"rebalance_interval":           "2h",
+		"max_concurrent_jobs":          "3",
+		"job_timeout":                  "24h",
+		"health_check_interval":        "1m",
+		"disk_usage_threshold":         "85",
+		"acceptable_imbalance_percent": "10",
+	}
+}
+
+// DefaultWorkerConfig returns default worker configuration
+func DefaultWorkerConfig() map[string]string {
+	return map[string]string{
+		"min_volume_size":              "500",
+		"max_volume_size":              "50000",
+		"data_node_count":              "5",
+		"replication_factor":           "2",
+		"prefer_balanced_distribution": "true",
+	}
+}
diff --git a/weed/admin/plugin/workers/balance/worker.go b/weed/admin/plugin/workers/balance/worker.go
new file mode 100644
index 000000000..a1cefdf2d
--- /dev/null
+++ b/weed/admin/plugin/workers/balance/worker.go
@@ -0,0 +1,342 @@
+package balance
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net"
+	"time"
+
+	"google.golang.org/grpc"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
+)
+
+// WorkerConfig holds worker-specific configuration for balance plugin
+type WorkerConfig struct {
+	WorkerID                   string
+	AdminHost                  string
+	AdminPort                  int
+	PluginPort                 int
+	MinVolumeSize              uint64
+	MaxVolumeSize              uint64
+	DataNodeCount              int
+	ReplicationFactor          int
+	PreferBalancedDistribution bool
+	RebalanceInterval          time.Duration
+	MaxConcurrentJobs          int
+	HealthCheckInterval        time.Duration
+	DiskUsageThreshold         float64
+	AcceptableImbalancePercent float64
+}
+
+// Worker represents the balance plugin worker
+type Worker struct {
+	config       *WorkerConfig
+	pluginClient plugin_pb.PluginServiceClient
+	conn         *grpc.ClientConn
+	detector     *Detector
+	executor     *Executor
+	activeJobs   map[string]*plugin_pb.ExecuteJobRequest
+	done         chan bool
+	isRunning    bool
+}
+
+// NewWorker creates a new balance worker
+func NewWorker(config *WorkerConfig) *Worker {
+	return &Worker{
+		config:     config,
+		activeJobs: make(map[string]*plugin_pb.ExecuteJobRequest),
+		done:       make(chan bool),
+	}
+}
+
+// Start initializes and starts the worker
+func (w *Worker) Start(ctx context.Context) error {
+	log.Printf("Starting balance worker: %s", w.config.WorkerID)
+
+	// Connect to admin server
+	if err := w.connectToAdmin(ctx); err != nil {
+		return fmt.Errorf("failed to connect to admin: %v", err)
+	}
+
+	// Initialize detector
+	w.detector = NewDetector(DetectionOptions{
+		MinVolumeSize:              w.config.MinVolumeSize,
+		MaxVolumeSize:              w.config.MaxVolumeSize,
+		DiskUsageThreshold:         w.config.DiskUsageThreshold,
+		AcceptableImbalancePercent: w.config.AcceptableImbalancePercent,
+		PreferBalancedDistribution: w.config.PreferBalancedDistribution,
+		DataNodeCount:              w.config.DataNodeCount,
+	})
+
+	// Initialize executor
+	w.executor = NewExecutor(&ExecutorConfig{
+		TimeoutPerStep: 1 * time.Hour,
+		MaxRetries:     3,
+		BatchSize:      10 * 1024 * 1024,
+	})
+
+	// Register with admin
+	if err := w.registerPlugin(ctx); err != nil {
+		return fmt.Errorf("failed to register: %v", err)
+	}
+
+	w.isRunning = true
+
+	// Start background goroutines
+	go w.heartbeatLoop(ctx)
+
+	log.Printf("Balance worker started successfully")
+	return nil
+}
+
+// connectToAdmin establishes connection to admin server
+func (w *Worker) connectToAdmin(ctx context.Context) error {
+	address := fmt.Sprintf("%s:%d", w.config.AdminHost, w.config.AdminPort)
+
+	dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	conn, err := grpc.DialContext(dialCtx, address, grpc.WithInsecure())
+	if err != nil {
+		return fmt.Errorf("failed to dial: %v", err)
+	}
+
+	w.conn = conn
+	w.pluginClient = plugin_pb.NewPluginServiceClient(conn)
+
+	return nil
+}
+
+// registerPlugin registers the plugin with the admin server
+func (w *Worker) registerPlugin(ctx context.Context) error {
+	schema := GetConfigurationSchema()
+
+	req := &plugin_pb.PluginConnectRequest{
+		PluginId:          w.config.WorkerID,
+		PluginName:        "balance-plugin",
+		Version:           "1.0.0",
+		Capabilities:      []string{"detect", "execute", "report_health"},
+		MaxConcurrentJobs: int32(w.config.MaxConcurrentJobs),
+		SupportsStreaming: true,
+		Port:              int32(w.config.PluginPort),
+	}
+
+	// Add capabilities detail
+	req.CapabilitiesDetail = &plugin_pb.PluginCapabilities{
+		Detection: []*plugin_pb.DetectionCapability{
+			{
+				Type:               "imbalance_detection",
+				Description:        "Detect imbalanced data distribution across nodes",
+				MinIntervalSeconds: int32(w.config.RebalanceInterval.Seconds()),
+				RequiresFullScan:   true,
+			},
+		},
+		Maintenance: []*plugin_pb.MaintenanceCapability{
+			{
+				Type:                     "rebalance_volume",
+				Description:              "Rebalance volume data across nodes",
+				RequiredDetectionTypes:   []string{"imbalance_detection"},
+				EstimatedDurationSeconds: 3600,
+			},
+		},
+	}
+
+	// Add schema to metadata
+	if schema != nil {
+		if req.Metadata == nil {
+			req.Metadata = make(map[string]string)
+		}
+		for k, v := range schema.Properties {
+			req.Metadata[k] = v
+		}
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	resp, err := w.pluginClient.Connect(ctx, req)
+	if err != nil {
+		return fmt.Errorf("connect RPC failed: %v", err)
+	}
+
+	if !resp.Success {
+		return fmt.Errorf("connect failed: %s", resp.Message)
+	}
+
+	log.Printf("Plugin registered with master: %s", resp.MasterId)
+	return nil
+}
+
+// heartbeatLoop sends periodic health reports
+func (w *Worker) heartbeatLoop(ctx context.Context) {
+	ticker := time.NewTicker(w.config.HealthCheckInterval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-w.done:
+			return
+		case <-ticker.C:
+			w.sendHealthReport(ctx)
+		}
+	}
+}
+
+// sendHealthReport sends a health report to the admin
+func (w *Worker) sendHealthReport(ctx context.Context) {
+	report := &plugin_pb.HealthReport{
+		PluginId:    w.config.WorkerID,
+		TimestampMs: time.Now().UnixMilli(),
+		Status:      plugin_pb.HealthStatus_HEALTH_STATUS_HEALTHY,
+		ActiveJobs:  int32(len(w.activeJobs)),
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+
+	_, err := w.pluginClient.ReportHealth(ctx, report)
+	if err != nil {
+		log.Printf("Failed to send health report: %v", err)
+	}
+}
+
+// ExecuteDetection performs detection for rebalancing candidates
+func (w *Worker) ExecuteDetection(ctx context.Context, nodeMetrics map[string]*NodeDiskMetric) ([]*RebalanceCandidate, error) {
+	return w.detector.DetectJobs(nodeMetrics)
+}
+
+// ExecuteJob executes a rebalancing job
+func (w *Worker) ExecuteJob(ctx context.Context, jobID string, payload *plugin_pb.JobPayload) error {
+	req := &plugin_pb.ExecuteJobRequest{
+		JobId:      jobID,
+		JobType:    "rebalance_volume",
+		Payload:    payload,
+		RetryCount: 0,
+	}
+
+	w.activeJobs[jobID] = req
+
+	defer delete(w.activeJobs, jobID)
+
+	// Execute the job
+	result, err := w.executor.ExecuteJob(req)
+	if err != nil {
+		log.Printf("Job execution failed: %v", err)
+		return err
+	}
+
+	if result.Success {
+		log.Printf("Job %s completed successfully", jobID)
+		return w.submitResult(ctx, jobID, result)
+	}
+
+	log.Printf("Job %s failed: %s", jobID, result.ErrorMessage)
+	return fmt.Errorf(result.ErrorMessage)
+}
+
+// submitResult submits job results to admin
+func (w *Worker) submitResult(ctx context.Context, jobID string, result *BalanceExecutionResult) error {
+	jobResult := &plugin_pb.JobResult{
+		Success:  result.Success,
+		Metadata: result.Metadata,
+	}
+
+	req := &plugin_pb.JobResultRequest{
+		JobId:          jobID,
+		JobType:        "rebalance_volume",
+		Status:         plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED,
+		Message:        "Rebalancing completed successfully",
+		Result:         jobResult,
+		RetryCountUsed: 0,
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+
+	_, err := w.pluginClient.SubmitResult(ctx, req)
+	return err
+}
+
+// Stop gracefully stops the worker
+func (w *Worker) Stop(ctx context.Context) error {
+	log.Printf("Stopping balance worker")
+	w.isRunning = false
+	close(w.done)
+
+	if w.conn != nil {
+		return w.conn.Close()
+	}
+
+	return nil
+}
+
+// GetStatus returns the current worker status
+func (w *Worker) GetStatus() map[string]interface{} {
+	return map[string]interface{}{
+		"worker_id":          w.config.WorkerID,
+		"is_running":         w.isRunning,
+		"active_jobs":        len(w.activeJobs),
+		"admin_connected":    w.conn != nil,
+		"data_node_count":    w.config.DataNodeCount,
+		"replication_factor": w.config.ReplicationFactor,
+	}
+}
+
+// ParseFlags parses command line flags for balance worker
+func ParseFlags() *WorkerConfig {
+	config := &WorkerConfig{
+		WorkerID:                   "balance-worker-1",
+		AdminHost:                  "localhost",
+		AdminPort:                  50051,
+		PluginPort:                 50053,
+		MinVolumeSize:              500,
+		MaxVolumeSize:              50000,
+		DataNodeCount:              5,
+		ReplicationFactor:          2,
+		PreferBalancedDistribution: true,
+		RebalanceInterval:          2 * time.Hour,
+		MaxConcurrentJobs:          3,
+		HealthCheckInterval:        1 * time.Minute,
+		DiskUsageThreshold:         85,
+		AcceptableImbalancePercent: 10,
+	}
+
+	flag.StringVar(&config.WorkerID, "worker-id", config.WorkerID, "Worker ID")
+	flag.StringVar(&config.AdminHost, "admin-host", config.AdminHost, "Admin server host")
+	flag.IntVar(&config.AdminPort, "admin-port", config.AdminPort, "Admin server port")
+	flag.IntVar(&config.PluginPort, "plugin-port", config.PluginPort, "Plugin server port")
+	flag.Uint64Var(&config.MinVolumeSize, "min-volume-size", config.MinVolumeSize, "Minimum volume size in MB")
+	flag.Uint64Var(&config.MaxVolumeSize, "max-volume-size", config.MaxVolumeSize, "Maximum volume size in MB")
+	flag.IntVar(&config.DataNodeCount, "data-node-count", config.DataNodeCount, "Expected data node count")
+	flag.IntVar(&config.ReplicationFactor, "replication-factor", config.ReplicationFactor, "Replication factor")
+	flag.BoolVar(&config.PreferBalancedDistribution, "prefer-balanced", config.PreferBalancedDistribution, "Prefer balanced distribution")
+	flag.DurationVar(&config.RebalanceInterval, "rebalance-interval", config.RebalanceInterval, "Rebalance interval")
+	flag.IntVar(&config.MaxConcurrentJobs, "max-concurrent-jobs", config.MaxConcurrentJobs, "Max concurrent jobs")
+	flag.DurationVar(&config.HealthCheckInterval, "health-check-interval", config.HealthCheckInterval, "Health check interval")
+	flag.Float64Var(&config.DiskUsageThreshold, "disk-usage-threshold", config.DiskUsageThreshold, "Disk usage threshold percent")
+	flag.Float64Var(&config.AcceptableImbalancePercent, "acceptable-imbalance", config.AcceptableImbalancePercent, "Acceptable imbalance percent")
+
+	flag.Parse()
+
+	return config
+}
+
+// ListenAndServe starts the gRPC server for the worker
+func (w *Worker) ListenAndServe(port int) error {
+	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
+	if err != nil {
+		return fmt.Errorf("failed to listen on port %d: %v", port, err)
+	}
+
+	server := grpc.NewServer()
+	// Register plugin service handlers here
+	// plugin_pb.RegisterPluginServiceServer(server, w)
+
+	log.Printf("Worker listening on port %d", port)
+	return server.Serve(listener)
+}