Horovod with Spark - Job Not Distributing Across Worker Nodes
Problem Description: Horovod with Spark - Job Not Distributing Across Worker Nodes
Environment:
Cluster Setup: 1 Master Node, 2 Worker Nodes Software Versions: Horovod: >= 0.19.0 TensorFlow: >= 1.12.0 Spark: >= 2.3.2 Python: 3.x MPI Version: Open MPI 4.0.5 Deployment Mode: YARN
Issue Summary: I am experiencing an issue where my distributed training job using Horovod on Spark is not properly utilizing the worker nodes in my cluster. Instead, all computation appears to be executed on the master node, leading to resource exhaustion on the master while the worker nodes remain idle.
Details:
I configured my Spark and Horovod environment following the official Horovod documentation. My setup involves one master node and two worker nodes, with the master node not participating as a worker (confirmed via the Hadoop UI). The job is submitted using mpirun with a spark-submit command embedded within it.
Symptoms:
The master node (lila) shows high CPU and memory usage, almost to the point of exhaustion.
Worker nodes (worker1, worker2) show no significant CPU or memory usage.
Both Hadoop and Spark UIs confirm that only the master node is active during the job execution.
The custom callback to print the hostname confirms that only the master node is processing the data.
Commands Used: Here is the mpirun command used to submit the job:
mpirun -np 4 -bind-to none -map-by slot -x NCCL_DEBUG=INFO -x LD_LIBRARY_PATH -x PATH -mca pml ob1 -mca btl ^openib \spark-submit --master yarn --deploy-mode cluster --conf spark.executor.cores=4 --conf spark.executor.instances=2 --conf spark.driver.memory=4g --conf spark.executor.memory=6g --conf spark.dynamicAllocation.enabled=false --conf spark.yarn.maxAppAttempts=1 codes/estimator_example.py > output/estimator_example.txt
Code Snippet: Here is a simplified version of the code being used:
from tensorflow import keras
import tensorflow as tf
import horovod.spark.keras as hvd
from pyspark.sql import SparkSession
import numpy as np
import os
import socket
# Initialize Horovod
hvd.init()
# Set up Spark session
spark = SparkSession.builder.appName("HorovodOnSparkExample").getOrCreate()
# Generate random data
def generate_data(num_samples):
num_features = 2
data = np.random.rand(num_samples, num_features)
labels = (data[:, 0] + data[:, 1] > 1).astype(int)
return spark.createDataFrame([(float(x[0]), float(x[1]), int(y)) for x, y in zip(data, labels)], ["feature1", "feature2", "label"])
train_df = generate_data(1000)
test_df = generate_data(200)
# Build a simple Keras model
model = keras.models.Sequential([
keras.layers.Dense(8, input_dim=2, activation='tanh'),
keras.layers.Dense(1, activation='sigmoid')
])
# Optimizer and loss
optimizer = keras.optimizers.SGD(learning_rate=0.1)
loss = 'binary_crossentropy'
# Store for checkpointing
store = hvd.spark.common.store.HDFSStore('/user/username/experiments')
# Define the KerasEstimator
keras_estimator = hvd.KerasEstimator(
num_proc=4,
store=store,
model=model,
optimizer=optimizer,
loss=loss,
feature_cols=['feature1', 'feature2'],
label_cols=['label'],
batch_size=32,
epochs=10
)
# Fit the model
keras_model = keras_estimator.fit(train_df).setOutputCols(['predict'])
# Transform the test data
predict_df = keras_model.transform(test_df)
predict_df.show()
# Custom callback to log worker information
class WorkerInfoCallback(tf.keras.callbacks.Callback):
def on_epoch_end(self, epoch, logs=None):
hostname = socket.gethostname()
rank = hvd.rank()
print(f"Epoch {epoch} ended. Worker rank: {rank}, Hostname: {hostname}")
# Enable Horovod timeline
os.environ["HOROVOD_TIMELINE"] = "/home/hadoop/horovod_timeline_gan.json"
os.environ["HOROVOD_TIMELINE_MARK_CYCLES"] = "1"
# Add the custom callback to the list of callbacks
callbacks = [
hvd.callbacks.BroadcastGlobalVariablesCallback(0),
hvd.callbacks.MetricAverageCallback(),
WorkerInfoCallback()
]
# Train the model
keras_model.fit(x_train, y_train,
batch_size=128,
callbacks=callbacks,
epochs=2,
verbose=2 if hvd.rank() == 0 else 0,
validation_data=(x_test, y_test))
# Save the model
if hvd.rank() == 0:
keras_model.save('/home/hadoop/keras_model.h5')
# Stop Spark session
spark.stop()
Questions:
How can I ensure that the training job is properly distributed across the worker nodes? Are there any additional configurations or steps required to ensure that mpirun properly utilizes the worker nodes? Are there specific debugging steps I should follow to identify why the worker nodes are not being utilized?
Any insights or suggestions to resolve this issue would be greatly appreciated.
Thank you!
Source: horovod/horovod