Руководство по программированию для создания масштабируемого конвейера обработки данных машинного обучения с помощью Daft

Руководство по программированию для создания масштабируемого конвейера обработки данных машинного обучения с помощью Daft

В этом руководстве мы рассмотрим, как использовать Daft в качестве высокопроизводительного механизма обработки данных на языке Python для создания комплексного аналитического конвейера. Мы начнём с загрузки реального набора данных MNIST, затем будем последовательно преобразовывать его с помощью UDF, разработки функций, агрегации, объединений и отложенного выполнения. Также мы покажем, как можно беспрепятственно сочетать обработку структурированных данных, численные вычисления и машинное обучение.

Установка библиотек

Мы устанавливаем Daft и поддерживающие его библиотеки непосредственно в Google Colab, чтобы обеспечить чистую и воспроизводимую среду.

```python

!pip -q install daft pyarrow pandas numpy scikit-learn

```

```python

import os

os.environ["DONOTTRACK"] = "true"

```

```python

import numpy as np

import pandas as pd

import daft

from daft import col

```

```python

print("Daft version:", getattr(daft, "version", "unknown"))

```

Загрузка данных

```python

URL = "https://github.com/Eventual-Inc/mnist-json/raw/master/mnisthandwrittentest.json.gz"

df = daft.read_json(URL)

print("\nSchema (sampled):")

print(df.schema())

print("\nPeek:")

df.show(5)

```

Преобразование данных

```python

def to_28x28(pixels):

arr = np.array(pixels, dtype=np.float32)

if arr.size != 784:

return None

return arr.reshape(28, 28)

df2 = (

df

.with_column(

"img_28x28",

col("image").apply(to28x28, returndtype=daft.DataType.python())

)

.with_column(

"pixel_mean",

col("img_28x28").apply(lambda x: float(np.mean(x)) if x is not None else None,

return_dtype=daft.DataType.float32())

)

.with_column(

"pixel_std",

col("img_28x28").apply(lambda x: float(np.std(x)) if x is not None else None,

return_dtype=daft.DataType.float32())

)

)

print("\nAfter reshaping + simple features:")

df2.select("label", "pixelmean", "pixelstd").show(5)

```

Создание столбца признаков

```python

@daft.udf(returndtype=daft.DataType.list(daft.DataType.float32()), batchsize=512)

def featurize(images_28x28):

out = []

for img in images28x28.topylist():

if img is None:

out.append(None)

continue

img = np.asarray(img, dtype=np.float32)

row_sums = img.sum(axis=1) / 255.0

col_sums = img.sum(axis=0) / 255.0

total = img.sum() + 1e-6

ys, xs = np.indices(img.shape)

cy = float((ys * img).sum() / total) / 28.0

cx = float((xs * img).sum() / total) / 28.0

vec = np.concatenate([rowsums, colsums, 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.withcolumn("features", featurize(col("img28x28")))

print("\nFeature column created (list[float]):")

df3.select("label", "features").show(2)

```

Группировка и объединение данных

```python

label_stats = (

df3.groupby("label")

.agg(

col...

Читать далее