helixml / helix

HelixML Services for Data Processing

Scenario: A developer, let’s call them Alex, is working on a machine learning project and needs to preprocess large datasets before training or inference. Alex has decided to use HelixML services for data processing due to its capabilities in data ingestion, transformation, and cleaning. In this example, we will walk through the steps Alex takes to prepare their dataset using HelixML.

  1. Data Ingestion:

First, Alex will use HelixML’s data ingestion service to load their dataset into the system. HelixML supports various data sources, including local files, databases, and cloud storage. For this example, let’s assume Alex has a CSV file named “data.csv” that they want to load.

To load the data, Alex will use the HelixML API. Here’s a code snippet in Go to load the data:

import (
"context"
"github.com/helixml/helix/api/pkg/data"
"github.com/helixml/helix/api/pkg/server"
)

func main() {
// Initialize the HelixML client
config := server.NewConfig()
client, err := server.NewClient(context.Background(), config)
if err != nil {
log.Fatal(err)
}

// Create a new dataset
dataset, err := data.NewDataset(context.Background(), client, "my_dataset")
if err != nil {
log.Fatal(err)
}

// Load the CSV file into the dataset
file, err := os.Open("data.csv")
if err != nil {
log.Fatal(err)
}
defer file.Close()

_, err = dataset.LoadCSV(context.Background(), file)
if err != nil {
log.Fatal(err)
}

// Print the dataset metadata
fmt.Println(dataset.Metadata)
}
  1. Data Transformation:

Once the data is loaded, Alex may need to transform it to prepare it for machine learning. HelixML provides various data transformation capabilities, including filtering, mapping, and aggregating data.

For this example, let’s assume Alex wants to filter the data to include only records where the age is above 30. Here’s a code snippet to apply the filter:

// Fetch the dataset
dataset, err := client.Data().GetDataset(context.Background(), "my_dataset")
if err != nil {
log.Fatal(err)
}

// Create a new transformation pipeline
pipeline := data.NewPipeline(context.Background(), client)

// Add a filter step to the pipeline
pipeline.AddStep(data.NewFilterStep("age > 30"))

// Apply the pipeline to the dataset
transformedDataset, err := pipeline.Transform(context.Background(), dataset)
if err != nil {
log.Fatal(err)
}

// Print the transformed dataset metadata
fmt.Println(transformedDataset.Metadata)
  1. Data Cleaning:

Finally, Alex may need to clean the data to remove any inconsistencies or errors. HelixML provides various data cleaning capabilities, including data normalization, data imputation, and outlier detection.

For this example, let’s assume Alex wants to normalize the “salary” column to have a mean of 50 and a standard deviation of 10. Here’s a code snippet to apply the normalization:

// Fetch the transformed dataset
transformedDataset, err := client.Data().GetDataset(context.Background(), "transformed_dataset")
if err != nil {
log.Fatal(err)
}

// Create a new transformation pipeline
pipeline := data.NewPipeline(context.Background(), client)

// Add a normalize step to the pipeline
pipeline.AddStep(data.NewNormalizeStep("salary", 50, 10))

// Apply the pipeline to the dataset
cleanedDataset, err := pipeline.Transform(context.Background(), transformedDataset)
if err != nil {
log.Fatal(err)
}

// Print the cleaned dataset metadata
fmt.Println(cleanedDataset.Metadata)

Tests:

To verify the answer, Alex can write tests to check the data loading, transformation, and cleaning steps. Here’s an example test function:

func TestDataProcessing(t *testing.T) {
// Initialize the HelixML client
config := server.NewConfig()
client, err := server.NewClient(context.Background(), config)
if err != nil {
t.Fatal(err)
}

// Create a new dataset
dataset, err := data.NewDataset(context.Background(), client, "my_dataset")
if err != nil {
t.Fatal(err)
}

// Load the CSV file into the dataset
file, err := os.Open("data.csv")
if err != nil {
t.Fatal(err)
}
defer file.Close()

_, err = dataset.LoadCSV(context.Background(), file)
if err != nil {
t.Fatal(err)
}

// Check the dataset metadata
expectedMetadata := map[string]interface{}{
"name":        "my_dataset",
"description": "Example dataset",
"size":        1024,
"format":      "csv",
"created_at":   time.Now(),
"updated_at":   time.Now(),
}
if !reflect.DeepEqual(dataset.Metadata, expectedMetadata) {
t.Errorf("Unexpected dataset metadata: %v", dataset.Metadata)
}

// Filter the dataset
pipeline := data.NewPipeline(context.Background(), client)
pipeline.AddStep(data.NewFilterStep("age > 30"))

transformedDataset, err := pipeline.Transform(context.Background(), dataset)
if err != nil {
t.Fatal(err)
}

// Check the transformed dataset metadata
expectedTransformedMetadata := map[string]interface{}{
"name":        "transformed_dataset",
"description": "Filtered dataset",
"size":        512,
"format":      "csv",
"created_at":   time.Now(),
"updated_at":   time.Now(),
}
if !reflect.DeepEqual(transformedDataset.Metadata, expectedTransformedMetadata) {
t.Errorf("Unexpected transformed dataset metadata: %v", transformedDataset.Metadata)
}

// Normalize the transformed dataset
pipeline = data.NewPipeline(context.Background(), client)
pipeline.AddStep(data.NewNormalizeStep("salary", 50, 10))

cleanedDataset, err := pipeline.Transform(context.Background(), transformedDataset)
if err != nil {
t.Fatal(err)
}

// Check the cleaned dataset metadata
expectedCleanedMetadata := map[string]interface{}{
"name":        "cleaned_dataset",
"description": "Cleaned dataset",
"size":        512,
"format":      "csv",
"created_at":   time.Now(),
"updated_at":   time.Now(),
}
if !reflect.DeepEqual(cleanedDataset.Metadata, expectedCleanedMetadata) {
t.Errorf("Unexpected cleaned dataset metadata: %v", cleanedDataset.Metadata)
}
}

This test function checks the dataset metadata after each processing step to ensure the data is loaded, transformed, and cleaned correctly.