Store MLflow Artifacts in HDFS

pip install mlflow scikit-learn pyarrow

export HADOOP_HOME=/usr/odp/3.3.6.3-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.3-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.3-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.")