Chapter 02
🛠️ Setup
NotebookPython 3 (ipykernel)243 cells
This notebooks contains the code for the 🔢 Data and 📈 Modeling lessons. After this proof of concept (PoC), we'll be moving all of this code to Python scripts to serve our application to production. Follow the accompanying lessons along with the code here to develop a deeper understanding of all the concepts.
🛠️ Setup
We'll be using Ray to develop our application using distributed workloads.
In [ ]python · cell 6
python
import os
import rayIn [ ]python · cell 7
python
import sys; sys.path.append("..")
import warnings; warnings.filterwarnings("ignore")
from dotenv import load_dotenv; load_dotenv()
%load_ext autoreload
%autoreload 2In [ ]python · cell 8
python
# Initialize Ray
if ray.is_initialized():
ray.shutdown()
ray.init()Output
2023-12-07 11:26:30,445 INFO worker.py:1633 -- Started a local Ray instance. View the dashboard at [1m[32m127.0.0.1:8265 [39m[22m
RayContext(dashboard_url='127.0.0.1:8265', python_version='3.10.11', ray_version='2.7.0', ray_commit='b4bba4717f5ba04ee25580fe8f88eed63ef0c5dc', protocol_version=None)
| Python version: | 3.10.11 |
| Ray version: | 2.7.0 |
| Dashboard: | http://127.0.0.1:8265 |
In [ ]python · cell 9
python
ray.cluster_resources()Output
{'memory': 30507458560.0,
'CPU': 12.0,
'node:__internal_head__': 1.0,
'node:127.0.0.1': 1.0,
'object_store_memory': 2147483648.0}These cluster resources only reflect our head node (m5.2xlarge). But recall in our setup lesson that our compute configuration that we also added g4dn.xlarge worker nodes (each has 1 GPU and 4 CPU) to our cluster. But because we set min_workers=0, our worker nodes will autoscale ( up to max_workers) as they're needed for specific workloads (ex. training).
In [ ]python · cell 11
python
# Workers (1 g4dn.xlarge)
num_workers = 1
resources_per_worker={"CPU": 3, "GPU": 1}If you are running this on a local laptop (no GPU), use the CPU count from ray.cluster_resources() to set your resources. For example if your machine has 10 CPUs:
python
num_workers = 6 # prefer to do a few less than total available CPU (1 for head node + 1 for background tasks)
resources_per_worker={"CPU": 1, "GPU": 0}In [ ]python · cell 13
python
# Storage
EFS_DIR = f"/efs/shared_storage/madewithml/{os.environ['GITHUB_USERNAME']}"
print (EFS_DIR)Output
/efs/shared_storage/madewithml/GokuMohandas
Data
🔢 Data ingestion
In [ ]python · cell 16
python
import pandas as pdIn [ ]python · cell 17
python
# Data ingestion
DATASET_LOC = "https://raw.githubusercontent.com/GokuMohandas/Made-With-ML/main/datasets/dataset.csv"
df = pd.read_csv(DATASET_LOC)
df.head()Output
id created_on title
0 6 2020-02-20 06:43:18 Comparison between YOLO and RCNN on real world... \
1 7 2020-02-20 06:47:21 Show, Infer & Tell: Contextual Inference for C...
2 9 2020-02-24 16:24:45 Awesome Graph Classification
3 15 2020-02-28 23:55:26 Awesome Monte Carlo Tree Search
4 25 2020-03-07 23:04:31 AttentionWalk
description tag
0 Bringing theory to experiment is cool. We can ... computer-vision
1 The beauty of the work lies in the way it arch... computer-vision
2 A collection of important graph embedding, cla... other
3 A curated list of Monte Carlo tree search pape... other
4 A PyTorch Implementation of "Watch Your Step: ... other
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| id | created_on | title | description | tag | |
|---|---|---|---|---|---|
| 0 | 6 | 2020-02-20 06:43:18 | Comparison between YOLO and RCNN on real world... | Bringing theory to experiment is cool. We can ... | computer-vision |
| 1 | 7 | 2020-02-20 06:47:21 | Show, Infer & Tell: Contextual Inference for C... | The beauty of the work lies in the way it arch... | computer-vision |
| 2 | 9 | 2020-02-24 16:24:45 | Awesome Graph Classification | A collection of important graph embedding, cla... | other |
| 3 | 15 | 2020-02-28 23:55:26 | Awesome Monte Carlo Tree Search | A curated list of Monte Carlo tree search pape... | other |
| 4 | 25 | 2020-03-07 23:04:31 | AttentionWalk | A PyTorch Implementation of "Watch Your Step: ... | other |
✂️ Data splitting
In [ ]python · cell 19
python
from sklearn.model_selection import train_test_splitIn [ ]python · cell 20
python
# Value counts
df.tag.value_counts()Output
tag natural-language-processing 310 computer-vision 285 other 106 mlops 63 Name: count, dtype: int64
In [ ]python · cell 21
python
# Split dataset
test_size = 0.2
train_df, val_df = train_test_split(df, stratify=df.tag, test_size=test_size, random_state=1234)In [ ]python · cell 22
python
# Train value counts
train_df.tag.value_counts()Output
tag natural-language-processing 248 computer-vision 228 other 85 mlops 50 Name: count, dtype: int64
In [ ]python · cell 23
python
# Validation (adjusted) value counts
val_df.tag.value_counts() * int((1-test_size) / test_size)Output
tag natural-language-processing 248 computer-vision 228 other 84 mlops 52 Name: count, dtype: int64
🔍 Exploratory Data Analysis (EDA)
Exploratory data analysis to understand the signals and nuances of our dataset. It's a cyclical process that can be done at various points of our development process (before/after labeling, preprocessing, etc.) depending on how well the problem is defined.
In [ ]python · cell 26
python
from collections import Counter
import matplotlib.pyplot as plt
import seaborn as sns; sns.set_theme()
import warnings; warnings.filterwarnings("ignore")
from wordcloud import WordCloud, STOPWORDSIn [ ]python · cell 27
python
# Most common tags
all_tags = Counter(df.tag)
all_tags.most_common()Output
[('natural-language-processing', 310),
('computer-vision', 285),
('other', 106),
('mlops', 63)]In [ ]python · cell 28
python
# Plot tag frequencies
tags, tag_counts = zip(*all_tags.most_common())
plt.figure(figsize=(10, 3))
ax = sns.barplot(x=list(tags), y=list(tag_counts))
ax.set_xticklabels(tags, rotation=0, fontsize=12)
plt.title("Tag distribution", fontsize=16)
plt.ylabel("# of projects", fontsize=14)
plt.show()Output
<Figure size 1000x300 with 1 Axes>
We'll address the data imbalance after splitting into our train split and prior to training our model.
In [ ]python · cell 30
python
# Most frequent tokens for each tag
tag="natural-language-processing"
plt.figure(figsize=(10, 3))
subset = df[df.tag==tag]
text = subset.title.values
cloud = WordCloud(
stopwords=STOPWORDS, background_color="black", collocations=False,
width=500, height=300).generate(" ".join(text))
plt.axis("off")
plt.imshow(cloud)Output
<matplotlib.image.AxesImage at 0x31f668c10>
<Figure size 1000x300 with 1 Axes>
[省略较大 image/png 输出]
Looks like the title text feature has some good signal for the respective classes and matches our intuition. We can repeat this for the description text feature as well. This information will become useful when we decide how to use our features for modeling.
✨ Data Preprocessing
Preprocessing the data via feature engineering, filtering and cleaning. Certain preprocessing steps are global (don't depend on our dataset, ex. lower casing text, removing stop words, etc.) and others are local (constructs are learned only from the training split, ex. vocabulary, standardization, etc.). For the local, dataset-dependent preprocessing steps, we want to ensure that we split the data first before preprocessing to avoid data leaks.
In [ ]python · cell 34
python
import json
import nltk
from nltk.corpus import stopwords
from nltk.stem import PorterStemmer
import reFeature engineering
We can combine existing input features to create new meaningful signal (helping the model learn).
In [ ]python · cell 37
python
# Feature engineering
df["text"] = df.title + " " + df.descriptionClean text
In [ ]python · cell 39
python
nltk.download("stopwords")
STOPWORDS = stopwords.words("english")Output
[nltk_data] Downloading package stopwords to /Users/goku/nltk_data... [nltk_data] Package stopwords is already up-to-date!
In [ ]python · cell 40
python
def clean_text(text, stopwords=STOPWORDS):
"""Clean raw text string."""
# Lower
text = text.lower()
# Remove stopwords
pattern = re.compile(r'\b(' + r"|".join(stopwords) + r")\b\s*")
text = pattern.sub('', text)
# Spacing and filters
text = re.sub(r"([!\"'#$%&()*\+,-./:;<=>?@\\\[\]^_`{|}~])", r" \1 ", text) # add spacing
text = re.sub("[^A-Za-z0-9]+", " ", text) # remove non alphanumeric chars
text = re.sub(" +", " ", text) # remove multiple spaces
text = text.strip() # strip white space at the ends
text = re.sub(r"http\S+", "", text) # remove links
return textIn [ ]python · cell 41
python
# Apply to dataframe
original_df = df.copy()
df.text = df.text.apply(clean_text)
print (f"{original_df.text.values[0]}\n{df.text.values[0]}")Output
Comparison between YOLO and RCNN on real world videos Bringing theory to experiment is cool. We can easily train models in colab and find the results in minutes. comparison yolo rcnn real world videos bringing theory experiment cool easily train models colab find results minutes
Clean DataFrame
In [ ]python · cell 43
python
# DataFrame cleanup
df = df.drop(columns=["id", "created_on", "title", "description"], errors="ignore") # drop cols
df = df.dropna(subset=["tag"]) # drop nulls
df = df[["text", "tag"]] # rearrange cols
df.head()Output
text tag 0 comparison yolo rcnn real world videos bringin... computer-vision 1 show infer tell contextual inference creative ... computer-vision 2 awesome graph classification collection import... other 3 awesome monte carlo tree search curated list m... other 4 attentionwalk pytorch implementation watch ste... other
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| text | tag | |
|---|---|---|
| 0 | comparison yolo rcnn real world videos bringin... | computer-vision |
| 1 | show infer tell contextual inference creative ... | computer-vision |
| 2 | awesome graph classification collection import... | other |
| 3 | awesome monte carlo tree search curated list m... | other |
| 4 | attentionwalk pytorch implementation watch ste... | other |
Label encoding
We need to encode our data into numerical values so that our models can process them. We'll start by encoding our text labels into unique indices.
In [ ]python · cell 46
python
# Label to index
tags = train_df.tag.unique().tolist()
num_classes = len(tags)
class_to_index = {tag: i for i, tag in enumerate(tags)}
class_to_indexOutput
{'mlops': 0,
'natural-language-processing': 1,
'computer-vision': 2,
'other': 3}In [ ]python · cell 47
python
# Encode labels
df["tag"] = df["tag"].map(class_to_index)
df.head()Output
text tag 0 comparison yolo rcnn real world videos bringin... 2 1 show infer tell contextual inference creative ... 2 2 awesome graph classification collection import... 3 3 awesome monte carlo tree search curated list m... 3 4 attentionwalk pytorch implementation watch ste... 3
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| text | tag | |
|---|---|---|
| 0 | comparison yolo rcnn real world videos bringin... | 2 |
| 1 | show infer tell contextual inference creative ... | 2 |
| 2 | awesome graph classification collection import... | 3 |
| 3 | awesome monte carlo tree search curated list m... | 3 |
| 4 | attentionwalk pytorch implementation watch ste... | 3 |
In [ ]python · cell 48
python
def decode(indices, index_to_class):
return [index_to_class[index] for index in indices]In [ ]python · cell 49
python
index_to_class = {v:k for k, v in class_to_index.items()}
decode(df.head()["tag"].values, index_to_class=index_to_class)Output
['computer-vision', 'computer-vision', 'other', 'other', 'other']
Tokenizer
In [ ]python · cell 52
python
import numpy as np
from transformers import BertTokenizerIn [ ]python · cell 53
python
# Bert tokenizer
tokenizer = BertTokenizer.from_pretrained("allenai/scibert_scivocab_uncased", return_dict=False)
text = "Transfer learning with transformers for text classification."
encoded_inputs = tokenizer([text], return_tensors="np", padding="longest") # pad to longest item in batch
print ("input_ids:", encoded_inputs["input_ids"])
print ("attention_mask:", encoded_inputs["attention_mask"])
print (tokenizer.decode(encoded_inputs["input_ids"][0]))Output
input_ids: [[ 102 2268 1904 190 29155 168 3267 2998 205 103]] attention_mask: [[1 1 1 1 1 1 1 1 1 1]] [CLS] transfer learning with transformers for text classification. [SEP]
In [ ]python · cell 54
python
def tokenize(batch):
tokenizer = BertTokenizer.from_pretrained("allenai/scibert_scivocab_uncased", return_dict=False)
encoded_inputs = tokenizer(batch["text"].tolist(), return_tensors="np", padding="longest")
return dict(ids=encoded_inputs["input_ids"], masks=encoded_inputs["attention_mask"], targets=np.array(batch["tag"]))In [ ]python · cell 55
python
# Tokenization
tokenize(df.head(1))Output
{'ids': array([[ 102, 2029, 1778, 609, 6446, 4857, 1332, 2399, 13572,
19125, 1983, 1954, 6240, 3717, 7434, 1262, 537, 201,
1040, 545, 4714, 103]]),
'masks': array([[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1]]),
'targets': array([2])}We'll combine all of our preprocessing steps into function:
In [ ]python · cell 58
python
def preprocess(df, class_to_index):
"""Preprocess the data."""
df["text"] = df.title + " " + df.description # feature engineering
df["text"] = df.text.apply(clean_text) # clean text
df = df.drop(columns=["id", "created_on", "title", "description"], errors="ignore") # clean dataframe
df = df[["text", "tag"]] # rearrange columns
df["tag"] = df["tag"].map(class_to_index) # label encoding
outputs = tokenize(df)
return outputsIn [ ]python · cell 59
python
# Apply
preprocess(df=train_df, class_to_index=class_to_index)Output
{'ids': array([[ 102, 856, 532, ..., 0, 0, 0],
[ 102, 2177, 29155, ..., 0, 0, 0],
[ 102, 2180, 3241, ..., 0, 0, 0],
...,
[ 102, 453, 2068, ..., 5730, 432, 103],
[ 102, 11268, 1782, ..., 0, 0, 0],
[ 102, 1596, 122, ..., 0, 0, 0]]),
'masks': array([[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
...,
[1, 1, 1, ..., 1, 1, 1],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0]]),
'targets': array([0, 1, 1, 1, 1, 2, 1, 2, 3, 1, 2, 2, 1, 1, 2, 2, 2, 2, 1, 2, 0, 1,
1, 1, 1, 1, 2, 1, 2, 0, 3, 2, 0, 1, 1, 1, 1, 2, 1, 1, 0, 2, 3, 3,
3, 0, 2, 1, 3, 3, 1, 1, 1, 1, 2, 1, 2, 2, 2, 3, 2, 1, 1, 3, 1, 0,
1, 2, 2, 2, 3, 2, 3, 2, 3, 2, 1, 1, 3, 3, 3, 1, 1, 2, 3, 0, 1, 1,
1, 1, 3, 3, 0, 2, 3, 2, 2, 1, 1, 3, 2, 3, 1, 1, 1, 1, 2, 0, 0, 2,
1, 1, 2, 2, 1, 1, 0, 3, 1, 2, 2, 1, 0, 2, 3, 1, 3, 1, 2, 3, 1, 1,
3, 3, 2, 1, 1, 0, 1, 3, 1, 1, 2, 2, 0, 0, 2, 1, 1, 1, 2, 3, 2, 1,
1, 2, 0, 1, 1, 3, 2, 1, 1, 2, 1, 2, 3, 1, 2, 2, 1, 2, 1, 2, 1, 3,
2, 2, 0, 1, 2, 2, 1, 2, 2, 1, 3, 2, 2, 1, 2, 2, 3, 2, 1, 1, 1, 1,
2, 2, 2, 0, 2, 1, 0, 2, 1, 3, 1, 1, 1, 1, 2, 1, 3, 3, 2, 1, 0, 1,
2, 0, 2, 2, 3, 1, 1, 2, 1, 1, 1, 1, 1, 2, 1, 1, 2, 1, 2, 2, 1, 2,
0, 2, 2, 1, 1, 2, 2, 2, 2, 2, 1, 1, 2, 3, 2, 1, 0, 2, 1, 2, 2, 1,
1, 2, 1, 2, 2, 1, 1, 2, 1, 2, 2, 2, 3, 2, 1, 2, 0, 2, 2, 1, 2, 3,
2, 1, 1, 1, 1, 1, 2, 1, 1, 2, 1, 2, 3, 2, 1, 2, 2, 2, 1, 2, 2, 2,
2, 1, 1, 2, 2, 1, 2, 2, 2, 2, 1, 1, 2, 1, 2, 2, 1, 3, 3, 0, 1, 3,
0, 2, 1, 1, 1, 1, 1, 0, 2, 1, 3, 2, 1, 2, 2, 1, 1, 3, 0, 3, 3, 2,
1, 1, 3, 3, 2, 3, 1, 1, 3, 1, 0, 1, 1, 1, 3, 0, 2, 2, 2, 1, 1, 2,
2, 1, 3, 2, 0, 1, 1, 1, 2, 1, 1, 1, 2, 1, 1, 2, 0, 3, 0, 1, 2, 1,
3, 2, 3, 2, 2, 0, 2, 3, 2, 2, 2, 1, 2, 1, 1, 1, 2, 1, 1, 1, 1, 1,
2, 2, 1, 2, 1, 1, 2, 2, 3, 1, 2, 2, 3, 2, 1, 1, 2, 0, 2, 0, 1, 1,
2, 1, 1, 3, 1, 1, 1, 1, 1, 3, 1, 2, 1, 0, 3, 1, 3, 2, 2, 1, 1, 3,
2, 1, 2, 1, 3, 1, 2, 2, 1, 2, 2, 2, 1, 0, 3, 2, 1, 3, 1, 1, 2, 1,
2, 2, 0, 1, 2, 1, 2, 2, 3, 1, 1, 2, 2, 1, 2, 2, 0, 0, 1, 2, 1, 1,
2, 1, 1, 2, 1, 1, 3, 2, 3, 1, 2, 2, 3, 0, 1, 1, 2, 1, 2, 1, 1, 1,
1, 1, 2, 1, 3, 1, 0, 2, 1, 3, 1, 2, 2, 1, 0, 2, 3, 2, 3, 2, 1, 1,
1, 2, 1, 2, 1, 2, 1, 3, 2, 2, 2, 2, 2, 1, 2, 0, 1, 0, 1, 2, 2, 1,
2, 3, 2, 1, 2, 2, 2, 3, 1, 3, 2, 1, 2, 2, 2, 1, 3, 1, 1, 2, 2, 1,
2, 3, 2, 2, 0, 1, 2, 2, 2, 0, 1, 2, 1, 3, 0, 2, 3])}Distributed preprocessing
The main issue with our approach above is that we're limited by our single machine in terms how much data our dataframe can hold and that we can preprocess. With the increasing trend in ML for larger unstructured datasets and larger models (LLMs), we can quickly outgrow our single machine constraints and will need to go distributed.
In [ ]python · cell 62
python
from madewithml.data import stratify_split
ray.data.DatasetContext.get_current().execution_options.preserve_order = TrueIn [ ]python · cell 63
python
# Data ingestion
ds = ray.data.read_csv(DATASET_LOC)
ds = ds.random_shuffle(seed=1234)
ds.take(1)Output
2023-12-07 11:26:57,951 INFO read_api.py:406 -- To satisfy the requested parallelism of 24, each read task output is split into 24 smaller blocks. 2023-12-07 11:26:57,959 INFO dataset.py:2380 -- Tip: Use `take_batch()` instead of `take() / show()` to return records in pandas or numpy batch format. 2023-12-07 11:26:57,960 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> LimitOperator[limit=1] 2023-12-07 11:26:57,961 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:26:57,961 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[{'id': 695,
'created_on': datetime.datetime(2020, 5, 2, 21, 33, 31),
'title': 'Five Cool Python Libraries for Data Science',
'description': 'Python is a best friend for the majority of the Data Scientists. Libraries make their life simpler. I have come across five cool Python libraries while working ',
'tag': 'natural-language-processing'}]In [ ]python · cell 64
python
# Split dataset
test_size = 0.2
train_ds, val_ds = stratify_split(ds, stratify="tag", test_size=test_size)Output
2023-12-07 11:26:59,973 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> LimitOperator[limit=1] 2023-12-07 11:26:59,974 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:26:59,974 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
In [ ]python · cell 65
python
# Mapping
tags = train_ds.unique(column="tag")
class_to_index = {tag: i for i, tag in enumerate(tags)}Output
2023-12-07 11:27:00,813 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> LimitOperator[limit=1] 2023-12-07 11:27:00,813 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:27:00,814 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/576 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/576 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/24 [00:00<?, ?it/s]
2023-12-07 11:27:01,560 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> AllToAllOperator[Aggregate] -> TaskPoolMapOperator[MapBatches(<lambda>)] 2023-12-07 11:27:01,561 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:27:01,562 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/576 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/576 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/576 [00:00<?, ?it/s]
- Aggregate 11: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 12: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 13: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/576 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/24 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/24 [00:00<?, ?it/s]
In [ ]python · cell 66
python
# Distributed preprocessing
sample_ds = train_ds.map_batches(preprocess, fn_kwargs={"class_to_index": class_to_index}, batch_format="pandas")
sample_ds.show(1)Output
2023-12-07 11:27:02,546 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] -> LimitOperator[limit=1] 2023-12-07 11:27:02,546 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:27:02,546 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/576 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/576 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/24 [00:00<?, ?it/s]
{'ids': array([ 102, 15820, 30126, 796, 15820, 30126, 796, 11669, 446,
3396, 17192, 9096, 15522, 1966, 15820, 30126, 487, 13387,
544, 3808, 4912, 6283, 2886, 19188, 1967, 881, 103,
0, 0]), 'masks': array([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1,
1, 1, 1, 1, 1, 0, 0]), 'targets': 2}
[2m[36m(MapBatches(preprocess) pid=21398)[0m Exception ignored in: <function Dataset.__del__ at 0x11b824e50> [2m[36m(MapBatches(preprocess) pid=21398)[0m Traceback (most recent call last): [2m[36m(MapBatches(preprocess) pid=21398)[0m File "/Users/goku/Documents/tobias/venv/lib/python3.10/site-packages/ray/data/dataset.py", line 5222, in __del__ [2m[36m(MapBatches(preprocess) pid=21398)[0m if self._current_executor and ray is not None and ray.is_initialized(): [2m[36m(MapBatches(preprocess) pid=21398)[0m KeyboardInterrupt:
Training
When developing models, it's always a best practice to start with the simplest models and slowly motivate more complex models. For example our baseline model progression would be:
- random model (predict labels randomly)
- rule-based model (pattern match labels in input text)
- logistic regression (td-idf vectors from text)
- CNN (apply character filters over text)
- Fine-tune LLM (this notebook)
We cover all of these methods in our other lessons but since our focus here in on MLOps, we will skip directly to fine-tuning an LLM for our task.
We'll first set up some functions that will help us achieve complete reproducibility.
In [ ]python · cell 70
python
import os
import random
import torch
from ray.data.preprocessor import PreprocessorIn [ ]python · cell 71
python
def set_seeds(seed=42):
"""Set seeds for reproducibility."""
np.random.seed(seed)
random.seed(seed)
torch.manual_seed(seed)
torch.cuda.manual_seed(seed)
eval("setattr(torch.backends.cudnn, 'deterministic', True)")
eval("setattr(torch.backends.cudnn, 'benchmark', False)")
os.environ["PYTHONHASHSEED"] = str(seed)In [ ]python · cell 72
python
def load_data(num_samples=None):
ds = ray.data.read_csv(DATASET_LOC)
ds = ds.random_shuffle(seed=1234)
ds = ray.data.from_items(ds.take(num_samples)) if num_samples else ds
return dsIn [ ]python · cell 73
python
class CustomPreprocessor():
"""Custom preprocessor class."""
def __init__(self, class_to_index={}):
self.class_to_index = class_to_index or {} # mutable defaults
self.index_to_class = {v: k for k, v in self.class_to_index.items()}
def fit(self, ds):
tags = ds.unique(column="tag")
self.class_to_index = {tag: i for i, tag in enumerate(tags)}
self.index_to_class = {v:k for k, v in self.class_to_index.items()}
return self
def transform(self, ds):
return ds.map_batches(
preprocess,
fn_kwargs={"class_to_index": self.class_to_index},
batch_format="pandas")🤖 Model
In [ ]python · cell 75
python
import torch.nn as nn
from transformers import BertModelIn [ ]python · cell 76
python
# Pretrained LLM
llm = BertModel.from_pretrained("allenai/scibert_scivocab_uncased", return_dict=False)
embedding_dim = llm.config.hidden_sizeOutput
Some weights of the model checkpoint at allenai/scibert_scivocab_uncased were not used when initializing BertModel: ['cls.predictions.decoder.weight', 'cls.predictions.transform.dense.bias', 'cls.predictions.bias', 'cls.seq_relationship.weight', 'cls.seq_relationship.bias', 'cls.predictions.transform.LayerNorm.weight', 'cls.predictions.transform.LayerNorm.bias', 'cls.predictions.transform.dense.weight', 'cls.predictions.decoder.bias'] - This IS expected if you are initializing BertModel from the checkpoint of a model trained on another task or with another architecture (e.g. initializing a BertForSequenceClassification model from a BertForPreTraining model). - This IS NOT expected if you are initializing BertModel from the checkpoint of a model that you expect to be exactly identical (initializing a BertForSequenceClassification model from a BertForSequenceClassification model).
In [ ]python · cell 77
python
# Sample
text = "Transfer learning with transformers for text classification."
batch = tokenizer([text], return_tensors="np", padding="longest")
batch = {k:torch.tensor(v) for k,v in batch.items()} # convert to torch tensors
seq, pool = llm(input_ids=batch["input_ids"], attention_mask=batch["attention_mask"])
np.shape(seq), np.shape(pool)Output
(torch.Size([1, 10, 768]), torch.Size([1, 768]))
In [ ]python · cell 78
python
class FinetunedLLM(nn.Module):
def __init__(self, llm, dropout_p, embedding_dim, num_classes):
super(FinetunedLLM, self).__init__()
self.llm = llm
self.dropout_p = dropout_p
self.embedding_dim = embedding_dim
self.num_classes = num_classes
self.dropout = torch.nn.Dropout(dropout_p)
self.fc1 = torch.nn.Linear(embedding_dim, num_classes)
def forward(self, batch):
ids, masks = batch["ids"], batch["masks"]
seq, pool = self.llm(input_ids=ids, attention_mask=masks)
z = self.dropout(pool)
z = self.fc1(z)
return z
@torch.inference_mode()
def predict(self, batch):
self.eval()
z = self(batch)
y_pred = torch.argmax(z, dim=1).cpu().numpy()
return y_pred
@torch.inference_mode()
def predict_proba(self, batch):
self.eval()
z = self(batch)
y_probs = F.softmax(z, dim=1).cpu().numpy()
return y_probs
def save(self, dp):
with open(Path(dp, "args.json"), "w") as fp:
contents = {
"dropout_p": self.dropout_p,
"embedding_dim": self.embedding_dim,
"num_classes": self.num_classes,
}
json.dump(contents, fp, indent=4, sort_keys=False)
torch.save(self.state_dict(), os.path.join(dp, "model.pt"))
@classmethod
def load(cls, args_fp, state_dict_fp):
with open(args_fp, "r") as fp:
kwargs = json.load(fp=fp)
llm = BertModel.from_pretrained("allenai/scibert_scivocab_uncased", return_dict=False)
model = cls(llm=llm, **kwargs)
model.load_state_dict(torch.load(state_dict_fp, map_location=torch.device("cpu")))
return modelIn [ ]python · cell 79
python
# Initialize model
model = FinetunedLLM(llm=llm, dropout_p=0.5, embedding_dim=embedding_dim, num_classes=num_classes)
print (model.named_parameters)Output
<bound method Module.named_parameters of FinetunedLLM(
(llm): BertModel(
(embeddings): BertEmbeddings(
(word_embeddings): Embedding(31090, 768, padding_idx=0)
(position_embeddings): Embedding(512, 768)
(token_type_embeddings): Embedding(2, 768)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
(encoder): BertEncoder(
(layer): ModuleList(
(0-11): 12 x BertLayer(
(attention): BertAttention(
(self): BertSelfAttention(
(query): Linear(in_features=768, out_features=768, bias=True)
(key): Linear(in_features=768, out_features=768, bias=True)
(value): Linear(in_features=768, out_features=768, bias=True)
(dropout): Dropout(p=0.1, inplace=False)
)
(output): BertSelfOutput(
(dense): Linear(in_features=768, out_features=768, bias=True)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
)
(intermediate): BertIntermediate(
(dense): Linear(in_features=768, out_features=3072, bias=True)
(intermediate_act_fn): GELUActivation()
)
(output): BertOutput(
(dense): Linear(in_features=3072, out_features=768, bias=True)
(LayerNorm): LayerNorm((768,), eps=1e-12, elementwise_affine=True)
(dropout): Dropout(p=0.1, inplace=False)
)
)
)
)
(pooler): BertPooler(
(dense): Linear(in_features=768, out_features=768, bias=True)
(activation): Tanh()
)
)
(dropout): Dropout(p=0.5, inplace=False)
(fc1): Linear(in_features=768, out_features=4, bias=True)
)>
📦 Batching
We can iterate through our dataset in batches however we may have batches of different sizes. Recall that our tokenizer padded the inputs to the longest item in the batch (padding="longest"). However, our batches for training will be smaller than our large data processing batches and so our batches here may have inputs with different lengths. To address this, we're going to define a custom collate_fn to repad the items in our training batches.
In [ ]python · cell 82
python
from ray.train.torch import get_deviceOutput
Created a temporary directory at /var/folders/x8/ffj4btjx0cv6r653rgfnffsc0000gn/T/tmpfieao6lq Writing /var/folders/x8/ffj4btjx0cv6r653rgfnffsc0000gn/T/tmpfieao6lq/_remote_module_non_scriptable.py
In [ ]python · cell 83
python
def pad_array(arr, dtype=np.int32):
max_len = max(len(row) for row in arr)
padded_arr = np.zeros((arr.shape[0], max_len), dtype=dtype)
for i, row in enumerate(arr):
padded_arr[i][:len(row)] = row
return padded_arrIn [ ]python · cell 84
python
def collate_fn(batch):
batch["ids"] = pad_array(batch["ids"])
batch["masks"] = pad_array(batch["masks"])
dtypes = {"ids": torch.int32, "masks": torch.int32, "targets": torch.int64}
tensor_batch = {}
for key, array in batch.items():
tensor_batch[key] = torch.as_tensor(array, dtype=dtypes[key], device=get_device())
return tensor_batch
pad=(0, max_len)in F.pad refers to (left_padding, right_padding) on the input. There will be no left-padding (hence the0) and only right-padding. And theconstantmode refers to each element being padded to a constant size (size of longest element in the input).
In [ ]python · cell 86
python
# Sample batch
sample_batch = sample_ds.take_batch(batch_size=128)
collate_fn(batch=sample_batch)Output
2023-12-07 11:27:34,483 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(24)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] -> LimitOperator[limit=128] 2023-12-07 11:27:34,484 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-12-07 11:27:34,484 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/576 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/576 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/576 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/576 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/576 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/24 [00:00<?, ?it/s]
{'ids': tensor([[ 102, 638, 30108, ..., 0, 0, 0],
[ 102, 11448, 20591, ..., 0, 0, 0],
[ 102, 13564, 1995, ..., 0, 0, 0],
...,
[ 102, 18403, 30114, ..., 0, 0, 0],
[ 102, 2567, 1995, ..., 0, 0, 0],
[ 102, 239, 30118, ..., 24660, 30131, 103]], dtype=torch.int32),
'masks': tensor([[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
...,
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 0, 0, 0],
[1, 1, 1, ..., 1, 1, 1]], dtype=torch.int32),
'targets': tensor([0, 0, 3, 0, 0, 0, 2, 1, 2, 2, 0, 3, 3, 0, 0, 2, 2, 2, 1, 0, 2, 3, 2, 1,
2, 0, 2, 2, 0, 0, 3, 0, 0, 0, 2, 1, 2, 2, 0, 3, 3, 0, 0, 2, 2, 2, 1, 0,
2, 3, 2, 1, 2, 0, 2, 2, 0, 2, 0, 0, 0, 2, 0, 0, 0, 0, 2, 3, 2, 3, 2, 2,
2, 0, 2, 3, 2, 3, 0, 2, 2, 0, 2, 0, 0, 0, 2, 0, 0, 0, 0, 2, 2, 1, 2, 2,
2, 1, 0, 2, 2, 1, 0, 2, 2, 0, 2, 0, 2, 2, 0, 0, 0, 0, 2, 2, 2, 2, 0, 0,
0, 2, 2, 0, 2, 2, 0, 0])}🧮 Utilities
In [ ]python · cell 88
python
from pathlib import Path
import ray.train as train
from ray.train import Checkpoint, CheckpointConfig, DataConfig, RunConfig, ScalingConfig
from ray.train.torch import TorchCheckpoint, TorchTrainer
import tempfile
import torch.nn.functional as F
from torch.nn.parallel.distributed import DistributedDataParallelIn [ ]python · cell 89
python
def train_step(ds, batch_size, model, num_classes, loss_fn, optimizer):
"""Train step."""
model.train()
loss = 0.0
ds_generator = ds.iter_torch_batches(batch_size=batch_size, collate_fn=collate_fn)
for i, batch in enumerate(ds_generator):
optimizer.zero_grad() # reset gradients
z = model(batch) # forward pass
targets = F.one_hot(batch["targets"], num_classes=num_classes).float() # one-hot (for loss_fn)
J = loss_fn(z, targets) # define loss
J.backward() # backward pass
optimizer.step() # update weights
loss += (J.detach().item() - loss) / (i + 1) # cumulative loss
return lossIn [ ]python · cell 90
python
def eval_step(ds, batch_size, model, num_classes, loss_fn):
"""Eval step."""
model.eval()
loss = 0.0
y_trues, y_preds = [], []
ds_generator = ds.iter_torch_batches(batch_size=batch_size, collate_fn=collate_fn)
with torch.inference_mode():
for i, batch in enumerate(ds_generator):
z = model(batch)
targets = F.one_hot(batch["targets"], num_classes=num_classes).float() # one-hot (for loss_fn)
J = loss_fn(z, targets).item()
loss += (J - loss) / (i + 1)
y_trues.extend(batch["targets"].cpu().numpy())
y_preds.extend(torch.argmax(z, dim=1).cpu().numpy())
return loss, np.vstack(y_trues), np.vstack(y_preds)In [ ]python · cell 91
python
# Training loop
def train_loop_per_worker(config):
# Hyperparameters
dropout_p = config["dropout_p"]
lr = config["lr"]
lr_factor = config["lr_factor"]
lr_patience = config["lr_patience"]
num_epochs = config["num_epochs"]
batch_size = config["batch_size"]
num_classes = config["num_classes"]
# Get datasets
set_seeds()
train_ds = train.get_dataset_shard("train")
val_ds = train.get_dataset_shard("val")
# Model
llm = BertModel.from_pretrained("allenai/scibert_scivocab_uncased", return_dict=False)
model = FinetunedLLM(llm=llm, dropout_p=dropout_p, embedding_dim=llm.config.hidden_size, num_classes=num_classes)
model = train.torch.prepare_model(model)
# Training components
loss_fn = nn.BCEWithLogitsLoss()
optimizer = torch.optim.Adam(model.parameters(), lr=lr)
scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode="min", factor=lr_factor, patience=lr_patience)
# Training
num_workers = train.get_context().get_world_size()
batch_size_per_worker = batch_size // num_workers
for epoch in range(num_epochs):
# Step
train_loss = train_step(train_ds, batch_size_per_worker, model, num_classes, loss_fn, optimizer)
val_loss, _, _ = eval_step(val_ds, batch_size_per_worker, model, num_classes, loss_fn)
scheduler.step(val_loss)
# Checkpoint
with tempfile.TemporaryDirectory() as dp:
if isinstance(model, DistributedDataParallel): # cpu
model.module.save(dp=dp)
else:
model.save(dp=dp)
metrics = dict(epoch=epoch, lr=optimizer.param_groups[0]["lr"], train_loss=train_loss, val_loss=val_loss)
checkpoint = Checkpoint.from_directory(dp)
train.report(metrics, checkpoint=checkpoint)Our dataset doesn't suffer from horrible class imbalance, but if it did, we could easily account for it through our loss function. There are also other strategies such as over-sampling less frequent classes and under-sampling popular classes.
python
# Class weights
batch_counts = []
for batch in train_ds.iter_torch_batches(batch_size=256, collate_fn=collate_fn):
batch_counts.append(np.bincount(batch["targets"].cpu().numpy()))
counts = [sum(count) for count in zip(*batch_counts)]
class_weights = np.array([1.0/count for i, count in enumerate(counts)])
class_weights_tensor = torch.Tensor(class_weights).to(get_device())
# Training components
loss_fn = nn.BCEWithLogitsLoss(weight=class_weights_tensor)
...🗂️ Configurations
In [ ]python · cell 94
python
from madewithml.config import EFS_DIRIn [ ]python · cell 95
python
# Train loop config
train_loop_config = {
"dropout_p": 0.5,
"lr": 1e-4,
"lr_factor": 0.8,
"lr_patience": 3,
"num_epochs": 10,
"batch_size": 256,
"num_classes": num_classes,
}In [ ]python · cell 96
python
# Scaling config
scaling_config = ScalingConfig(
num_workers=num_workers,
use_gpu=bool(resources_per_worker["GPU"]),
resources_per_worker=resources_per_worker
)In [ ]python · cell 97
python
# Run config
checkpoint_config = CheckpointConfig(num_to_keep=1, checkpoint_score_attribute="val_loss", checkpoint_score_order="min")
run_config = RunConfig(name="llm", checkpoint_config=checkpoint_config, storage_path=EFS_DIR)🚂 Training
In [ ]python · cell 99
python
# Dataset
ds = load_data()
train_ds, val_ds = stratify_split(ds, stratify="tag", test_size=test_size)Output
2023-09-18 21:57:50,561 INFO read_api.py:406 -- To satisfy the requested parallelism of 64, each read task output is split into 64 smaller blocks. 2023-09-18 21:57:50,565 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> LimitOperator[limit=1] 2023-09-18 21:57:50,565 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-18 21:57:50,566 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
In [ ]python · cell 100
python
# Preprocess
preprocessor = CustomPreprocessor()
preprocessor = preprocessor.fit(train_ds)
train_ds = preprocessor.transform(train_ds)
val_ds = preprocessor.transform(val_ds)
train_ds = train_ds.materialize()
val_ds = val_ds.materialize()Output
2023-09-18 21:57:51,011 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> LimitOperator[limit=1] 2023-09-18 21:57:51,012 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-18 21:57:51,012 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-18 21:57:53,336 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> AllToAllOperator[Aggregate] -> TaskPoolMapOperator[MapBatches(<lambda>)] 2023-09-18 21:57:53,337 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-18 21:57:53,338 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
- Aggregate 11: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 12: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 13: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-18 21:57:54,975 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-18 21:57:54,977 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-18 21:57:54,977 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-18 21:57:56,897 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-18 21:57:56,897 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-18 21:57:56,898 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
In [ ]python · cell 101
python
# Dataset config
options = ray.data.ExecutionOptions(preserve_order=True)
dataset_config = DataConfig(
datasets_to_split=["train"],
execution_options=options)In [ ]python · cell 102
python
# Trainer
trainer = TorchTrainer(
train_loop_per_worker=train_loop_per_worker,
train_loop_config=train_loop_config,
scaling_config=scaling_config,
run_config=run_config,
datasets={"train": train_ds, "val": val_ds},
dataset_config=dataset_config,
metadata={"class_to_index": preprocessor.class_to_index}
)In [ ]python · cell 103
python
%%time
# Train
results = trainer.fit()Output
<IPython.core.display.HTML object>
Tune Status
| Current time: | 2023-09-18 21:59:12 |
| Running for: | 00:01:13.39 |
| Memory: | 22.2/62.1 GiB |
System Info
Using FIFO scheduling algorithm.Logical resource usage: 4.0/32 CPUs, 1.0/2 GPUs (0.0/2.0 accelerator_type:A10G)
Trial Status
| Trial name | status | loc | iter | total time (s) | epoch | lr | train_loss |
|---|---|---|---|---|---|---|---|
| TorchTrainer_1a81f_00000 | TERMINATED | 10.0.34.101:727773 | 10 | 68.3685 | 9 | 0.0001 | 0.0525064 |
[2m[36m(TorchTrainer pid=727773, ip=10.0.34.101)[0m Starting distributed worker processes: ['727834 (10.0.34.101)'] [2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Setting up process group for: env:// [rank=0, world_size=1] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Auto configuring locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'] [2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Moving model to device: cuda:0 [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.) [2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000000) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000001) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000002) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000003) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000004) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000005) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000006) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000007) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000008) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=727900, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=727900, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=727834, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_1a81f_00000_0_2023-09-18_21-57-58/checkpoint_000009) 2023-09-18 21:59:12,318 INFO tune.py:1143 -- Total run time: 73.40 seconds (73.36 seconds for the tuning loop).
CPU times: user 1.31 s, sys: 364 ms, total: 1.68 s Wall time: 1min 13s
In [ ]python · cell 104
python
# Metrics per epoch
results.metrics_dataframeOutput
epoch lr train_loss val_loss timestamp should_checkpoint done \ 0 0 0.0001 0.577060 0.499463 1695099495 True False 1 1 0.0001 0.497796 0.454631 1695099501 True False 2 2 0.0001 0.432476 0.324219 1695099507 True False 3 3 0.0001 0.319699 0.250069 1695099513 True False 4 4 0.0001 0.243872 0.208918 1695099519 True False 5 5 0.0001 0.188871 0.183051 1695099525 True False 6 6 0.0001 0.145348 0.139673 1695099532 True False 7 7 0.0001 0.102919 0.126615 1695099538 True False 8 8 0.0001 0.074093 0.135746 1695099544 True False 9 9 0.0001 0.052506 0.113568 1695099550 True False training_iteration trial_id date ... \ 0 1 1a81f_00000 2023-09-18_21-58-15 ... 1 2 1a81f_00000 2023-09-18_21-58-21 ... 2 3 1a81f_00000 2023-09-18_21-58-27 ... 3 4 1a81f_00000 2023-09-18_21-58-33 ... 4 5 1a81f_00000 2023-09-18_21-58-39 ... 5 6 1a81f_00000 2023-09-18_21-58-45 ... 6 7 1a81f_00000 2023-09-18_21-58-52 ... 7 8 1a81f_00000 2023-09-18_21-58-58 ... 8 9 1a81f_00000 2023-09-18_21-59-04 ... 9 10 1a81f_00000 2023-09-18_21-59-10 ... time_since_restore iterations_since_restore checkpoint_dir_name \ 0 12.883882 1 checkpoint_000000 1 18.986689 2 checkpoint_000001 2 25.088869 3 checkpoint_000002 3 31.192535 4 checkpoint_000003 4 37.338241 5 checkpoint_000004 5 43.458540 6 checkpoint_000005 6 49.559415 7 checkpoint_000006 7 56.116772 8 checkpoint_000007 8 62.219563 9 checkpoint_000008 9 68.368526 10 checkpoint_000009 config/train_loop_config/dropout_p config/train_loop_config/lr \ 0 0.5 0.0001 1 0.5 0.0001 2 0.5 0.0001 3 0.5 0.0001 4 0.5 0.0001 5 0.5 0.0001 6 0.5 0.0001 7 0.5 0.0001 8 0.5 0.0001 9 0.5 0.0001 config/train_loop_config/lr_factor config/train_loop_config/lr_patience \ 0 0.8 3 1 0.8 3 2 0.8 3 3 0.8 3 4 0.8 3 5 0.8 3 6 0.8 3 7 0.8 3 8 0.8 3 9 0.8 3 config/train_loop_config/num_epochs config/train_loop_config/batch_size \ 0 10 256 1 10 256 2 10 256 3 10 256 4 10 256 5 10 256 6 10 256 7 10 256 8 10 256 9 10 256 config/train_loop_config/num_classes 0 4 1 4 2 4 3 4 4 4 5 4 6 4 7 4 8 4 9 4 [10 rows x 25 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| epoch | lr | train_loss | val_loss | timestamp | should_checkpoint | done | training_iteration | trial_id | date | ... | time_since_restore | iterations_since_restore | checkpoint_dir_name | config/train_loop_config/dropout_p | config/train_loop_config/lr | config/train_loop_config/lr_factor | config/train_loop_config/lr_patience | config/train_loop_config/num_epochs | config/train_loop_config/batch_size | config/train_loop_config/num_classes | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 0 | 0.0001 | 0.577060 | 0.499463 | 1695099495 | True | False | 1 | 1a81f_00000 | 2023-09-18_21-58-15 | ... | 12.883882 | 1 | checkpoint_000000 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 1 | 1 | 0.0001 | 0.497796 | 0.454631 | 1695099501 | True | False | 2 | 1a81f_00000 | 2023-09-18_21-58-21 | ... | 18.986689 | 2 | checkpoint_000001 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 2 | 2 | 0.0001 | 0.432476 | 0.324219 | 1695099507 | True | False | 3 | 1a81f_00000 | 2023-09-18_21-58-27 | ... | 25.088869 | 3 | checkpoint_000002 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 3 | 3 | 0.0001 | 0.319699 | 0.250069 | 1695099513 | True | False | 4 | 1a81f_00000 | 2023-09-18_21-58-33 | ... | 31.192535 | 4 | checkpoint_000003 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 4 | 4 | 0.0001 | 0.243872 | 0.208918 | 1695099519 | True | False | 5 | 1a81f_00000 | 2023-09-18_21-58-39 | ... | 37.338241 | 5 | checkpoint_000004 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 5 | 5 | 0.0001 | 0.188871 | 0.183051 | 1695099525 | True | False | 6 | 1a81f_00000 | 2023-09-18_21-58-45 | ... | 43.458540 | 6 | checkpoint_000005 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 6 | 6 | 0.0001 | 0.145348 | 0.139673 | 1695099532 | True | False | 7 | 1a81f_00000 | 2023-09-18_21-58-52 | ... | 49.559415 | 7 | checkpoint_000006 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 7 | 7 | 0.0001 | 0.102919 | 0.126615 | 1695099538 | True | False | 8 | 1a81f_00000 | 2023-09-18_21-58-58 | ... | 56.116772 | 8 | checkpoint_000007 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 8 | 8 | 0.0001 | 0.074093 | 0.135746 | 1695099544 | True | False | 9 | 1a81f_00000 | 2023-09-18_21-59-04 | ... | 62.219563 | 9 | checkpoint_000008 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 9 | 9 | 0.0001 | 0.052506 | 0.113568 | 1695099550 | True | False | 10 | 1a81f_00000 | 2023-09-18_21-59-10 | ... | 68.368526 | 10 | checkpoint_000009 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
10 rows × 25 columns
In [ ]python · cell 105
python
# Best checkpoints
results.best_checkpointsOutput
[(Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/llm/TorchTrainer_e2941_00000_0_2023-09-17_22-40-33/checkpoint_000009),
{'epoch': 9,
'lr': 0.0001,
'train_loss': 0.052506402134895325,
'val_loss': 0.11356845498085022,
'timestamp': 1695015703,
'should_checkpoint': True,
'done': False,
'training_iteration': 10,
'trial_id': 'e2941_00000',
'date': '2023-09-17_22-41-43',
'time_this_iter_s': 6.128554105758667,
'time_total_s': 66.49748063087463,
'pid': 842775,
'hostname': 'ip-10-0-35-174',
'node_ip': '10.0.35.174',
'config': {'train_loop_config': {'dropout_p': 0.5,
'lr': 0.0001,
'lr_factor': 0.8,
'lr_patience': 3,
'num_epochs': 10,
'batch_size': 256,
'num_classes': 4}},
'time_since_restore': 66.49748063087463,
'iterations_since_restore': 10})]Evaluation
In [ ]python · cell 107
python
from sklearn.metrics import precision_recall_fscore_supportIn [ ]python · cell 108
python
class TorchPredictor:
def __init__(self, preprocessor, model):
self.preprocessor = preprocessor
self.model = model
self.model.eval()
def __call__(self, batch):
results = self.model.predict(collate_fn(batch))
return {"output": results}
def predict_proba(self, batch):
results = self.model.predict_proba(collate_fn(batch))
return {"output": results}
def get_preprocessor(self):
return self.preprocessor
@classmethod
def from_checkpoint(cls, checkpoint):
metadata = checkpoint.get_metadata()
preprocessor = CustomPreprocessor(class_to_index=metadata["class_to_index"])
model = FinetunedLLM.load(Path(checkpoint.path, "args.json"), Path(checkpoint.path, "model.pt"))
return cls(preprocessor=preprocessor, model=model)In [ ]python · cell 109
python
# Artifacts
best_checkpoint = results.best_checkpoints[0][0]
predictor = TorchPredictor.from_checkpoint(best_checkpoint)
preprocessor = predictor.get_preprocessor()In [ ]python · cell 110
python
# Test (holdout) dataset
HOLDOUT_LOC = "https://raw.githubusercontent.com/GokuMohandas/Made-With-ML/main/datasets/holdout.csv"
test_ds = ray.data.read_csv(HOLDOUT_LOC)
preprocessed_ds = preprocessor.transform(test_ds)
preprocessed_ds.take(1)Output
2023-09-17 22:41:46,264 INFO read_api.py:406 -- To satisfy the requested parallelism of 64, each read task output is split into 64 smaller blocks. 2023-09-17 22:41:46,268 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)] -> LimitOperator[limit=1] 2023-09-17 22:41:46,268 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:41:46,269 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[{'ids': array([ 102, 4905, 2069, 2470, 2848, 4905, 30132, 22081, 691,
4324, 7491, 5896, 341, 6136, 934, 30137, 103, 0]),
'masks': array([1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 0]),
'targets': 3}]In [ ]python · cell 111
python
# y_true
values = preprocessed_ds.select_columns(cols=["targets"]).take_all()
y_true = np.stack([item["targets"] for item in values])
print (y_true)Output
2023-09-17 22:41:46,831 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(<lambda>)] 2023-09-17 22:41:46,831 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:41:46,832 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
[3 3 3 0 2 0 0 0 0 2 0 0 2 3 0 0 2 2 3 2 3 0 3 2 0 2 2 1 1 2 2 2 2 2 2 0 0 0 0 0 1 1 2 0 0 3 1 2 0 2 2 3 3 0 2 3 2 3 3 3 3 0 0 0 0 2 2 0 2 1 0 2 3 0 0 2 2 2 2 2 0 0 2 0 1 0 0 0 0 3 0 0 2 0 2 2 3 2 0 2 0 2 0 3 0 0 0 0 0 2 0 0 2 2 2 2 3 0 2 0 2 0 2 3 3 3 2 0 2 2 2 2 0 2 2 2 0 1 2 2 2 2 2 1 2 0 3 0 2 2 1 1 2 0 0 0 0 0 0 2 2 2 0 2 1 1 2 0 0 1 2 3 2 2 2 0 0 2 0 2 0 3 0 2 2 0 1 2 1 2 2]
In [ ]python · cell 112
python
# y_pred
predictions = preprocessed_ds.map_batches(predictor).take_all()
y_pred = np.array([d["output"] for d in predictions])Output
2023-09-17 22:41:47,783 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor)] 2023-09-17 22:41:47,785 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:41:47,786 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
[2m[36m(MapBatches(preprocess)->MapBatches(TorchPredictor) pid=348532, ip=10.0.34.101)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
In [ ]python · cell 113
python
# Evaluate
metrics = precision_recall_fscore_support(y_true, y_pred, average="weighted")
{"precision": metrics[0], "recall": metrics[1], "f1": metrics[2]}Output
{'precision': 0.9138952286238713,
'recall': 0.9109947643979057,
'f1': 0.9114851103432928}In [ ]python · cell 114
python
def evaluate(ds, predictor):
# y_true
preprocessor = predictor.get_preprocessor()
preprocessed_ds = preprocessor.transform(ds)
values = preprocessed_ds.select_columns(cols=["targets"]).take_all()
y_true = np.stack([item["targets"] for item in values])
# y_pred
predictions = preprocessed_ds.map_batches(predictor).take_all()
y_pred = np.array([d["output"] for d in predictions])
# Evaluate
metrics = precision_recall_fscore_support(y_true, y_pred, average="weighted")
performance = {"precision": metrics[0], "recall": metrics[1], "f1": metrics[2]}
return performanceIn [ ]python · cell 115
python
# Evaluate on test split
performance = evaluate(ds=test_ds, predictor=predictor)
print (json.dumps(performance, indent=2))Output
2023-09-17 22:41:54,734 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(<lambda>)] 2023-09-17 22:41:54,734 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:41:54,735 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
2023-09-17 22:41:55,455 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor)] 2023-09-17 22:41:55,456 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:41:55,456 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
{
"precision": 0.9138952286238713,
"recall": 0.9109947643979057,
"f1": 0.9114851103432928
}
Inference
In [ ]python · cell 117
python
import pandas as pdIn [ ]python · cell 118
python
def format_prob(prob, index_to_class):
d = {}
for i, item in enumerate(prob):
d[index_to_class[i]] = item
return dIn [ ]python · cell 119
python
def predict_proba(ds, predictor):
preprocessor = predictor.get_preprocessor()
preprocessed_ds = preprocessor.transform(ds)
outputs = preprocessed_ds.map_batches(predictor.predict_proba)
y_prob = np.array([d["output"] for d in outputs.take_all()])
results = []
for i, prob in enumerate(y_prob):
tag = preprocessor.index_to_class[prob.argmax()]
results.append({"prediction": tag, "probabilities": format_prob(prob, preprocessor.index_to_class)})
return resultsIn [ ]python · cell 120
python
# Predict on sample
title = "Transfer learning with transformers"
description = "Using transformers for transfer learning on text classification tasks."
sample_ds = ray.data.from_items([{"title": title, "description": description, "tag": "other"}])
predict_proba(ds=sample_ds, predictor=predictor)Output
2023-09-17 22:42:00,133 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:42:00,134 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:00,134 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[{'prediction': 'natural-language-processing',
'probabilities': {'computer-vision': 0.00035399722,
'mlops': 0.00030543839,
'natural-language-processing': 0.9990171,
'other': 0.0003234732}}]🧪 Experiment tracking
So far, we've been training our models but we don't have a way to more deeply track and compare them. We'll achieve this but defining a proper process for experiment tracking which we'll use for all future experiments (including hyperparameter optimization). Experiment tracking is the processing of managing all the different experiments and their components, such as parameters, metrics, models and other artifacts and it enables us to:
- Organize all the necessary components of a specific experiment. It's important to have everything in one place and know where it is so you can use them later.
- Reproduce past results (easily) using saved experiments.
- Log iterative improvements across time, data, ideas, teams, etc.
There are many options for experiment tracking but we're going to use MLflow (100% free and open-source) because it has all the functionality we'll need (and growing integration support). There are also several popular options such as a Comet ML (Used by Google AI, HuggingFace, etc.) and Weights and Biases (Used by Open AI, Toyota Research, etc.). These are fantastic options if you want a fully managed experiment tracking solution.
In [ ]python · cell 123
python
import mlflow
from pathlib import Path
from ray.tune.logger.mlflow import MLflowLoggerCallback
import timeIn [ ]python · cell 124
python
# Config MLflow
MODEL_REGISTRY = Path(f"{EFS_DIR}/mlflow")
Path(MODEL_REGISTRY).mkdir(parents=True, exist_ok=True)
MLFLOW_TRACKING_URI = "file://" + str(MODEL_REGISTRY.absolute())
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
print (mlflow.get_tracking_uri())Output
file:///efs/shared_storage/madewithml/GokuMohandas/mlflow
In [ ]python · cell 125
python
# MLflow callback
experiment_name = f"llm-{int(time.time())}"
mlflow_callback = MLflowLoggerCallback(
tracking_uri=MLFLOW_TRACKING_URI,
experiment_name=experiment_name,
save_artifact=True)In [ ]python · cell 126
python
# Run configuration with MLflow callback
run_config = RunConfig(
callbacks=[mlflow_callback],
checkpoint_config=checkpoint_config,
storage_path=EFS_DIR,
local_dir=EFS_DIR
)In [ ]python · cell 127
python
# Dataset
ds = load_data()
train_ds, val_ds = stratify_split(ds, stratify="tag", test_size=test_size)Output
2023-09-17 22:42:01,332 INFO read_api.py:406 -- To satisfy the requested parallelism of 64, each read task output is split into 64 smaller blocks. 2023-09-17 22:42:01,337 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> LimitOperator[limit=1] 2023-09-17 22:42:01,337 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:01,338 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
In [ ]python · cell 128
python
# Preprocess
preprocessor = CustomPreprocessor()
preprocessor = preprocessor.fit(train_ds)
train_ds = preprocessor.transform(train_ds)
val_ds = preprocessor.transform(val_ds)
train_ds = train_ds.materialize()
val_ds = val_ds.materialize()Output
2023-09-17 22:42:01,795 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> LimitOperator[limit=1] 2023-09-17 22:42:01,795 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:01,796 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:42:03,339 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> AllToAllOperator[Aggregate] -> TaskPoolMapOperator[MapBatches(<lambda>)] 2023-09-17 22:42:03,340 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:03,340 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
- Aggregate 11: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 12: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 13: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:42:05,078 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-17 22:42:05,079 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:05,079 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:42:07,292 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-17 22:42:07,293 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:42:07,293 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
In [ ]python · cell 129
python
# Trainer
trainer = TorchTrainer(
train_loop_per_worker=train_loop_per_worker,
train_loop_config=train_loop_config,
scaling_config=scaling_config,
run_config=run_config, # uses RunConfig with MLflow callback
datasets={"train": train_ds, "val": val_ds},
dataset_config=dataset_config,
metadata={"class_to_index": preprocessor.class_to_index}
)In [ ]python · cell 130
python
%%time
# Train
results = trainer.fit()Output
<IPython.core.display.HTML object>
Tune Status
| Current time: | 2023-09-17 22:43:31 |
| Running for: | 00:01:21.68 |
| Memory: | 20.8/62.1 GiB |
System Info
Using FIFO scheduling algorithm.Logical resource usage: 4.0/32 CPUs, 1.0/2 GPUs (0.0/2.0 accelerator_type:A10G)
Trial Status
| Trial name | status | loc | iter | total time (s) | epoch | lr | train_loss |
|---|---|---|---|---|---|---|---|
| TorchTrainer_1bd07_00000 | TERMINATED | 10.0.35.174:844750 | 10 | 63.6994 | 9 | 0.0001 | 0.0421994 |
[2m[36m(TorchTrainer pid=844750)[0m Starting distributed worker processes: ['844833 (10.0.35.174)'] [2m[36m(RayTrainWorker pid=844833)[0m Setting up process group for: env:// [rank=0, world_size=1] [2m[36m(SplitCoordinator pid=844909)[0m Auto configuring locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'] [2m[36m(RayTrainWorker pid=844833)[0m Moving model to device: cuda:0
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.) [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000000)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000001)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000002)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000003)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000004)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000005)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000006)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)]
[2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False)
[2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
2023-09-17 22:43:01,995 WARNING worker.py:2012 -- [tqdm_ray] Failed to decode {"__magic_token__": "__ray_tqdm_magic_token__", "x": 47, "pos": 0, "desc": "Running: 0.0/32.0 CPU, 0.0/2.0 GPU, 0.28 MiB/8.95 GiB object_store_memory", "total": 64, "ip": "10.0.35.174", "pid": 844909, "uuid": "54df93474628434c, this may be due to logging too fast. This warning will not be printed again.
[2m[36m(SplitCoordinator pid=844909)[0m 8d2fe8f67788261f", "closed": false}
[2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000007)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000008)
(pid=844909) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=844909)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=844909)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=844909)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=844833)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-42-09/TorchTrainer_1bd07_00000_0_2023-09-17_22-42-09/checkpoint_000009) 2023-09-17 22:43:31,004 INFO tune.py:1143 -- Total run time: 81.76 seconds (81.66 seconds for the tuning loop).
CPU times: user 1.23 s, sys: 1.31 s, total: 2.53 s Wall time: 1min 21s
In [ ]python · cell 131
python
results.metrics_dataframeOutput
epoch lr train_loss val_loss timestamp should_checkpoint done \ 0 0 0.0001 0.577780 0.493102 1695015745 True False 1 1 0.0001 0.489653 0.431958 1695015751 True False 2 2 0.0001 0.398824 0.306201 1695015757 True False 3 3 0.0001 0.300513 0.238803 1695015763 True False 4 4 0.0001 0.220608 0.174411 1695015769 True False 5 5 0.0001 0.151499 0.158648 1695015775 True False 6 6 0.0001 0.105087 0.112829 1695015781 True False 7 7 0.0001 0.077385 0.091922 1695015788 True False 8 8 0.0001 0.053857 0.109810 1695015794 True False 9 9 0.0001 0.042199 0.121122 1695015800 True False training_iteration trial_id date ... \ 0 1 1bd07_00000 2023-09-17_22-42-25 ... 1 2 1bd07_00000 2023-09-17_22-42-31 ... 2 3 1bd07_00000 2023-09-17_22-42-37 ... 3 4 1bd07_00000 2023-09-17_22-42-43 ... 4 5 1bd07_00000 2023-09-17_22-42-49 ... 5 6 1bd07_00000 2023-09-17_22-42-55 ... 6 7 1bd07_00000 2023-09-17_22-43-01 ... 7 8 1bd07_00000 2023-09-17_22-43-08 ... 8 9 1bd07_00000 2023-09-17_22-43-14 ... 9 10 1bd07_00000 2023-09-17_22-43-20 ... time_since_restore iterations_since_restore checkpoint_dir_name \ 0 11.882575 1 checkpoint_000000 1 17.654435 2 checkpoint_000001 2 23.353752 3 checkpoint_000002 3 29.118992 4 checkpoint_000003 4 34.897485 5 checkpoint_000004 5 40.645384 6 checkpoint_000005 6 46.387568 7 checkpoint_000006 7 52.153343 8 checkpoint_000007 8 57.901084 9 checkpoint_000008 9 63.699444 10 checkpoint_000009 config/train_loop_config/dropout_p config/train_loop_config/lr \ 0 0.5 0.0001 1 0.5 0.0001 2 0.5 0.0001 3 0.5 0.0001 4 0.5 0.0001 5 0.5 0.0001 6 0.5 0.0001 7 0.5 0.0001 8 0.5 0.0001 9 0.5 0.0001 config/train_loop_config/lr_factor config/train_loop_config/lr_patience \ 0 0.8 3 1 0.8 3 2 0.8 3 3 0.8 3 4 0.8 3 5 0.8 3 6 0.8 3 7 0.8 3 8 0.8 3 9 0.8 3 config/train_loop_config/num_epochs config/train_loop_config/batch_size \ 0 10 256 1 10 256 2 10 256 3 10 256 4 10 256 5 10 256 6 10 256 7 10 256 8 10 256 9 10 256 config/train_loop_config/num_classes 0 4 1 4 2 4 3 4 4 4 5 4 6 4 7 4 8 4 9 4 [10 rows x 25 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| epoch | lr | train_loss | val_loss | timestamp | should_checkpoint | done | training_iteration | trial_id | date | ... | time_since_restore | iterations_since_restore | checkpoint_dir_name | config/train_loop_config/dropout_p | config/train_loop_config/lr | config/train_loop_config/lr_factor | config/train_loop_config/lr_patience | config/train_loop_config/num_epochs | config/train_loop_config/batch_size | config/train_loop_config/num_classes | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 0 | 0.0001 | 0.577780 | 0.493102 | 1695015745 | True | False | 1 | 1bd07_00000 | 2023-09-17_22-42-25 | ... | 11.882575 | 1 | checkpoint_000000 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 1 | 1 | 0.0001 | 0.489653 | 0.431958 | 1695015751 | True | False | 2 | 1bd07_00000 | 2023-09-17_22-42-31 | ... | 17.654435 | 2 | checkpoint_000001 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 2 | 2 | 0.0001 | 0.398824 | 0.306201 | 1695015757 | True | False | 3 | 1bd07_00000 | 2023-09-17_22-42-37 | ... | 23.353752 | 3 | checkpoint_000002 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 3 | 3 | 0.0001 | 0.300513 | 0.238803 | 1695015763 | True | False | 4 | 1bd07_00000 | 2023-09-17_22-42-43 | ... | 29.118992 | 4 | checkpoint_000003 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 4 | 4 | 0.0001 | 0.220608 | 0.174411 | 1695015769 | True | False | 5 | 1bd07_00000 | 2023-09-17_22-42-49 | ... | 34.897485 | 5 | checkpoint_000004 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 5 | 5 | 0.0001 | 0.151499 | 0.158648 | 1695015775 | True | False | 6 | 1bd07_00000 | 2023-09-17_22-42-55 | ... | 40.645384 | 6 | checkpoint_000005 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 6 | 6 | 0.0001 | 0.105087 | 0.112829 | 1695015781 | True | False | 7 | 1bd07_00000 | 2023-09-17_22-43-01 | ... | 46.387568 | 7 | checkpoint_000006 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 7 | 7 | 0.0001 | 0.077385 | 0.091922 | 1695015788 | True | False | 8 | 1bd07_00000 | 2023-09-17_22-43-08 | ... | 52.153343 | 8 | checkpoint_000007 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 8 | 8 | 0.0001 | 0.053857 | 0.109810 | 1695015794 | True | False | 9 | 1bd07_00000 | 2023-09-17_22-43-14 | ... | 57.901084 | 9 | checkpoint_000008 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
| 9 | 9 | 0.0001 | 0.042199 | 0.121122 | 1695015800 | True | False | 10 | 1bd07_00000 | 2023-09-17_22-43-20 | ... | 63.699444 | 10 | checkpoint_000009 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 256 | 4 |
10 rows × 25 columns
In [ ]python · cell 132
python
# Sorted runs
sorted_runs = mlflow.search_runs(experiment_names=[experiment_name], order_by=["metrics.val_loss ASC"])
sorted_runsOutput
run_id experiment_id status \
0 a67f6ec4e10b49a7845b91db61efc7e0 703905858072772764 FINISHED
artifact_uri \
0 file:///efs/shared_storage/madewithml/GokuMoha...
start_time end_time \
0 2023-09-18 05:42:12.863000+00:00 2023-09-18 05:43:30.925000+00:00
metrics.config/train_loop_config/lr_patience metrics.time_total_s \
0 3.0 63.699444
metrics.time_since_restore metrics.config/train_loop_config/num_epochs \
0 63.699444 10.0
... metrics.epoch params.train_loop_config/num_epochs \
0 ... 9.0 10
params.train_loop_config/lr_patience params.train_loop_config/batch_size \
0 3 256
params.train_loop_config/lr params.train_loop_config/lr_factor \
0 0.0001 0.8
params.train_loop_config/num_classes params.train_loop_config/dropout_p \
0 4 0.5
tags.trial_name tags.mlflow.runName
0 TorchTrainer_1bd07_00000 TorchTrainer_1bd07_00000
[1 rows x 35 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| run_id | experiment_id | status | artifact_uri | start_time | end_time | metrics.config/train_loop_config/lr_patience | metrics.time_total_s | metrics.time_since_restore | metrics.config/train_loop_config/num_epochs | ... | metrics.epoch | params.train_loop_config/num_epochs | params.train_loop_config/lr_patience | params.train_loop_config/batch_size | params.train_loop_config/lr | params.train_loop_config/lr_factor | params.train_loop_config/num_classes | params.train_loop_config/dropout_p | tags.trial_name | tags.mlflow.runName | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | a67f6ec4e10b49a7845b91db61efc7e0 | 703905858072772764 | FINISHED | file:///efs/shared_storage/madewithml/GokuMoha... | 2023-09-18 05:42:12.863000+00:00 | 2023-09-18 05:43:30.925000+00:00 | 3.0 | 63.699444 | 63.699444 | 10.0 | ... | 9.0 | 10 | 3 | 256 | 0.0001 | 0.8 | 4 | 0.5 | TorchTrainer_1bd07_00000 | TorchTrainer_1bd07_00000 |
1 rows × 35 columns
In [ ]python · cell 133
python
# Best run
best_run = sorted_runs.iloc[0]
best_runOutput
run_id a67f6ec4e10b49a7845b91db61efc7e0 experiment_id 703905858072772764 status FINISHED artifact_uri file:///efs/shared_storage/madewithml/GokuMoha... start_time 2023-09-18 05:42:12.863000+00:00 end_time 2023-09-18 05:43:30.925000+00:00 metrics.config/train_loop_config/lr_patience 3.0 metrics.time_total_s 63.699444 metrics.time_since_restore 63.699444 metrics.config/train_loop_config/num_epochs 10.0 metrics.iterations_since_restore 10.0 metrics.config/train_loop_config/batch_size 256.0 metrics.pid 844750.0 metrics.should_checkpoint 1.0 metrics.timestamp 1695015800.0 metrics.config/train_loop_config/num_classes 4.0 metrics.done 0.0 metrics.time_this_iter_s 5.79836 metrics.train_loss 0.042199 metrics.config/train_loop_config/lr 0.0001 metrics.lr 0.0001 metrics.val_loss 0.121122 metrics.config/train_loop_config/dropout_p 0.5 metrics.training_iteration 10.0 metrics.config/train_loop_config/lr_factor 0.8 metrics.epoch 9.0 params.train_loop_config/num_epochs 10 params.train_loop_config/lr_patience 3 params.train_loop_config/batch_size 256 params.train_loop_config/lr 0.0001 params.train_loop_config/lr_factor 0.8 params.train_loop_config/num_classes 4 params.train_loop_config/dropout_p 0.5 tags.trial_name TorchTrainer_1bd07_00000 tags.mlflow.runName TorchTrainer_1bd07_00000 Name: 0, dtype: object
Dashboard
Let's view what we've tracked from our experiment. MLFlow serves a dashboard for us to view and explore our experiments on a localhost port:
bash
mlflow server -h 0.0.0.0 -p 8080 --backend-store-uri $EFS_DIR/mlflowMLFlow creates a main dashboard with all your experiments and their respective runs. We can sort runs by clicking on the column headers.
And within each run, we can view metrics, parameters, artifacts, etc.
And we can even create custom plots to help us visualize our results.
Loading
In [ ]python · cell 138
python
from ray.train import Result
from urllib.parse import urlparseIn [ ]python · cell 139
python
def get_best_checkpoint(run_id):
artifact_dir = urlparse(mlflow.get_run(run_id).info.artifact_uri).path # get path from mlflow
results = Result.from_path(artifact_dir)
return results.best_checkpoints[0][0]In [ ]python · cell 140
python
# Artifacts
best_checkpoint = get_best_checkpoint(run_id=best_run.run_id)
predictor = TorchPredictor.from_checkpoint(best_checkpoint)
preprocessor = predictor.get_preprocessor()In [ ]python · cell 141
python
# Evaluate on test split
performance = evaluate(ds=test_ds, predictor=predictor)
print (json.dumps(performance, indent=2))Output
2023-09-17 22:43:36,691 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(<lambda>)] 2023-09-17 22:43:36,692 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:36,692 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
2023-09-17 22:43:37,551 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor)] 2023-09-17 22:43:37,552 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:37,553 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
[2m[36m(MapBatches(preprocess)->MapBatches(TorchPredictor) pid=845827)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
{
"precision": 0.9168092951711627,
"recall": 0.9109947643979057,
"f1": 0.9105512639658029
}
In [ ]python · cell 142
python
# Predict on sample
title = "Transfer learning with transformers"
description = "Using transformers for transfer learning on text classification tasks."
sample_ds = ray.data.from_items([{"title": title, "description": description, "tag": "other"}])
predict_proba(ds=sample_ds, predictor=predictor)Output
2023-09-17 22:43:43,282 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:43:43,282 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:43,283 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[{'prediction': 'natural-language-processing',
'probabilities': {'computer-vision': 0.00043165276,
'mlops': 0.0008155016,
'natural-language-processing': 0.9978746,
'other': 0.0008782037}}]⚙ Hyperparameter tuning
In [ ]python · cell 144
python
from ray import tune
from ray.tune import Tuner
from ray.tune.schedulers import AsyncHyperBandScheduler
from ray.tune.search import ConcurrencyLimiter
from ray.tune.search.hyperopt import HyperOptSearchIn [ ]python · cell 145
python
# Number of trials (small sample)
num_runs = 2In [ ]python · cell 146
python
# Set up
set_seeds()In [ ]python · cell 147
python
# Dataset
ds = load_data()
train_ds, val_ds = stratify_split(ds, stratify="tag", test_size=test_size)Output
2023-09-17 22:43:44,422 INFO read_api.py:406 -- To satisfy the requested parallelism of 64, each read task output is split into 64 smaller blocks. 2023-09-17 22:43:44,426 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> LimitOperator[limit=1] 2023-09-17 22:43:44,427 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:44,427 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
In [ ]python · cell 148
python
# Preprocess
preprocessor = CustomPreprocessor()
preprocessor = preprocessor.fit(train_ds)
train_ds = preprocessor.transform(train_ds)
val_ds = preprocessor.transform(val_ds)
train_ds = train_ds.materialize()
val_ds = val_ds.materialize()Output
2023-09-17 22:43:44,885 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> LimitOperator[limit=1] 2023-09-17 22:43:44,886 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:44,887 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:43:46,122 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> AllToAllOperator[Aggregate] -> TaskPoolMapOperator[MapBatches(<lambda>)] 2023-09-17 22:43:46,123 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:46,123 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
- Aggregate 11: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 12: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 13: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:43:47,913 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-17 22:43:47,914 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:47,914 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
2023-09-17 22:43:50,092 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> AllToAllOperator[RandomShuffle] -> AllToAllOperator[Sort] -> AllToAllOperator[MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle] -> TaskPoolMapOperator[MapBatches(preprocess)] 2023-09-17 22:43:50,092 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:43:50,093 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
- RandomShuffle 1: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 2: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 3: 0%| | 0/4096 [00:00<?, ?it/s]
- Sort 4: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 5: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 6: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 7: 0%| | 0/4096 [00:00<?, ?it/s]
- MapBatches(group_fn)->MapBatches(_filter_split)->RandomShuffle 8: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Map 9: 0%| | 0/4096 [00:00<?, ?it/s]
Shuffle Reduce 10: 0%| | 0/4096 [00:00<?, ?it/s]
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
Sort Sample 0: 0%| | 0/64 [00:00<?, ?it/s]
In [ ]python · cell 149
python
# Trainer
trainer = TorchTrainer(
train_loop_per_worker=train_loop_per_worker,
train_loop_config=train_loop_config,
scaling_config=scaling_config,
datasets={"train": train_ds, "val": val_ds},
dataset_config=dataset_config,
metadata={"class_to_index": preprocessor.class_to_index}
)In [ ]python · cell 150
python
# MLflow callback
mlflow_callback = MLflowLoggerCallback(
tracking_uri=MLFLOW_TRACKING_URI,
experiment_name=experiment_name,
save_artifact=True)In [ ]python · cell 151
python
# Run configuration
checkpoint_config = CheckpointConfig(num_to_keep=1, checkpoint_score_attribute="val_loss", checkpoint_score_order="min")
run_config = RunConfig(
callbacks=[mlflow_callback],
checkpoint_config=checkpoint_config,
storage_path=EFS_DIR,
local_dir=EFS_DIR)In [ ]python · cell 152
python
# Hyperparameters to start with
initial_params = [{"train_loop_config": {"dropout_p": 0.5, "lr": 1e-4, "lr_factor": 0.8, "lr_patience": 3}}]
search_alg = HyperOptSearch(points_to_evaluate=initial_params)
search_alg = ConcurrencyLimiter(search_alg, max_concurrent=2) # trade off b/w optimization and search spaceIn [ ]python · cell 153
python
# Parameter space
param_space = {
"train_loop_config": {
"dropout_p": tune.uniform(0.3, 0.9),
"lr": tune.loguniform(1e-5, 5e-4),
"lr_factor": tune.uniform(0.1, 0.9),
"lr_patience": tune.uniform(1, 10),
}
}In [ ]python · cell 154
python
# Scheduler
scheduler = AsyncHyperBandScheduler(
max_t=train_loop_config["num_epochs"], # max epoch (<time_attr>) per trial
grace_period=5, # min epoch (<time_attr>) per trial
)In [ ]python · cell 155
python
# Tune config
tune_config = tune.TuneConfig(
metric="val_loss",
mode="min",
search_alg=search_alg,
scheduler=scheduler,
num_samples=num_runs,
)In [ ]python · cell 156
python
# Tuner
tuner = Tuner(
trainable=trainer,
run_config=run_config,
param_space=param_space,
tune_config=tune_config,
)In [ ]python · cell 157
python
%%time
# Tune
results = tuner.fit()Output
<IPython.core.display.HTML object>
Tune Status
| Current time: | 2023-09-17 22:45:41 |
| Running for: | 00:01:49.07 |
| Memory: | 21.8/62.1 GiB |
System Info
Using AsyncHyperBand: num_stopped=2Bracket: Iter 5.000: -0.2372603341937065
Logical resource usage: 4.0/32 CPUs, 1.0/2 GPUs (0.0/2.0 accelerator_type:A10G)
Trial Status
| Trial name | status | loc | train_loop_config/dr opout_p | train_loop_config/lr | train_loop_config/lr _factor | train_loop_config/lr _patience | iter | total time (s) | epoch | lr | train_loss |
|---|---|---|---|---|---|---|---|---|---|---|---|
| TorchTrainer_639d7776 | TERMINATED | 10.0.35.174:846705 | 0.5 | 0.0001 | 0.8 | 3 | 10 | 68.3885 | 9 | 0.0001 | 0.0520358 |
| TorchTrainer_145c1bc2 | TERMINATED | 10.0.34.101:349716 | 0.841192 | 5.18042e-05 | 0.758627 | 2.25374 | 5 | 30.7297 | 4 | 5.18042e-05 | 0.392521 |
[2m[36m(TorchTrainer pid=846705)[0m Starting distributed worker processes: ['846788 (10.0.35.174)'] [2m[36m(RayTrainWorker pid=846788)[0m Setting up process group for: env:// [rank=0, world_size=1] [2m[36m(SplitCoordinator pid=846864)[0m Auto configuring locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'] [2m[36m(RayTrainWorker pid=846788)[0m Moving model to device: cuda:0
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=846788)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.) [2m[36m(TorchTrainer pid=349716, ip=10.0.34.101)[0m Starting distributed worker processes: ['349780 (10.0.34.101)'] [2m[36m(RayTrainWorker pid=349780, ip=10.0.34.101)[0m Setting up process group for: env:// [rank=0, world_size=1] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Auto configuring locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b']
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000000) [2m[36m(RayTrainWorker pid=349780, ip=10.0.34.101)[0m Moving model to device: cuda:0 [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=349780, ip=10.0.34.101)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000001)[32m [repeated 2x across cluster][0m [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:44:16,982 WARNING util.py:315 -- The `callbacks.on_trial_result` operation took 3.185 s, which may be a performance bottleneck. 2023-09-17 22:44:16,983 WARNING util.py:315 -- The `process_trial_result` operation took 3.187 s, which may be a performance bottleneck. 2023-09-17 22:44:16,984 WARNING util.py:315 -- Processing trial results took 3.188 s, which may be a performance bottleneck. Please consider reporting results less frequently to Ray Tune. 2023-09-17 22:44:16,984 WARNING util.py:315 -- The `process_trial_result` operation took 3.188 s, which may be a performance bottleneck.
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000002)[32m [repeated 2x across cluster][0m [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:44:25,141 WARNING util.py:315 -- The `callbacks.on_trial_result` operation took 3.025 s, which may be a performance bottleneck. 2023-09-17 22:44:25,143 WARNING util.py:315 -- The `process_trial_result` operation took 3.028 s, which may be a performance bottleneck. 2023-09-17 22:44:25,144 WARNING util.py:315 -- Processing trial results took 3.028 s, which may be a performance bottleneck. Please consider reporting results less frequently to Ray Tune. 2023-09-17 22:44:25,144 WARNING util.py:315 -- The `process_trial_result` operation took 3.029 s, which may be a performance bottleneck.
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000003)[32m [repeated 2x across cluster][0m [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:44:33,395 WARNING util.py:315 -- The `callbacks.on_trial_result` operation took 3.181 s, which may be a performance bottleneck. 2023-09-17 22:44:33,397 WARNING util.py:315 -- The `process_trial_result` operation took 3.183 s, which may be a performance bottleneck. 2023-09-17 22:44:33,398 WARNING util.py:315 -- Processing trial results took 3.184 s, which may be a performance bottleneck. Please consider reporting results less frequently to Ray Tune. 2023-09-17 22:44:33,398 WARNING util.py:315 -- The `process_trial_result` operation took 3.184 s, which may be a performance bottleneck.
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000004)[32m [repeated 2x across cluster][0m [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:44:41,574 WARNING util.py:315 -- The `callbacks.on_trial_result` operation took 3.101 s, which may be a performance bottleneck. 2023-09-17 22:44:41,576 WARNING util.py:315 -- The `process_trial_result` operation took 3.102 s, which may be a performance bottleneck. 2023-09-17 22:44:41,576 WARNING util.py:315 -- Processing trial results took 3.103 s, which may be a performance bottleneck. Please consider reporting results less frequently to Ray Tune. 2023-09-17 22:44:41,577 WARNING util.py:315 -- The `process_trial_result` operation took 3.104 s, which may be a performance bottleneck.
(pid=349841, ip=10.0.34.101) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000005)[32m [repeated 2x across cluster][0m [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=349841, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:44:49,825 WARNING util.py:315 -- The `callbacks.on_trial_result` operation took 3.286 s, which may be a performance bottleneck. 2023-09-17 22:44:49,828 WARNING util.py:315 -- The `process_trial_result` operation took 3.289 s, which may be a performance bottleneck. 2023-09-17 22:44:49,828 WARNING util.py:315 -- Processing trial results took 3.289 s, which may be a performance bottleneck. Please consider reporting results less frequently to Ray Tune. 2023-09-17 22:44:49,828 WARNING util.py:315 -- The `process_trial_result` operation took 3.289 s, which may be a performance bottleneck. [2m[36m(RayTrainWorker pid=349780, ip=10.0.34.101)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_145c1bc2_2_dropout_p=0.8412,lr=0.0001,lr_factor=0.7586,lr_patience=2.2537_2023-09-17_22-43-55/checkpoint_000005) [2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000006)
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000007)
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000008)
(pid=846864) Running 0: 0%| | 0/64 [00:00<?, ?it/s]
[2m[36m(SplitCoordinator pid=846864)[0m Executing DAG InputDataBuffer[Input] -> OutputSplitter[split(1, equal=True)] [2m[36m(SplitCoordinator pid=846864)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=['c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a'], preserve_order=True, actor_locality_enabled=True, verbose_progress=False) [2m[36m(SplitCoordinator pid=846864)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` [2m[36m(RayTrainWorker pid=846788)[0m Checkpoint successfully created at: Checkpoint(filesystem=local, path=/efs/shared_storage/madewithml/GokuMohandas/TorchTrainer_2023-09-17_22-43-52/TorchTrainer_639d7776_1_dropout_p=0.5000,lr=0.0001,lr_factor=0.8000,lr_patience=3.0000_2023-09-17_22-43-52/checkpoint_000009) 2023-09-17 22:45:41,369 INFO tune.py:1143 -- Total run time: 109.13 seconds (109.03 seconds for the tuning loop).
CPU times: user 1.67 s, sys: 1.85 s, total: 3.52 s Wall time: 1min 49s
In [ ]python · cell 158
python
# All trials in experiment
results.get_dataframe()Output
epoch lr train_loss val_loss timestamp should_checkpoint done \
0 9 0.000100 0.052036 0.096391 1695015936 True True
1 4 0.000052 0.392521 0.326320 1695015885 True True
training_iteration trial_id date ... hostname \
0 10 639d7776 2023-09-17_22-45-36 ... ip-10-0-35-174
1 5 145c1bc2 2023-09-17_22-44-46 ... ip-10-0-34-101
node_ip time_since_restore iterations_since_restore \
0 10.0.35.174 68.388524 10
1 10.0.34.101 30.729746 5
checkpoint_dir_name config/train_loop_config/dropout_p \
0 checkpoint_000009 0.500000
1 checkpoint_000004 0.841192
config/train_loop_config/lr config/train_loop_config/lr_factor \
0 0.000100 0.800000
1 0.000052 0.758627
config/train_loop_config/lr_patience logdir
0 3.000000 639d7776
1 2.253736 145c1bc2
[2 rows x 23 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| epoch | lr | train_loss | val_loss | timestamp | should_checkpoint | done | training_iteration | trial_id | date | ... | hostname | node_ip | time_since_restore | iterations_since_restore | checkpoint_dir_name | config/train_loop_config/dropout_p | config/train_loop_config/lr | config/train_loop_config/lr_factor | config/train_loop_config/lr_patience | logdir | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 9 | 0.000100 | 0.052036 | 0.096391 | 1695015936 | True | True | 10 | 639d7776 | 2023-09-17_22-45-36 | ... | ip-10-0-35-174 | 10.0.35.174 | 68.388524 | 10 | checkpoint_000009 | 0.500000 | 0.000100 | 0.800000 | 3.000000 | 639d7776 |
| 1 | 4 | 0.000052 | 0.392521 | 0.326320 | 1695015885 | True | True | 5 | 145c1bc2 | 2023-09-17_22-44-46 | ... | ip-10-0-34-101 | 10.0.34.101 | 30.729746 | 5 | checkpoint_000004 | 0.841192 | 0.000052 | 0.758627 | 2.253736 | 145c1bc2 |
2 rows × 23 columns
In [ ]python · cell 159
python
# Best trial's epochs
best_trial = results.get_best_result(metric="val_loss", mode="min")
best_trial.metrics_dataframeOutput
epoch lr train_loss val_loss timestamp should_checkpoint done \
0 0 0.0001 0.577860 0.492227 1695015848 True False
1 1 0.0001 0.475743 0.387187 1695015856 True False
2 2 0.0001 0.402374 0.372390 1695015864 True False
3 3 0.0001 0.349068 0.297704 1695015873 True False
4 4 0.0001 0.256888 0.207574 1695015881 True False
5 5 0.0001 0.204279 0.161283 1695015889 True False
6 6 0.0001 0.142716 0.138546 1695015897 True False
7 7 0.0001 0.097615 0.108034 1695015924 True False
8 8 0.0001 0.070097 0.102292 1695015930 True False
9 9 0.0001 0.052036 0.096391 1695015936 True True
training_iteration trial_id date ... pid \
0 1 639d7776 2023-09-17_22-44-08 ... 846705
1 2 639d7776 2023-09-17_22-44-16 ... 846705
2 3 639d7776 2023-09-17_22-44-24 ... 846705
3 4 639d7776 2023-09-17_22-44-33 ... 846705
4 5 639d7776 2023-09-17_22-44-41 ... 846705
5 6 639d7776 2023-09-17_22-44-49 ... 846705
6 7 639d7776 2023-09-17_22-45-18 ... 846705
7 8 639d7776 2023-09-17_22-45-24 ... 846705
8 9 639d7776 2023-09-17_22-45-30 ... 846705
9 10 639d7776 2023-09-17_22-45-36 ... 846705
hostname node_ip time_since_restore iterations_since_restore \
0 ip-10-0-35-174 10.0.35.174 12.828465 1
1 ip-10-0-35-174 10.0.35.174 20.651286 2
2 ip-10-0-35-174 10.0.35.174 28.235628 3
3 ip-10-0-35-174 10.0.35.174 35.873792 4
4 ip-10-0-35-174 10.0.35.174 43.487612 5
5 ip-10-0-35-174 10.0.35.174 51.113538 6
6 ip-10-0-35-174 10.0.35.174 51.115670 7
7 ip-10-0-35-174 10.0.35.174 56.781526 8
8 ip-10-0-35-174 10.0.35.174 62.607184 9
9 ip-10-0-35-174 10.0.35.174 68.388524 10
checkpoint_dir_name config/train_loop_config/dropout_p \
0 checkpoint_000000 0.5
1 checkpoint_000001 0.5
2 checkpoint_000002 0.5
3 checkpoint_000003 0.5
4 checkpoint_000004 0.5
5 checkpoint_000005 0.5
6 checkpoint_000006 0.5
7 checkpoint_000007 0.5
8 checkpoint_000008 0.5
9 checkpoint_000009 0.5
config/train_loop_config/lr config/train_loop_config/lr_factor \
0 0.0001 0.8
1 0.0001 0.8
2 0.0001 0.8
3 0.0001 0.8
4 0.0001 0.8
5 0.0001 0.8
6 0.0001 0.8
7 0.0001 0.8
8 0.0001 0.8
9 0.0001 0.8
config/train_loop_config/lr_patience
0 3.0
1 3.0
2 3.0
3 3.0
4 3.0
5 3.0
6 3.0
7 3.0
8 3.0
9 3.0
[10 rows x 22 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| epoch | lr | train_loss | val_loss | timestamp | should_checkpoint | done | training_iteration | trial_id | date | ... | pid | hostname | node_ip | time_since_restore | iterations_since_restore | checkpoint_dir_name | config/train_loop_config/dropout_p | config/train_loop_config/lr | config/train_loop_config/lr_factor | config/train_loop_config/lr_patience | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | 0 | 0.0001 | 0.577860 | 0.492227 | 1695015848 | True | False | 1 | 639d7776 | 2023-09-17_22-44-08 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 12.828465 | 1 | checkpoint_000000 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 1 | 1 | 0.0001 | 0.475743 | 0.387187 | 1695015856 | True | False | 2 | 639d7776 | 2023-09-17_22-44-16 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 20.651286 | 2 | checkpoint_000001 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 2 | 2 | 0.0001 | 0.402374 | 0.372390 | 1695015864 | True | False | 3 | 639d7776 | 2023-09-17_22-44-24 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 28.235628 | 3 | checkpoint_000002 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 3 | 3 | 0.0001 | 0.349068 | 0.297704 | 1695015873 | True | False | 4 | 639d7776 | 2023-09-17_22-44-33 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 35.873792 | 4 | checkpoint_000003 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 4 | 4 | 0.0001 | 0.256888 | 0.207574 | 1695015881 | True | False | 5 | 639d7776 | 2023-09-17_22-44-41 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 43.487612 | 5 | checkpoint_000004 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 5 | 5 | 0.0001 | 0.204279 | 0.161283 | 1695015889 | True | False | 6 | 639d7776 | 2023-09-17_22-44-49 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 51.113538 | 6 | checkpoint_000005 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 6 | 6 | 0.0001 | 0.142716 | 0.138546 | 1695015897 | True | False | 7 | 639d7776 | 2023-09-17_22-45-18 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 51.115670 | 7 | checkpoint_000006 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 7 | 7 | 0.0001 | 0.097615 | 0.108034 | 1695015924 | True | False | 8 | 639d7776 | 2023-09-17_22-45-24 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 56.781526 | 8 | checkpoint_000007 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 8 | 8 | 0.0001 | 0.070097 | 0.102292 | 1695015930 | True | False | 9 | 639d7776 | 2023-09-17_22-45-30 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 62.607184 | 9 | checkpoint_000008 | 0.5 | 0.0001 | 0.8 | 3.0 |
| 9 | 9 | 0.0001 | 0.052036 | 0.096391 | 1695015936 | True | True | 10 | 639d7776 | 2023-09-17_22-45-36 | ... | 846705 | ip-10-0-35-174 | 10.0.35.174 | 68.388524 | 10 | checkpoint_000009 | 0.5 | 0.0001 | 0.8 | 3.0 |
10 rows × 22 columns
In [ ]python · cell 160
python
# Best trial's hyperparameters
best_trial.config["train_loop_config"]Output
{'dropout_p': 0.5, 'lr': 0.0001, 'lr_factor': 0.8, 'lr_patience': 3.0}In [ ]python · cell 161
python
# Sorted runs
sorted_runs = mlflow.search_runs(experiment_names=[experiment_name], order_by=["metrics.val_loss ASC"])
sorted_runsOutput
run_id experiment_id status \
0 ebc5576cd02e4430bbe949612a25bd45 703905858072772764 FINISHED
1 a67f6ec4e10b49a7845b91db61efc7e0 703905858072772764 FINISHED
2 0d8a50554ecf498cb6f1471b58aedb9d 703905858072772764 FINISHED
artifact_uri \
0 file:///efs/shared_storage/madewithml/GokuMoha...
1 file:///efs/shared_storage/madewithml/GokuMoha...
2 file:///efs/shared_storage/madewithml/GokuMoha...
start_time end_time \
0 2023-09-18 05:43:55.499000+00:00 2023-09-18 05:45:41.257000+00:00
1 2023-09-18 05:42:12.863000+00:00 2023-09-18 05:43:30.925000+00:00
2 2023-09-18 05:43:59.010000+00:00 2023-09-18 05:45:17.549000+00:00
metrics.time_total_s metrics.config/train_loop_config/lr_patience \
0 68.388524 3.000000
1 63.699444 3.000000
2 30.729746 2.253736
metrics.time_since_restore metrics.iterations_since_restore ... \
0 68.388524 10.0 ...
1 63.699444 10.0 ...
2 30.729746 5.0 ...
metrics.config/train_loop_config/num_epochs params.train_loop_config/lr \
0 NaN 0.0001
1 10.0 0.0001
2 NaN 5.1804178109970566e-05
params.train_loop_config/lr_patience params.train_loop_config/dropout_p \
0 3.0 0.5
1 3 0.5
2 2.253735644669187 0.8411920116073033
params.train_loop_config/lr_factor params.train_loop_config/num_classes \
0 0.8 None
1 0.8 4
2 0.7586266131367906 None
params.train_loop_config/num_epochs params.train_loop_config/batch_size \
0 None None
1 10 256
2 None None
tags.trial_name tags.mlflow.runName
0 TorchTrainer_639d7776 TorchTrainer_639d7776
1 TorchTrainer_1bd07_00000 TorchTrainer_1bd07_00000
2 TorchTrainer_145c1bc2 TorchTrainer_145c1bc2
[3 rows x 35 columns]
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| run_id | experiment_id | status | artifact_uri | start_time | end_time | metrics.time_total_s | metrics.config/train_loop_config/lr_patience | metrics.time_since_restore | metrics.iterations_since_restore | ... | metrics.config/train_loop_config/num_epochs | params.train_loop_config/lr | params.train_loop_config/lr_patience | params.train_loop_config/dropout_p | params.train_loop_config/lr_factor | params.train_loop_config/num_classes | params.train_loop_config/num_epochs | params.train_loop_config/batch_size | tags.trial_name | tags.mlflow.runName | |
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| 0 | ebc5576cd02e4430bbe949612a25bd45 | 703905858072772764 | FINISHED | file:///efs/shared_storage/madewithml/GokuMoha... | 2023-09-18 05:43:55.499000+00:00 | 2023-09-18 05:45:41.257000+00:00 | 68.388524 | 3.000000 | 68.388524 | 10.0 | ... | NaN | 0.0001 | 3.0 | 0.5 | 0.8 | None | None | None | TorchTrainer_639d7776 | TorchTrainer_639d7776 |
| 1 | a67f6ec4e10b49a7845b91db61efc7e0 | 703905858072772764 | FINISHED | file:///efs/shared_storage/madewithml/GokuMoha... | 2023-09-18 05:42:12.863000+00:00 | 2023-09-18 05:43:30.925000+00:00 | 63.699444 | 3.000000 | 63.699444 | 10.0 | ... | 10.0 | 0.0001 | 3 | 0.5 | 0.8 | 4 | 10 | 256 | TorchTrainer_1bd07_00000 | TorchTrainer_1bd07_00000 |
| 2 | 0d8a50554ecf498cb6f1471b58aedb9d | 703905858072772764 | FINISHED | file:///efs/shared_storage/madewithml/GokuMoha... | 2023-09-18 05:43:59.010000+00:00 | 2023-09-18 05:45:17.549000+00:00 | 30.729746 | 2.253736 | 30.729746 | 5.0 | ... | NaN | 5.1804178109970566e-05 | 2.253735644669187 | 0.8411920116073033 | 0.7586266131367906 | None | None | None | TorchTrainer_145c1bc2 | TorchTrainer_145c1bc2 |
3 rows × 35 columns
In [ ]python · cell 162
python
# Artifacts
best_checkpoint = get_best_checkpoint(run_id=best_run.run_id)
predictor = TorchPredictor.from_checkpoint(best_checkpoint)
preprocessor = predictor.get_preprocessor()In [ ]python · cell 163
python
# Evaluate on test split
performance = evaluate(ds=test_ds, predictor=predictor)
print (json.dumps(performance, indent=2))Output
2023-09-17 22:45:42,824 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(<lambda>)] 2023-09-17 22:45:42,824 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:42,825 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
2023-09-17 22:45:43,740 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor)] 2023-09-17 22:45:43,741 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:43,742 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
[2m[36m(MapBatches(preprocess)->MapBatches(TorchPredictor) pid=350325, ip=10.0.34.101)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
{
"precision": 0.9168092951711627,
"recall": 0.9109947643979057,
"f1": 0.9105512639658029
}
In [ ]python · cell 164
python
# Predict on sample
title = "Transfer learning with transformers"
description = "Using transformers for transfer learning on text classification tasks."
sample_ds = ray.data.from_items([{"title": title, "description": description, "tag": "other"}])
predict_proba(ds=sample_ds, predictor=predictor)Output
2023-09-17 22:45:49,875 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:45:49,875 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:49,876 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[{'prediction': 'natural-language-processing',
'probabilities': {'computer-vision': 0.00043165276,
'mlops': 0.0008155016,
'natural-language-processing': 0.9978746,
'other': 0.0008782037}}]⚖️ Evaluation
So far we've been evaluating our models by determing the overall precision, recall and f1 scores. But since performance is one of the key decision making factors when comparing different models, we should have even more nuanced evaluation strategies.
- Coarse-grained metrics
- Fine-grained metrics
- Confusion matrix
- Confidence learning
- Slice metrics
In [ ]python · cell 167
python
# Metrics
metrics = {"overall": {}, "class": {}}In [ ]python · cell 168
python
# Artifacts
predictor = TorchPredictor.from_checkpoint(best_checkpoint)
preprocessor = predictor.get_preprocessor()In [ ]python · cell 169
python
# y_test
preprocessed_ds = preprocessor.transform(test_ds)
values = preprocessed_ds.select_columns(cols=["targets"]).take_all()
y_test = np.stack([item["targets"] for item in values])Output
2023-09-17 22:45:52,090 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(<lambda>)] 2023-09-17 22:45:52,091 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:52,091 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
In [ ]python · cell 170
python
# y_prob
outputs = preprocessed_ds.map_batches(predictor.predict_proba)
y_prob = np.array([d["output"] for d in outputs.take_all()])Output
2023-09-17 22:45:52,987 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:45:52,988 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:52,988 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
In [ ]python · cell 171
python
# y_prob
print (np.shape(y_test))
print (np.shape(y_prob))Output
(191,) (191, 4)
In [ ]python · cell 172
python
# Add columns (for convenience)
test_df = test_ds.to_pandas()
test_df["text"] = test_df["title"] + " " + test_df["description"]
test_df["prediction"] = test_df.index.map(lambda i: preprocessor.index_to_class[y_pred[i]])
test_df.head()Output
Read progress 0: 0%| | 0/1 [00:00<?, ?it/s]
Read progress 0: 0%| | 0/1 [00:00<?, ?it/s]
id created_on title \
0 19 2020-03-03 13:54:31 Diffusion to Vector
1 26 2020-03-07 23:11:58 Graph Wavelet Neural Network
2 44 2020-03-08 00:32:58 Capsule Graph Neural Network
3 80 2020-03-20 05:59:32 NeRF: Neural Radiance Fields
4 84 2020-03-20 15:18:43 Mention Classifier
description \
0 Reference implementation of Diffusion2Vec (Com...
1 A PyTorch implementation of "Graph Wavelet Neu...
2 A PyTorch implementation of "Capsule Graph Neu...
3 Representing scenes as neural radiance fields ...
4 Category prediction model\nThis repo contains ...
tag \
0 other
1 other
2 other
3 computer-vision
4 natural-language-processing
text \
0 Diffusion to Vector Reference implementation o...
1 Graph Wavelet Neural Network A PyTorch impleme...
2 Capsule Graph Neural Network A PyTorch impleme...
3 NeRF: Neural Radiance Fields Representing scen...
4 Mention Classifier Category prediction model\n...
prediction
0 computer-vision
1 other
2 other
3 computer-vision
4 natural-language-processing
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| id | created_on | title | description | tag | text | prediction | |
|---|---|---|---|---|---|---|---|
| 0 | 19 | 2020-03-03 13:54:31 | Diffusion to Vector | Reference implementation of Diffusion2Vec (Com... | other | Diffusion to Vector Reference implementation o... | computer-vision |
| 1 | 26 | 2020-03-07 23:11:58 | Graph Wavelet Neural Network | A PyTorch implementation of "Graph Wavelet Neu... | other | Graph Wavelet Neural Network A PyTorch impleme... | other |
| 2 | 44 | 2020-03-08 00:32:58 | Capsule Graph Neural Network | A PyTorch implementation of "Capsule Graph Neu... | other | Capsule Graph Neural Network A PyTorch impleme... | other |
| 3 | 80 | 2020-03-20 05:59:32 | NeRF: Neural Radiance Fields | Representing scenes as neural radiance fields ... | computer-vision | NeRF: Neural Radiance Fields Representing scen... | computer-vision |
| 4 | 84 | 2020-03-20 15:18:43 | Mention Classifier | Category prediction model\nThis repo contains ... | natural-language-processing | Mention Classifier Category prediction model\n... | natural-language-processing |
Coarse-grained metrics
In [ ]python · cell 174
python
# Overall metrics
overall_metrics = precision_recall_fscore_support(y_test, y_pred, average="weighted")
metrics["overall"]["precision"] = overall_metrics[0]
metrics["overall"]["recall"] = overall_metrics[1]
metrics["overall"]["f1"] = overall_metrics[2]
metrics["overall"]["num_samples"] = np.float64(len(y_test))
print (json.dumps(metrics["overall"], indent=4))Output
{
"precision": 0.9138952286238713,
"recall": 0.9109947643979057,
"f1": 0.9114851103432928,
"num_samples": 191.0
}
Fine-grained metrics
In [ ]python · cell 176
python
from collections import OrderedDictIn [ ]python · cell 177
python
# Per-class metrics
class_metrics = precision_recall_fscore_support(y_test, y_pred, average=None)
for i, _class in enumerate(preprocessor.class_to_index):
metrics["class"][_class] = {
"precision": class_metrics[0][i],
"recall": class_metrics[1][i],
"f1": class_metrics[2][i],
"num_samples": np.float64(class_metrics[3][i]),
}In [ ]python · cell 178
python
# Metrics for a specific class
tag = "natural-language-processing"
print (json.dumps(metrics["class"][tag], indent=2))Output
{
"precision": 0.9146341463414634,
"recall": 0.9615384615384616,
"f1": 0.9375000000000001,
"num_samples": 78.0
}
In [ ]python · cell 179
python
# Sorted tags
sorted_tags_by_f1 = OrderedDict(sorted(
metrics["class"].items(), key=lambda tag: tag[1]["f1"], reverse=True))
for item in sorted_tags_by_f1.items():
print (json.dumps(item, indent=2))Output
[
"other",
{
"precision": 0.96,
"recall": 0.9230769230769231,
"f1": 0.9411764705882353,
"num_samples": 26.0
}
]
[
"natural-language-processing",
{
"precision": 0.9146341463414634,
"recall": 0.9615384615384616,
"f1": 0.9375000000000001,
"num_samples": 78.0
}
]
[
"computer-vision",
{
"precision": 0.9393939393939394,
"recall": 0.8732394366197183,
"f1": 0.9051094890510948,
"num_samples": 71.0
}
]
[
"mlops",
{
"precision": 0.7222222222222222,
"recall": 0.8125,
"f1": 0.7647058823529411,
"num_samples": 16.0
}
]
Confusion matrix
- True positives (TP): learn about where our model performs well.
- False positives (FP): potentially identify samples which may need to be relabeled.
- False negatives (FN): identify the model's less performant areas to oversample later.
It's a good to have our FP/FN samples feed back into our annotation pipelines in the event we want to fix their labels and have those changes be reflected everywhere.
In [ ]python · cell 182
python
# TP, FP, FN samples
tag = "natural-language-processing"
index = preprocessor.class_to_index[tag]
tp, fp, fn = [], [], []
for i, true in enumerate(y_test):
pred = y_pred[i]
if index==true==pred:
tp.append(i)
elif index!=true and index==pred:
fp.append(i)
elif index==true and index!=pred:
fn.append(i)In [ ]python · cell 183
python
print (tp)
print (fp)
print (fn)Output
[4, 9, 12, 17, 19, 23, 25, 26, 29, 30, 31, 32, 33, 34, 42, 47, 49, 50, 54, 56, 65, 66, 68, 71, 75, 77, 78, 79, 82, 92, 94, 95, 97, 99, 101, 109, 113, 114, 115, 118, 120, 122, 126, 128, 129, 130, 131, 133, 134, 135, 138, 139, 140, 141, 142, 144, 148, 149, 152, 159, 160, 161, 163, 166, 170, 172, 173, 174, 177, 179, 183, 184, 187, 189, 190] [41, 61, 102, 104, 154, 165, 188] [16, 76, 112]
In [ ]python · cell 184
python
# Samples
num_samples = 3
cm = [(tp, "True positives"), (fp, "False positives"), (fn, "False negatives")]
for item in cm:
if len(item[0]):
print (f"\n=== {item[1]} ===")
for index in item[0][:num_samples]:
print (f"{test_df.iloc[index].text}")
print (f" true: {test_df.tag[index]}")
print (f" pred: {test_df.prediction[index]}\n")Output
=== True positives ===
Mention Classifier Category prediction model
This repo contains AllenNLP model for prediction of Named Entity categories by its mentions.
true: natural-language-processing
pred: natural-language-processing
Finetune: Scikit-learn Style Model Finetuning for NLP Finetune is a library that allows users to leverage state-of-the-art pretrained NLP models for a wide variety of downstream tasks.
true: natural-language-processing
pred: natural-language-processing
Finetuning Transformers with JAX + Haiku Walking through a port of the RoBERTa pre-trained model to JAX + Haiku, then fine-tuning the model to solve a downstream task.
true: natural-language-processing
pred: natural-language-processing
=== False positives ===
How Docker Can Help You Become A More Effective Data Scientist A look at Docker from the perspective of a data scientist.
true: mlops
pred: natural-language-processing
Differential Subspace Search in High-Dimensional Latent Space Differential subspace search to allow efficient iterative user exploration in such a space, without relying on domain- or data-specific assumptions.
true: computer-vision
pred: natural-language-processing
EfficientDet (PyTorch) A PyTorch implementation of EfficientDet faithful to the original Google implementation with ported weights.
true: computer-vision
pred: natural-language-processing
=== False negatives ===
The Unreasonable Effectiveness of Recurrent Neural Networks A close look at how RNNs are able to perform so well.
true: natural-language-processing
pred: computer-vision
Get Subreddit Suggestions for a Post Trained on 4M Reddit posts from 4k Subreddits. End-to-end ML pipeline built with fasttext and FastAPI, deployed to Valohai.
true: natural-language-processing
pred: computer-vision
Machine Learning Projects This Repo contains projects done by me while learning the basics. All the familiar types of regression, classification, and clustering methods have been used.
true: natural-language-processing
pred: mlops
Confidence learning
While the confusion-matrix sample analysis was a coarse-grained process, we can also use fine-grained confidence based approaches to identify potentially mislabeled samples. Here we’re going to focus on the specific labeling quality as opposed to the final model predictions.
Simple confidence based techniques include identifying samples whose:
Categorical
- prediction is incorrect (also indicate TN, FP, FN)
- confidence score for the correct class is below a threshold
- confidence score for an incorrect class is above a threshold
- standard deviation of confidence scores over top N samples is low
- different predictions from same model using different parameters
Continuous
- difference between predicted and ground-truth values is above some %
The operations in this section can be applied to entire labeled dataset to discover labeling errors via confidence learning.
In [ ]python · cell 188
python
# Tag to inspect
tag = "natural-language-processing"
index = class_to_index[tag]
indices = np.where(y_test==index)[0]In [ ]python · cell 189
python
# Confidence score for the correct class is below a threshold
low_confidence = []
min_threshold = 0.5
for i in indices:
prob = y_prob[i][index]
if prob <= 0.5:
low_confidence.append({
"text": f"{test_df.iloc[i].text}",
"true": test_df.tag[i],
"pred": test_df.prediction[i],
"prob": prob})In [ ]python · cell 190
python
low_confidence[0:3]Output
[{'text': 'The Unreasonable Effectiveness of Recurrent Neural Networks A close look at how RNNs are able to perform so well.',
'true': 'natural-language-processing',
'pred': 'computer-vision',
'prob': 0.008757317},
{'text': 'Machine Learning Projects This Repo contains projects done by me while learning the basics. All the familiar types of regression, classification, and clustering methods have been used.',
'true': 'natural-language-processing',
'pred': 'mlops',
'prob': 0.020190474}]But these are fairly crude techniques because neural networks are easily overconfident and so their confidences cannot be used without calibrating them.
- Assumption: “the probability associated with the predicted class label should reflect its ground truth correctness likelihood.”
- Reality: “modern (large) neural networks are no longer well-calibrated”
- Solution: apply temperature scaling (extension of Platt scaling{:target="_blank"}) on model outputs
Recent work on confident learning focuses on identifying noisy labels while accounting for this overconfidence which can then be properly relabeled and used for training.
In [ ]python · cell 192
python
import cleanlab
from cleanlab.filter import find_label_issuesIn [ ]python · cell 193
python
# Find label issues
label_issues = find_label_issues(labels=y_test, pred_probs=y_prob, return_indices_ranked_by="self_confidence")
test_df.iloc[label_issues].drop(columns=["text"]).head()Output
*** SIGTERM received at time=1695015958 on cpu 1 ***
PC: @ 0x4edd8a (unknown) _PyEval_MakeFrameVector
@ 0x7fe3ff0e9420 514711536 (unknown)
@ 0x72c720 (unknown) (unknown)
[2023-09-17 22:45:58,798 E 848689 841208] logging.cc:361: *** SIGTERM received at time=1695015958 on cpu 1 ***
[2023-09-17 22:45:58,799 E 848689 841208] logging.cc:361: PC: @ 0x4edd8a (unknown) _PyEval_MakeFrameVector
[2023-09-17 22:45:58,803 E 848689 841208] logging.cc:361: @ 0x7fe3ff0e9420 514711536 (unknown)
[2023-09-17 22:45:58,808 E 848689 841208] logging.cc:361: @ 0x72c720 (unknown) (unknown)
id created_on \
103 1459 2020-06-16 03:06:10
102 1451 2020-06-16 01:21:09
165 2137 2020-08-13 02:10:03
61 710 2020-05-05 04:01:24
16 264 2020-04-06 21:33:32
title \
103 SuperGlue: Learning Feature Matching with Grap...
102 EfficientDet (PyTorch)
165 Unpopular Opinion - Data Scientists Should Be ...
61 Differential Subspace Search in High-Dimension...
16 The Unreasonable Effectiveness of Recurrent Ne...
description \
103 SuperGlue, a neural network that matches two s...
102 A PyTorch implementation of EfficientDet faith...
165 I believe data scientists can be more effectiv...
61 Differential subspace search to allow efficien...
16 A close look at how RNNs are able to perform s...
tag prediction
103 other computer-vision
102 computer-vision natural-language-processing
165 mlops natural-language-processing
61 computer-vision natural-language-processing
16 natural-language-processing computer-vision
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| id | created_on | title | description | tag | prediction | |
|---|---|---|---|---|---|---|
| 103 | 1459 | 2020-06-16 03:06:10 | SuperGlue: Learning Feature Matching with Grap... | SuperGlue, a neural network that matches two s... | other | computer-vision |
| 102 | 1451 | 2020-06-16 01:21:09 | EfficientDet (PyTorch) | A PyTorch implementation of EfficientDet faith... | computer-vision | natural-language-processing |
| 165 | 2137 | 2020-08-13 02:10:03 | Unpopular Opinion - Data Scientists Should Be ... | I believe data scientists can be more effectiv... | mlops | natural-language-processing |
| 61 | 710 | 2020-05-05 04:01:24 | Differential Subspace Search in High-Dimension... | Differential subspace search to allow efficien... | computer-vision | natural-language-processing |
| 16 | 264 | 2020-04-06 21:33:32 | The Unreasonable Effectiveness of Recurrent Ne... | A close look at how RNNs are able to perform s... | natural-language-processing | computer-vision |
Not all of these are necessarily labeling errors but situations where the predicted probabilities were not so confident. Therefore, it will be useful to attach the predictions alongside the data. This way, we can know if we need to relabel, upsample, etc. to improve our performance. Analysis like this could also shed light on the task itself. For example, you may notice that some projects involve multiple data modalities and so it's difficult to just assing one tag. So perhaps it might be better to make this taks a multilabel classification task instead (it does but we simplified it for this course).
Slice metrics
Just inspecting the overall and class metrics isn't enough to deploy our new version to production. There may be key slices of our dataset that we need to do really well on:
- Target / predicted classes (+ combinations)
- Features (explicit and implicit)
- Metadata (timestamps, sources, etc.)
- Priority slices / experience (minority groups, large customers, etc.)
An easy way to create and evaluate slices is to define slicing functions.
In [ ]python · cell 197
python
from snorkel.slicing import PandasSFApplier
from snorkel.slicing import slice_dataframe
from snorkel.slicing import slicing_functionIn [ ]python · cell 198
python
@slicing_function()
def nlp_llm(x):
"""NLP projects that use LLMs."""
nlp_project = "natural-language-processing" in x.tag
llm_terms = ["transformer", "llm", "bert"]
llm_project = any(s.lower() in x.text.lower() for s in llm_terms)
return (nlp_project and llm_project)In [ ]python · cell 199
python
@slicing_function()
def short_text(x):
"""Projects with short titles and descriptions."""
return len(x.text.split()) < 8 # less than 8 wordsHere we're using Snorkel's slicing_function to create our different slices. We can visualize our slices by applying this slicing function to a relevant DataFrame using slice_dataframe.
In [ ]python · cell 201
python
nlp_llm_df = slice_dataframe(test_df, nlp_llm)
nlp_llm_df[["text", "tag"]].head()Output
100%|██████████| 191/191 [00:00<00:00, 32805.57it/s]
text \
12 Finetuning Transformers with JAX + Haiku Walki...
19 Question Answering with a Fine-Tuned BERT What...
29 BertViz Tool for visualizing attention in the ...
30 The Transformer Family This post presents how ...
31 Pruning Bert to Accelerate Inference After pre...
tag
12 natural-language-processing
19 natural-language-processing
29 natural-language-processing
30 natural-language-processing
31 natural-language-processing
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| text | tag | |
|---|---|---|
| 12 | Finetuning Transformers with JAX + Haiku Walki... | natural-language-processing |
| 19 | Question Answering with a Fine-Tuned BERT What... | natural-language-processing |
| 29 | BertViz Tool for visualizing attention in the ... | natural-language-processing |
| 30 | The Transformer Family This post presents how ... | natural-language-processing |
| 31 | Pruning Bert to Accelerate Inference After pre... | natural-language-processing |
In [ ]python · cell 202
python
short_text_df = slice_dataframe(test_df, short_text)
short_text_df[["text", "tag"]].head()Output
100%|██████████| 191/191 [00:00<00:00, 64413.61it/s]
text \
75 NLPAug Data augmentation for NLP
123 Offline Reinforcement Learning Challenges, alg...
127 Image Classifier Pure JavaScript Image Classifier
132 imgaug Image augmentation for machine learning...
140 QSVM Quantum SVM for sentiment analysis
tag
75 natural-language-processing
123 other
127 computer-vision
132 computer-vision
140 natural-language-processing
.dataframe tbody tr th:only-of-type {
vertical-align: middle;
}
.dataframe tbody tr th {
vertical-align: top;
}
.dataframe thead th {
text-align: right;
}
| text | tag | |
|---|---|---|
| 75 | NLPAug Data augmentation for NLP | natural-language-processing |
| 123 | Offline Reinforcement Learning Challenges, alg... | other |
| 127 | Image Classifier Pure JavaScript Image Classifier | computer-vision |
| 132 | imgaug Image augmentation for machine learning... | computer-vision |
| 140 | QSVM Quantum SVM for sentiment analysis | natural-language-processing |
We can define even more slicing functions and create a slices record array using the PandasSFApplier. The slices array has N (# of data points) items and each item has S (# of slicing functions) items, indicating whether that data point is part of that slice. Think of this record array as a masking layer for each slicing function on our data.
In [ ]python · cell 204
python
# Slices
slicing_functions = [nlp_llm, short_text]
applier = PandasSFApplier(slicing_functions)
slices = applier.apply(test_df)
slicesOutput
100%|██████████| 191/191 [00:00<00:00, 27137.02it/s]
rec.array([(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (1, 0), (1, 0), (1, 0),
(1, 0), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 1), (0, 0), (0, 0), (1, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (1, 0), (0, 0),
(0, 0), (1, 0), (0, 0), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (1, 0), (1, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(1, 0), (0, 0), (0, 0), (0, 1), (0, 0), (0, 0), (0, 0), (0, 1),
(1, 0), (0, 0), (1, 0), (1, 0), (0, 1), (1, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (1, 0), (0, 1), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 1), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(1, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(1, 0), (1, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (1, 0), (0, 0), (0, 0),
(0, 0), (1, 0), (0, 0), (0, 0), (0, 1), (0, 0), (0, 0), (0, 0),
(0, 0), (0, 0), (0, 0), (0, 0), (0, 0), (1, 0), (0, 0)],
dtype=[('nlp_llm', '<i8'), ('short_text', '<i8')])To calculate metrics for our slices, we could use snorkel.analysis.Scorer but we've implemented a version that will work for multiclass or multilabel scenarios.
In [ ]python · cell 206
python
# Score slices
metrics["slices"] = {}
for slice_name in slices.dtype.names:
mask = slices[slice_name].astype(bool)
if sum(mask):
slice_metrics = precision_recall_fscore_support(
y_test[mask], y_pred[mask], average="micro"
)
metrics["slices"][slice_name] = {}
metrics["slices"][slice_name]["precision"] = slice_metrics[0]
metrics["slices"][slice_name]["recall"] = slice_metrics[1]
metrics["slices"][slice_name]["f1"] = slice_metrics[2]
metrics["slices"][slice_name]["num_samples"] = len(y_test[mask])In [ ]python · cell 207
python
print(json.dumps(metrics["slices"], indent=2))Output
{
"nlp_llm": {
"precision": 1.0,
"recall": 1.0,
"f1": 1.0,
"num_samples": 28
},
"short_text": {
"precision": 0.8571428571428571,
"recall": 0.8571428571428571,
"f1": 0.8571428571428571,
"num_samples": 7
}
}
Slicing can help identify sources of bias in our data. For example, our model has most likely learned to associated algorithms with certain applications such as CNNs used for computer vision or transformers used for NLP projects. However, these algorithms are not being applied beyond their initial use cases. We’d need ensure that our model learns to focus on the application over algorithm. This could be learned with:
- enough data (new or oversampling incorrect predictions)
- masking the algorithm (using text matching heuristics).
Interpretability
Besides just comparing predicted outputs with ground truth values, we can also inspect the inputs to our models. What aspects of the input are more influential towards the prediction? If the focus is not on the relevant features of our input, then we need to explore if there is a hidden pattern we're missing or if our model has learned to overfit on the incorrect features. We can use techniques such as SHAP (SHapley Additive exPlanations) or LIME (Local Interpretable Model-agnostic Explanations) to inspect feature importance. On a high level, these techniques learn which features have the most signal by assessing the performance in their absence. These inspections can be performed on a global level (ex. per-class) or on a local level (ex. single prediction).
In [ ]python · cell 211
python
from lime.lime_text import LimeTextExplainer
from sklearn.pipeline import make_pipelineLimeTextExplainer.explain_instance function requires a classifier_fn that takes in a list of strings and outputs the predicted probabilities.
In [ ]python · cell 213
python
def classifier_fn(texts):
ds = ray.data.from_items([{"title": text, "description": "", "tag": "other"} for text in texts])
preprocessed_ds = preprocessor.transform(ds)
outputs = preprocessed_ds.map_batches(predictor.predict_proba)
y_prob = np.array([d["output"] for d in outputs.take_all()])
return y_probIn [ ]python · cell 214
python
# Explain instance
text = "Using pretrained convolutional neural networks for object detection."
explainer = LimeTextExplainer(class_names=list(class_to_index.keys()))
explainer.explain_instance(text, classifier_fn=classifier_fn, top_labels=1).show_in_notebook(text=True)Output
2023-09-17 22:45:59,447 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:45:59,447 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:45:59,448 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/200 [00:00<?, ?it/s]
<IPython.core.display.HTML object>
Behavioral testing
In [ ]python · cell 216
python
# INVariance via verb injection (changes should not affect outputs)
tokens = ["revolutionized", "disrupted"]
texts = [f"Transformers applied to NLP have {token} the ML field." for token in tokens]
[preprocessor.index_to_class[y_prob.argmax()] for y_prob in classifier_fn(texts=texts)]Output
2023-09-17 22:46:15,716 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:46:15,717 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:46:15,718 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/2 [00:00<?, ?it/s]
['natural-language-processing', 'natural-language-processing']
In [ ]python · cell 217
python
# DIRectional expectations (changes with known outputs)
tokens = ["text classification", "image classification"]
texts = [f"ML applied to {token}." for token in tokens]
[preprocessor.index_to_class[y_prob.argmax()] for y_prob in classifier_fn(texts=texts)]Output
2023-09-17 22:46:16,783 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:46:16,784 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:46:16,784 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/2 [00:00<?, ?it/s]
['natural-language-processing', 'computer-vision']
In [ ]python · cell 218
python
# Minimum Functionality Tests (simple input/output pairs)
tokens = ["natural language processing", "mlops"]
texts = [f"{token} is the next big wave in machine learning." for token in tokens]
[preprocessor.index_to_class[y_prob.argmax()] for y_prob in classifier_fn(texts=texts)]Output
2023-09-17 22:46:17,810 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] 2023-09-17 22:46:17,811 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:46:17,811 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
Running 0: 0%| | 0/2 [00:00<?, ?it/s]
['natural-language-processing', 'mlops']
We'll learn how to systematically create tests in our testing lesson. Be sure to also checkout the evaluation lesson where we cover more ways to evaluate our model, including generating slices, counterfactuals and more.
🚀 Serving
Batch inference (offline)
In [ ]python · cell 222
python
import ray.data
from ray.data import ActorPoolStrategyIn [ ]python · cell 223
python
# Artifacts
run_id = sorted_runs.iloc[0].run_id
best_checkpoint = get_best_checkpoint(run_id=run_id)
predictor = TorchPredictor.from_checkpoint(best_checkpoint)
preprocessor = predictor.get_preprocessor()In [ ]python · cell 224
python
# Batch inference
preprocessed_ds = preprocessor.transform(test_ds)
compute = ActorPoolStrategy(min_size=1, max_size=2)
outputs = preprocessed_ds.map_batches(predictor, batch_size=128, compute=compute)
np.array([d["output"] for d in outputs.take_all()])Output
2023-09-17 22:46:24,211 INFO streaming_executor.py:93 -- Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[ReadCSV->SplitBlocks(64)] -> ActorPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor)] 2023-09-17 22:46:24,213 INFO streaming_executor.py:94 -- Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=True, actor_locality_enabled=True, verbose_progress=False) 2023-09-17 22:46:24,214 INFO streaming_executor.py:96 -- Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True` 2023-09-17 22:46:24,951 INFO actor_pool_map_operator.py:106 -- MapBatches(preprocess)->MapBatches(TorchPredictor): Waiting for 1 pool actors to start...
Running 0: 0%| | 0/4096 [00:00<?, ?it/s]
[2m[36m(MapWorker(MapBatches(preprocess)->MapBatches(TorchPredictor)) pid=848995)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
array([0, 3, 3, 0, 2, 0, 0, 0, 0, 2, 0, 0, 2, 3, 0, 0, 0, 2, 3, 2, 3, 0,
3, 2, 0, 0, 2, 1, 1, 2, 2, 2, 2, 2, 2, 0, 0, 0, 0, 0, 1, 2, 2, 0,
0, 3, 1, 2, 0, 2, 2, 3, 3, 0, 2, 1, 2, 3, 3, 3, 3, 2, 0, 0, 0, 2,
2, 3, 2, 1, 0, 2, 3, 1, 0, 2, 0, 2, 2, 2, 0, 0, 2, 1, 1, 0, 0, 0,
0, 3, 0, 0, 2, 0, 2, 2, 3, 2, 0, 2, 0, 2, 2, 0, 1, 0, 0, 0, 0, 2,
0, 0, 1, 2, 2, 2, 3, 0, 2, 0, 2, 3, 2, 3, 3, 3, 2, 0, 2, 2, 2, 2,
0, 2, 2, 2, 0, 1, 2, 2, 2, 2, 2, 1, 2, 0, 3, 0, 2, 2, 1, 1, 2, 0,
2, 0, 0, 0, 0, 2, 2, 2, 0, 2, 1, 2, 2, 0, 0, 1, 2, 3, 2, 2, 2, 0,
0, 2, 0, 2, 1, 3, 0, 2, 2, 0, 1, 2, 1, 2, 2])Online inference (real-time)
While we can achieve batch inference at scale, many models will need to be served in an real-time manner where we may need to deliver predictions for many incoming requests (high throughput) with low latency.
In [ ]python · cell 227
python
from fastapi import FastAPI
from ray import serve
import requests
from starlette.requests import RequestIn [ ]python · cell 228
python
# Define application
app = FastAPI(
title="Made With ML",
description="Classify machine learning projects.",
version="0.1")We'll start by defining our FastAPI application which involves initializing a predictor (and preprocessor) from the best checkpoint for a particular run (specified by run_id). We'll also define a predict function that will be used to make predictions on our input data.
In [ ]python · cell 230
python
@serve.deployment(num_replicas="1", ray_actor_options={"num_cpus": 8, "num_gpus": 0})
@serve.ingress(app)
class ModelDeployment:
def __init__(self, run_id):
"""Initialize the model."""
self.run_id = run_id
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) # so workers have access to model registry
best_checkpoint = get_best_checkpoint(run_id=run_id)
self.predictor = TorchPredictor.from_checkpoint(best_checkpoint)
@app.post("/predict/")
async def _predict(self, request: Request):
data = await request.json()
sample_ds = ray.data.from_items([{"title": data.get("title", ""), "description": data.get("description", ""), "tag": ""}])
results = predict_proba(ds=sample_ds, predictor=self.predictor)
return {"results": results}
async defrefers to an asynchronous function (when we call the function we don't have to wait for the function to complete executing). Theawaitkeyword is used inside an asynchronous function to wait for the completion of therequest.json()operation.
In [ ]python · cell 232
python
# Run service
sorted_runs = mlflow.search_runs(experiment_names=[experiment_name], order_by=["metrics.val_loss ASC"])
run_id = sorted_runs.iloc[0].run_id
serve.run(ModelDeployment.bind(run_id=run_id), route_prefix="/")Output
Connecting to existing Serve app in namespace "serve". New http options will not be applied.
[2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:46,923 controller 840558 deployment_state.py:1390 - Deploying new version of deployment ModelDeployment in application 'default'. [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:47,027 controller 840558 deployment_state.py:1560 - Stopping 1 replicas of deployment 'ModelDeployment' in application 'default' with outdated versions. [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:47,027 controller 840558 http_state.py:265 - Start to drain the proxy actor on node 860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:49,260 controller 840558 deployment_state.py:2027 - Replica default#ModelDeployment#qtfPwF is stopped. [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:49,261 controller 840558 deployment_state.py:1679 - Adding 1 replica to deployment ModelDeployment in application 'default'. [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:46:52,625 controller 840558 http_state.py:276 - Stop draining the proxy actor on node 860a06117a62bfc12416a8163d19974a865e594674344a920da1e53b
Deployment 'ModelDeployment:fHCaeN' is ready at `http://127.0.0.1:8000/`. component=serve deployment=ModelDeployment
2023-09-17 22:46:58,941 INFO router.py:1226 -- Using router <class 'ray.serve._private.router.PowerOfTwoChoicesReplicaScheduler'>.
2023-09-17 22:46:58,947 INFO router.py:537 -- Got updated replicas for deployment 'ModelDeployment' in application 'default': {'default#ModelDeployment#aAAvxd'}.
RayServeSyncHandle(deployment='ModelDeployment')
In [ ]python · cell 233
python
# Query
title = "Transfer learning with transformers"
description = "Using transformers for transfer learning on text classification tasks."
json_data = json.dumps({"title": title, "description": description})
requests.post("http://127.0.0.1:8000/predict/", data=json_data).json()Output
[2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=False, actor_locality_enabled=True, verbose_progress=False) [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=350914, ip=10.0.34.101) Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[2m[36m(MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba) pid=351034, ip=10.0.34.101)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
{'results': [{'prediction': 'natural-language-processing',
'probabilities': {'computer-vision': 0.0006470938096754253,
'mlops': 0.0003823915321845561,
'natural-language-processing': 0.9983959794044495,
'other': 0.0005744708469137549}}]}The issue with neural networks (and especially LLMs) is that they are notoriously overconfident. For every input, they will always make some prediction. And to account for this, we have an other class but that class only has projects that are not in our accepted tags but are still machine learning related nonetheless. Here's what happens when we input complete noise as our input:
In [ ]python · cell 235
python
# Query (noise)
title = "this is random noise" # random noise
json_data = json.dumps({"title": title, "description": ""})
requests.post("http://127.0.0.1:8000/predict/", data=json_data).json()Output
[2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m INFO 2023-09-17 22:47:02,865 ModelDeployment default#ModelDeployment#aAAvxd 7abfafdb-67ae-4204-a021-9f7507e22f78 /predict/ default replica.py:749 - __CALL__ OK 3898.9ms [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=False, actor_locality_enabled=True, verbose_progress=False) [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=350914, ip=10.0.34.101) Running 0: 0%| | 0/1 [00:00<?, ?it/s]
[2m[36m(MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba) pid=849454)[0m /tmp/ipykernel_841208/1209796013.py:7: UserWarning: The given NumPy array is not writable, and PyTorch does not support non-writable tensors. This means writing to this tensor will result in undefined behavior. You may want to copy the array to protect its data or make it writable before converting it to a tensor. This type of warning will be suppressed for the rest of this program. (Triggered internally at ../torch/csrc/utils/tensor_numpy.cpp:206.)
{'results': [{'prediction': 'other',
'probabilities': {'computer-vision': 0.010165845975279808,
'mlops': 0.2043360322713852,
'natural-language-processing': 0.05693763121962547,
'other': 0.7285605072975159}}]}In [ ]python · cell 236
python
# Shutdown
serve.shutdown()Output
2023-09-17 22:47:06,943 INFO router.py:537 -- Got updated replicas for deployment 'ModelDeployment' in application 'default': set(). [2m[36m(ServeReplica:default:ModelDeployment pid=350914, ip=10.0.34.101)[0m INFO 2023-09-17 22:47:06,917 ModelDeployment default#ModelDeployment#aAAvxd 3c3a4798-af4d-45f3-a2b1-c73d15e1ce83 /predict/ default replica.py:749 - __CALL__ OK 4037.2ms [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:47:06,939 controller 840558 deployment_state.py:1707 - Removing 1 replica from deployment 'ModelDeployment' in application 'default'. [2m[36m(ServeController pid=840558)[0m INFO 2023-09-17 22:47:09,134 controller 840558 deployment_state.py:2027 - Replica default#ModelDeployment#aAAvxd is stopped.
Custom logic
To make our service a bit more robust, let's add some custom logic to predict the other class if the probability of the predicted class is below a certain threshold probability.
In [ ]python · cell 239
python
@serve.deployment(num_replicas="1", ray_actor_options={"num_cpus": 8, "num_gpus": 0})
@serve.ingress(app)
class ModelDeploymentRobust:
def __init__(self, run_id, threshold=0.9):
"""Initialize the model."""
self.run_id = run_id
self.threshold = threshold
mlflow.set_tracking_uri(MLFLOW_TRACKING_URI) # so workers have access to model registry
best_checkpoint = get_best_checkpoint(run_id=run_id)
self.predictor = TorchPredictor.from_checkpoint(best_checkpoint)
@app.post("/predict/")
async def _predict(self, request: Request):
data = await request.json()
sample_ds = ray.data.from_items([{"title": data.get("title", ""), "description": data.get("description", ""), "tag": ""}])
results = predict_proba(ds=sample_ds, predictor=self.predictor)
# Apply custom logic
for i, result in enumerate(results):
pred = result["prediction"]
prob = result["probabilities"]
if prob[pred] < self.threshold:
results[i]["prediction"] = "other"
return {"results": results}In [ ]python · cell 240
python
# Run service
serve.run(ModelDeploymentRobust.bind(run_id=run_id, threshold=0.9), route_prefix="/")Output
[2m[36m(ServeController pid=850735)[0m INFO 2023-09-17 22:48:54,599 controller 850735 application_state.py:183 - Recovering target state for application 'default' from checkpoint.
Started detached Serve instance in namespace "serve".
[2m[36m(HTTPProxyActor pid=850784)[0m INFO 2023-09-17 22:48:55,884 http_proxy 10.0.35.174 http_proxy.py:1441 - Proxy actor 8bf33b9832cc186ec5c3be470b000000 starting on node c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a. [2m[36m(HTTPProxyActor pid=850784)[0m INFO 2023-09-17 22:48:55,890 http_proxy 10.0.35.174 http_proxy.py:1626 - Starting HTTP server on node: c268d87a05f14889a7bdc3c259e78ced37c574c65ab440657e85be5a listening on port 8000 [2m[36m(HTTPProxyActor pid=850784)[0m INFO: Started server process [850784] [2m[36m(ServeController pid=850735)[0m INFO 2023-09-17 22:48:56,036 controller 850735 deployment_state.py:1390 - Deploying new version of deployment ModelDeploymentRobust in application 'default'. [2m[36m(ServeController pid=850735)[0m INFO 2023-09-17 22:48:56,138 controller 850735 deployment_state.py:1679 - Adding 1 replica to deployment ModelDeploymentRobust in application 'default'.
Deployment 'ModelDeploymentRobust:tMjDvR' is ready at `http://127.0.0.1:8000/`. component=serve deployment=ModelDeploymentRobust
2023-09-17 22:49:02,011 INFO router.py:1226 -- Using router <class 'ray.serve._private.router.PowerOfTwoChoicesReplicaScheduler'>.
2023-09-17 22:49:02,015 INFO router.py:537 -- Got updated replicas for deployment 'ModelDeploymentRobust' in application 'default': {'default#ModelDeploymentRobust#YziEno'}.
RayServeSyncHandle(deployment='ModelDeploymentRobust')
In [ ]python · cell 241
python
# Query (noise)
title = "this is random noise" # random noise
json_data = json.dumps({"title": title, "description": ""})
requests.post("http://127.0.0.1:8000/predict/", data=json_data).json()Output
[2m[36m(ServeReplica:default:ModelDeploymentRobust pid=850829)[0m Executing DAG InputDataBuffer[Input] -> TaskPoolMapOperator[MapBatches(preprocess)->MapBatches(TorchPredictor.predict_proba)] [2m[36m(ServeReplica:default:ModelDeploymentRobust pid=850829)[0m Execution config: ExecutionOptions(resource_limits=ExecutionResources(cpu=None, gpu=None, object_store_memory=None), locality_with_output=False, preserve_order=False, actor_locality_enabled=True, verbose_progress=False) [2m[36m(ServeReplica:default:ModelDeploymentRobust pid=850829)[0m Tip: For detailed progress reporting, run `ray.data.DataContext.get_current().execution_options.verbose_progress = True`
(pid=850829) Running 0: 0%| | 0/1 [00:00<?, ?it/s]
{'results': [{'prediction': 'other',
'probabilities': {'computer-vision': 0.010165845975279808,
'mlops': 0.2043360322713852,
'natural-language-processing': 0.05693763121962547,
'other': 0.7285605072975159}}]}[2m[36m(ServeReplica:default:ModelDeploymentRobust pid=850829)[0m INFO 2023-09-17 22:49:40,637 ModelDeploymentRobust default#ModelDeploymentRobust#YziEno 333d113e-1023-404d-b299-f40a53693dca /predict/ default replica.py:749 - __CALL__ OK 840.9ms
In [ ]python · cell 242
python
# Shutdown
serve.shutdown()Output
2023-09-17 22:47:23,037 INFO router.py:537 -- Got updated replicas for deployment 'ModelDeploymentRobust' in application 'default': set(). [2m[36m(ServeController pid=849560)[0m INFO 2023-09-17 22:47:23,033 controller 849560 deployment_state.py:1707 - Removing 1 replica from deployment 'ModelDeploymentRobust' in application 'default'. [2m[36m(ServeReplica:default:ModelDeploymentRobust pid=849661)[0m INFO 2023-09-17 22:47:22,994 ModelDeploymentRobust default#ModelDeploymentRobust#MzdUXm 0b2038b6-fe98-4a0d-a31b-a1034a075ebf /predict/ default replica.py:749 - __CALL__ OK 3964.4ms [2m[36m(ServeController pid=849560)[0m INFO 2023-09-17 22:47:25,326 controller 849560 deployment_state.py:2027 - Replica default#ModelDeploymentRobust#MzdUXm is stopped.