On this tutorial, we discover how we use Daft as a high-performance, Python-native information engine to construct an end-to-end analytical pipeline. We begin by loading a real-world MNIST dataset, then progressively rework it utilizing UDFs, function engineering, aggregations, joins, and lazy execution. Additionally, we show the way to seamlessly mix structured information processing, numerical computation, and machine studying. By the tip, we’re not simply manipulating information, we’re constructing a whole model-ready pipeline powered by Daft’s scalable execution engine.
!pip -q set up daft pyarrow pandas numpy scikit-learn
import os
os.environ["DO_NOT_TRACK"] = "true"
import numpy as np
import pandas as pd
import daft
from daft import col
print("Daft version:", getattr(daft, "__version__", "unknown"))
URL = "
df = daft.read_json(URL)
print("nSchema (sampled):")
print(df.schema())
print("nPeek:")
df.present(5)We set up Daft and its supporting libraries straight in Google Colab to make sure a clear, reproducible surroundings. We configure non-obligatory settings and confirm the put in model to verify the whole lot is working accurately. By doing this, we set up a secure basis for constructing our end-to-end information pipeline.
def to_28x28(pixels):
arr = np.array(pixels, dtype=np.float32)
if arr.measurement != 784:
return None
return arr.reshape(28, 28)
df2 = (
df
.with_column(
"img_28x28",
col("image").apply(to_28x28, return_dtype=daft.DataType.python())
)
.with_column(
"pixel_mean",
col("img_28x28").apply(lambda x: float(np.imply(x)) if x shouldn't be None else None,
return_dtype=daft.DataType.float32())
)
.with_column(
"pixel_std",
col("img_28x28").apply(lambda x: float(np.std(x)) if x shouldn't be None else None,
return_dtype=daft.DataType.float32())
)
)
print("nAfter reshaping + simple features:")
df2.choose("label", "pixel_mean", "pixel_std").present(5)We load a real-world MNIST JSON dataset straight from a distant URL utilizing Daft’s native reader. We examine the schema and preview the info to know its construction and column sorts. It permits us to validate the dataset earlier than making use of transformations and have engineering.
@daft.udf(return_dtype=daft.DataType.checklist(daft.DataType.float32()), batch_size=512)
def featurize(images_28x28):
out = []
for img in images_28x28.to_pylist():
if img is None:
out.append(None)
proceed
img = np.asarray(img, dtype=np.float32)
row_sums = img.sum(axis=1) / 255.0
col_sums = img.sum(axis=0) / 255.0
whole = img.sum() + 1e-6
ys, xs = np.indices(img.form)
cy = float((ys * img).sum() / whole) / 28.0
cx = float((xs * img).sum() / whole) / 28.0
vec = np.concatenate([row_sums, col_sums, np.array([cy, cx, img.mean()/255.0, img.std()/255.0], dtype=np.float32)])
out.append(vec.astype(np.float32).tolist())
return out
df3 = df2.with_column("features", featurize(col("img_28x28")))
print("nFeature column created (list[float]):")
df3.choose("label", "features").present(2)We reshape the uncooked pixel arrays into structured 28×28 photographs utilizing a row-wise UDF. We compute statistical options, such because the imply and customary deviation, to complement the dataset. By making use of these transformations, we convert uncooked picture information into structured and model-friendly representations.
label_stats = (
df3.groupby("label")
.agg(
col("label").depend().alias("n"),
col("pixel_mean").imply().alias("mean_pixel_mean"),
col("pixel_std").imply().alias("mean_pixel_std"),
)
.type("label")
)
print("nLabel distribution + summary stats:")
label_stats.present(10)
df4 = df3.be a part of(label_stats, on="label", how="left")
print("nJoined label stats back onto each row:")
df4.choose("label", "n", "mean_pixel_mean", "mean_pixel_std").present(5)We implement a batch UDF to extract richer function vectors from the reshaped photographs. We carry out group-by aggregations and be a part of abstract statistics again to the dataset for contextual enrichment. This demonstrates how we mix scalable computation with superior analytics inside Daft.
small = df4.choose("label", "features").acquire().to_pandas()
small = small.dropna(subset=["label", "features"]).reset_index(drop=True)
X = np.vstack(small["features"].apply(np.array).values).astype(np.float32)
y = small["label"].astype(int).values
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import accuracy_score, classification_report
clf = LogisticRegression(max_iter=1000, n_jobs=None)
clf.match(X_train, y_train)
pred = clf.predict(X_test)
acc = accuracy_score(y_test, pred)
print("nBaseline accuracy (feature-engineered LogisticRegression):", spherical(acc, 4))
print("nClassification report:")
print(classification_report(y_test, pred, digits=4))
out_df = df4.choose("label", "features", "pixel_mean", "pixel_std", "n")
out_path = "/content/daft_mnist_features.parquet"
out_df.write_parquet(out_path)
print("nWrote parquet to:", out_path)
df_back = daft.read_parquet(out_path)
print("nRead-back check:")
df_back.present(3)We materialize chosen columns into pandas and prepare a baseline Logistic Regression mannequin. We consider efficiency to validate the usefulness of our engineered options. Additionally, we persist the processed dataset to Parquet format, finishing our end-to-end pipeline from uncooked information ingestion to production-ready storage.
On this tutorial, we constructed a production-style information workflow utilizing Daft, transferring from uncooked JSON ingestion to function engineering, aggregation, mannequin coaching, and Parquet persistence. We demonstrated the way to combine superior UDF logic, carry out environment friendly groupby and be a part of operations, and materialize outcomes for downstream machine studying, all inside a clear, scalable framework. By way of this course of, we noticed how Daft permits us to deal with advanced transformations whereas remaining Pythonic and environment friendly. We completed with a reusable, end-to-end pipeline that showcases how we will mix fashionable information engineering and machine studying workflows in a unified surroundings.
Try the Full Codes right here. Additionally, be at liberty to observe us on Twitter and don’t overlook to affix our 120k+ ML SubReddit and Subscribe to our E-newsletter. Wait! are you on telegram? now you’ll be able to be a part of us on telegram as properly.
Michal Sutter is an information science skilled with a Grasp of Science in Knowledge Science from the College of Padova. With a stable basis in statistical evaluation, machine studying, and information engineering, Michal excels at remodeling advanced datasets into actionable insights.
