Store MLflow Artifacts in HDFS

pip install mlflow scikit-learn pyarrow

export HADOOP_HOME=/usr/odp/3.3.6.2-1/hadoop

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

export LD_LIBRARY_PATH=$HADOOP_HOME/lib/native:$LD_LIBRARY_PATH

export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath --glob)

mlflow server \

  --backend-store-uri mysql+pymysql://<username>:<password>@<mysql-host>:<port>/<database-name> \

  --default-artifact-root hdfs://<hdfs-host>:<port>/<artifact-path> \

  --host <IP Address> \

  --port 5000

cat hdfstest.py 

# 1. Setup

import mlflow

import mlflow.sklearn

from sklearn.linear_model import LinearRegression

from sklearn.metrics import mean_squared_error, r2_score

from sklearn.model_selection import train_test_split

from sklearn.datasets import make_regression

import pandas as pd

import numpy as np

import os

import matplotlib.pyplot as plt

​

# 2. Configure MLflow Tracking URI (point to your MLflow server)

mlflow.set_tracking_uri("http://10.100.11.39:5000")

mlflow.set_experiment("HDFS_LR_Experiment")

​

# 3. Generate dummy regression data

X, y = make_regression(n_samples=1000, n_features=2, noise=15, random_state=42)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

​

# 4. Start MLflow Run

with mlflow.start_run():

    # Log parameters

    mlflow.log_param("fit_intercept", True)

​

    # Train model

    model = LinearRegression(fit_intercept=True)

    model.fit(X_train, y_train)

​

    # Predict and evaluate

    predictions = model.predict(X_test)

    mse = mean_squared_error(y_test, predictions)

    r2 = r2_score(y_test, predictions)

​

    # Log metrics

    mlflow.log_metric("mse", mse)

    mlflow.log_metric("r2", r2)

​

    # Log model

    mlflow.sklearn.log_model(model, "model")

​

    # Save and log a plot

    plt.figure(figsize=(6, 6))

    plt.scatter(y_test, predictions, alpha=0.7)

    plt.xlabel("Actual")

    plt.ylabel("Predicted")

    plt.title("Actual vs Predicted")

    plot_path = "actual_vs_predicted.png"

    plt.savefig(plot_path)

    mlflow.log_artifact(plot_path)

​

    print("Run complete. Artifacts saved to HDFS.")

python /usr/odp/3.3.6.2-1/mlflow/hdfstest.py

hdfs dfs -ls /tmp/4

export MLFLOW_KERBEROS_TICKET_CACHE=/tmp/krb5cc_1025

export MLFLOW_KERBEROS_TICKET_CACHE=/tmp/krb5cc_$(id -u)

ls

bin  include  lib  lib64  pyvenv.cfg  share

[root@kafkaingestion2 mlflow]# 

[root@kafkaingestion2 mlflow]# 

[root@kafkaingestion2 mlflow]# source bin/activate

export HADOOP_HOME=/usr/odp/3.3.6.2-1/hadoop

export HADOOP_CONF_DIR=$HADOOP_HOME/etc/hadoop

export LD_LIBRARY_PATH=$HADOOP_HOME/lib/native:$LD_LIBRARY_PATH

export CLASSPATH=$($HADOOP_HOME/bin/hadoop classpath --glob)

cat hdfs.py 

import mlflow

import mlflow.sklearn

from sklearn.linear_model import LinearRegression

from sklearn.metrics import mean_squared_error, r2_score

from sklearn.model_selection import train_test_split

from sklearn.datasets import make_regression

import matplotlib.pyplot as plt

import os

import time

​

def log(msg):

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)

​

# 1. Setup MLflow tracking

log("Setting MLflow tracking URI...")

mlflow.set_tracking_uri("http://10.100.11.72:5000")

​

experiment_name = "HDFS_LR_Experiment_test"

artifact_location = "hdfs://kafkaingestion.acceldata.ce:8020/tmp/mlflow-jupyter"

​

# 2. Create experiment (safe creation)

log(f"Trying to create experiment '{experiment_name}' with artifact location '{artifact_location}'...")

try:

    mlflow.create_experiment(name=experiment_name, artifact_location=artifact_location)

    log("Experiment created successfully.")

except mlflow.exceptions.MlflowException as e:

    log(f"Experiment already exists or failed to create: {str(e)}")

​

log(f"Setting experiment to '{experiment_name}'...")

mlflow.set_experiment(experiment_name)

​

# 3. Generate data

log("Generating regression dataset...")

X, y = make_regression(n_samples=1000, n_features=2, noise=15, random_state=42)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

​

# 4. Start MLflow run

log("Starting MLflow run...")

with mlflow.start_run() as run:

    log("Logging parameters...")

    mlflow.log_param("fit_intercept", True)

​

    log("Training LinearRegression model...")

    model = LinearRegression(fit_intercept=True)

    model.fit(X_train, y_train)

​

    log("Generating predictions and calculating metrics...")

    predictions = model.predict(X_test)

    mse = mean_squared_error(y_test, predictions)

    r2 = r2_score(y_test, predictions)

​

    log(f"Logging metrics: mse={mse}, r2={r2}...")

    mlflow.log_metric("mse", mse)

    mlflow.log_metric("r2", r2)

​

    log("Logging model...")

    mlflow.sklearn.log_model(model, "model")

​

    log("Generating and logging plot...")

    plt.figure(figsize=(6, 6))

    plt.scatter(y_test, predictions, alpha=0.7)

    plt.xlabel("Actual")

    plt.ylabel("Predicted")

    plt.title("Actual vs Predicted")

    plot_path = "actual_vs_predicted.png"

    plt.savefig(plot_path)

    mlflow.log_artifact(plot_path)

​

    log("MLflow run complete. Artifacts should be saved to HDFS.")

import os

import subprocess

​

# Set environment variables

os.environ['HADOOP_HOME'] = '/usr/odp/3.3.6.2-1/hadoop'

os.environ['HADOOP_CONF_DIR'] = '/usr/odp/3.3.6.2-1/hadoop/etc/hadoop'

os.environ['LD_LIBRARY_PATH'] = '/usr/odp/3.3.6.2-1/hadoop/lib/native:/usr/lib:/usr/lib64'

​

# Get evaluated CLASSPATH via subprocess and assign it

classpath = subprocess.check_output(

    ['/usr/odp/3.3.6.2-1/hadoop/bin/hadoop', 'classpath', '--glob'],

    text=True

).strip()

os.environ['CLASSPATH'] = classpath

​

# Confirm

print("CLASSPATH:", os.environ['CLASSPATH'])

import mlflow

import mlflow.sklearn

from sklearn.linear_model import LinearRegression

from sklearn.metrics import mean_squared_error, r2_score

from sklearn.model_selection import train_test_split

from sklearn.datasets import make_regression

import matplotlib.pyplot as plt

import os

import time

​

def log(msg):

    print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}", flush=True)

​

# 1. Setup MLflow tracking

log("Setting MLflow tracking URI...")

mlflow.set_tracking_uri("http://10.100.11.72:5000")

​

experiment_name = "HDFS_LR_Experiment_test"

artifact_location = "hdfs://kafkaingestion.acceldata.ce:8020/tmp/mlflow-jupyter"

​

# 2. Create experiment (safe creation)

log(f"Trying to create experiment '{experiment_name}' with artifact location '{artifact_location}'...")

try:

    mlflow.create_experiment(name=experiment_name, artifact_location=artifact_location)

    log("Experiment created successfully.")

except mlflow.exceptions.MlflowException as e:

    log(f"Experiment already exists or failed to create: {str(e)}")

​

log(f"Setting experiment to '{experiment_name}'...")

mlflow.set_experiment(experiment_name)

​

# 3. Generate data

log("Generating regression dataset...")

X, y = make_regression(n_samples=1000, n_features=2, noise=15, random_state=42)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

​

# 4. Start MLflow run

log("Starting MLflow run...")

with mlflow.start_run() as run:

    log("Logging parameters...")

    mlflow.log_param("fit_intercept", True)

​

    log("Training LinearRegression model...")

    model = LinearRegression(fit_intercept=True)

    model.fit(X_train, y_train)

​

    log("Generating predictions and calculating metrics...")

    predictions = model.predict(X_test)

    mse = mean_squared_error(y_test, predictions)

    r2 = r2_score(y_test, predictions)

​

    log(f"Logging metrics: mse={mse}, r2={r2}...")

    mlflow.log_metric("mse", mse)

    mlflow.log_metric("r2", r2)

​

    log("Logging model...")

    mlflow.sklearn.log_model(model, "model")

​

    log("Generating and logging plot...")

    plt.figure(figsize=(6, 6))

    plt.scatter(y_test, predictions, alpha=0.7)

    plt.xlabel("Actual")

    plt.ylabel("Predicted")

    plt.title("Actual vs Predicted")

    plot_path = "actual_vs_predicted.png"

    plt.savefig(plot_path)

    mlflow.log_artifact(plot_path)

​

    log("MLflow run complete. Artifacts should be saved to HDFS.")

import os

import subprocess

​

# Set environment variables

os.environ['HADOOP_HOME'] = '/usr/odp/3.3.6.2-1/hadoop'

os.environ['HADOOP_CONF_DIR'] = '/usr/odp/3.3.6.2-1/hadoop/etc/hadoop'

os.environ['LD_LIBRARY_PATH'] = '/usr/odp/3.3.6.2-1/hadoop/lib/native:/usr/lib:/usr/lib64'

​

# Get evaluated CLASSPATH via subprocess and assign it

classpath = subprocess.check_output(

    ['/usr/odp/3.3.6.2-1/hadoop/bin/hadoop', 'classpath', '--glob'],

    text=True

).strip()

os.environ['CLASSPATH'] = classpath

​

# Confirm

print("CLASSPATH:", os.environ['CLASSPATH'])

hdfs dfs -mkdir -p /tmp/mlflow-jupyter

hdfs dfs -chown -R jupyterhub:hdfs /tmp/mlflow-jupyter

hdfs dfs -chmod -R 777 /tmp

import mlflow

import mlflow.sklearn

from sklearn.linear_model import LinearRegression

from sklearn.metrics import mean_squared_error, r2_score

from sklearn.model_selection import train_test_split

from sklearn.datasets import make_regression

import matplotlib.pyplot as plt

import os

​

# Set MLflow tracking server

mlflow.set_tracking_uri("http://10.100.11.39:5000")

​

# Ensure experiment writes to HDFS under /tmp

experiment_name = "HDFS_LR_Experiment"

artifact_location = "hdfs://airflowdb.acceldata.ce:8020/tmp/mlflow-jupyter"

​

# Create experiment

try:

    mlflow.create_experiment(name="HDFS_LR_Experiment", artifact_location=artifact_location)

except mlflow.exceptions.MlflowException:

    pass  # Already exists

​

mlflow.set_experiment("HDFS_LR_Experiment")

​

​

# Create experiment if not exists

try:

    mlflow.create_experiment(name=experiment_name, artifact_location=artifact_location)

except mlflow.exceptions.MlflowException:

    pass  # Already exists

​

mlflow.set_experiment(experiment_name)

​

# Generate dummy data

X, y = make_regression(n_samples=1000, n_features=2, noise=15, random_state=42)

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

​

# Start MLflow run

with mlflow.start_run():

    mlflow.log_param("fit_intercept", True)

​

    model = LinearRegression(fit_intercept=True)

    model.fit(X_train, y_train)

​

    predictions = model.predict(X_test)

    mse = mean_squared_error(y_test, predictions)

    r2 = r2_score(y_test, predictions)

​

    mlflow.log_metric("mse", mse)

    mlflow.log_metric("r2", r2)

​

    # Log model

    mlflow.sklearn.log_model(model, "model")

​

    # Plot and log artifact

    plt.figure(figsize=(6, 6))

    plt.scatter(y_test, predictions, alpha=0.7)

    plt.xlabel("Actual")

    plt.ylabel("Predicted")

    plt.title("Actual vs Predicted")

    plot_path = "actual_vs_predicted.png"

    plt.savefig(plot_path)

    mlflow.log_artifact(plot_path)

​

    print("Run complete. Artifacts saved to HDFS at /tmp/mlflow-artifacts.")

hdfs dfs -ls /tmp/4

hdfs dfs -ls /tmp/mlflow-jupyter

/tmp/4/<run-id>/artifacts/model/

/tmp/mlflow-jupyter/<experiment-name>/