diff --git a/intermediate_notebooks/E2E/mortgage/mortgage.ipynb b/intermediate_notebooks/E2E/mortgage/mortgage.ipynb
new file mode 100644
index 00000000..9d99bfd5
--- /dev/null
+++ b/intermediate_notebooks/E2E/mortgage/mortgage.ipynb
@@ -0,0 +1,1198 @@
+{
+ "cells": [
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "# Mortgage Workflow\n",
+ "\n",
+ "## The Dataset\n",
+ "The dataset used with this workflow is derived from [Fannie Mae’s Single-Family Loan Performance Data](http://www.fanniemae.com/portal/funding-the-market/data/loan-performance-data.html) with all rights reserved by Fannie Mae. This processed dataset is redistributed with permission and consent from Fannie Mae.\n",
+ "\n",
+ "To acquire this dataset, please visit [RAPIDS Datasets Homepage](https://docs.rapids.ai/datasets/mortgage-data). We've also added some cells below to make it easier to download the data, set parameters, and run the notebook.\n",
+ "\n",
+ "## Introduction\n",
+ "The Mortgage workflow is composed of three core phases:\n",
+ "\n",
+ "1. ETL - Extract, Transform, Load\n",
+ "2. Data Conversion\n",
+ "3. ML - Training\n",
+ "\n",
+ "### ETL\n",
+ "Data is \n",
+ "1. Read in from storage\n",
+ "2. Transformed to emphasize key features\n",
+ "3. Loaded into volatile memory for conversion\n",
+ "\n",
+ "### Data Conversion\n",
+ "Features are\n",
+ "1. Broken into (labels, data) pairs\n",
+ "2. Distributed across many workers\n",
+ "3. Converted into compressed sparse row (CSR) matrix format for XGBoost\n",
+ "\n",
+ "### Machine Learning\n",
+ "The CSR data is fed into a distributed training session with Dask-XGBoost"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Imports statements"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 1,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "env: NCCL_P2P_DISABLE=1 # Necessary for NCCL < 2.4\n"
+ ]
+ }
+ ],
+ "source": [
+ "%env NCCL_P2P_DISABLE=1 # Necessary for NCCL < 2.4"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 2,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "import dask_xgboost as dxgb_gpu\n",
+ "import dask\n",
+ "import dask_cudf\n",
+ "from dask_cuda import LocalCUDACluster\n",
+ "from dask.delayed import delayed\n",
+ "from dask.distributed import Client, wait\n",
+ "import cudf\n",
+ "\n",
+ "import pynvml\n",
+ "import numpy as np\n",
+ "import xgboost as xgb\n",
+ "\n",
+ "from collections import OrderedDict\n",
+ "import gc\n",
+ "from glob import glob\n",
+ "import os"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "## Set up RMM and Dask_cudf cluster"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "While `dask_cudf` allows RAPIDS to use all the GPUs in the cluster as a single, large GPU, if the data is far larger than your current dask cluster size, you will have to use `RMM`. RMM requires you to declare the number of workers RMM defaults to 50% of your GPU memory size. However, you are able to manage the memory to use up to 100% and that may help your performance, and maybe some out of memory errors. Below are some 90%+ values that you can use to set your `initial_pool_size`. They are commented out, as is setting the inital pool size. Uncomment the one you want to use."
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Define your GPU Memory Allocation"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 4,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "# Decide initial pool size based on the memory size of the smallest GPU in your cluster and uncomment that line, as well as the initial_pool_size in the cell after\n",
+ "# ips = 6<<30 #8gb card\n",
+ "# ips = 14<<30 #16gb card\n",
+ "# ips = 22<<30 #24gb card\n",
+ "# ips = 29<<30 #32gb card\n",
+ "# ips = 43<<30 #48gb card\n"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Define the number of GPU workers that you have"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 5,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "n_workers = 4 # Please change your n_workers amount to the number of GPUs you have"
+ ]
+ },
+ {
+ "cell_type": "markdown",
+ "metadata": {},
+ "source": [
+ "### Set up RMM pool and start Dask cluster"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 6,
+ "metadata": {},
+ "outputs": [],
+ "source": [
+ "def initialize_rmm_pool():\n",
+ " import rmm\n",
+ " rmm.reinitialize(pool_allocator=True, \n",
+ " #initial_pool_size=ips, #RMM defaults to 50% GPU memory. You may get further performance by increasing the pool size per GPU. Do not go above your smallest GPU mem size\n",
+ " managed_memory=True)\n",
+ "\n",
+ "def initialize_rmm_no_pool():\n",
+ " import rmm\n",
+ "\n",
+ " rmm.reinitialize(pool_allocator=False, managed_memory=True)"
+ ]
+ },
+ {
+ "cell_type": "code",
+ "execution_count": 7,
+ "metadata": {},
+ "outputs": [
+ {
+ "name": "stdout",
+ "output_type": "stream",
+ "text": [
+ "LocalCUDACluster('tcp://127.0.0.1:43807', workers=4, threads=4, memory=270.40 GB)\n"
+ ]
+ },
+ {
+ "data": {
+ "text/html": [
+ "