> ## Documentation Index
> Fetch the complete documentation index at: https://docs.hopx.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Data Analysis Pipeline Service

> Build automated data analysis services with multi-step workflows, data transformation chains, result export, and scheduled jobs

Build a production-ready data analysis pipeline service that processes data through multiple steps, transforms it, and exports results. This cookbook demonstrates how to create automated analysis workflows using HopX.

## Overview

Data analysis pipelines automate the process of ingesting, transforming, analyzing, and exporting data. The service executes multi-step workflows, handles data transformations, and generates reports. This pattern is used in data processing platforms and analytics services.

## Prerequisites

* HopX API key ([Get one here](https://console.hopx.dev/api-keys))
* Python 3.8+ or Node.js 16+
* Understanding of data pipelines
* Basic knowledge of data transformation

## Architecture

```
┌──────────────┐
│  Data Input  │ Raw data
└──────┬───────┘
       │
       ▼
┌─────────────────┐
│  Pipeline       │ Step 1 → Step 2 → Step 3
│   Engine        │
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  HopX Sandbox   │ Execute each step
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  Results Export │ Reports, visualizations
└─────────────────┘
```

## Implementation

### Step 1: Pipeline Definition

Define and execute multi-step pipelines:

<CodeGroup>
  ```python Python theme={null}
  from hopx_ai import Sandbox
  import os
  from typing import Dict, List, Any

  class PipelineStep:
      def __init__(self, step_id: str, code: str, dependencies: List[str] = None):
          self.step_id = step_id
          self.code = code
          self.dependencies = dependencies or []

  class DataAnalysisPipeline:
      def __init__(self, api_key: str):
          self.api_key = api_key
          self.sandbox = None
          self.steps = []
      
      def initialize(self):
          """Initialize pipeline sandbox"""
          self.sandbox = Sandbox.create(
              template="code-interpreter",
              api_key=self.api_key,
              timeout_seconds=3600
          )
      
      def add_step(self, step: PipelineStep):
          """Add pipeline step"""
          self.steps.append(step)
      
      def execute_pipeline(self, input_data: str = None) -> Dict[str, Any]:
          """Execute entire pipeline"""
          results = []
          
          # Upload input data if provided
          if input_data:
              self.sandbox.files.write("/workspace/input.csv", input_data)
          
          # Execute steps in order
          for step in self.steps:
              result = self.sandbox.run_code(step.code, timeout=60)
              results.append({
                  "step_id": step.step_id,
                  "success": result.success,
                  "stdout": result.stdout,
                  "stderr": result.stderr
              })
              
              if not result.success:
                  break  # Stop on failure
          
          # Export results
          export_result = self._export_results()
          
          return {
              "success": all(r["success"] for r in results),
              "steps_executed": len(results),
              "step_results": results,
              "export": export_result
          }
      
      def _export_results(self) -> Dict[str, Any]:
          """Export pipeline results"""
          try:
              # Check for output files
              files = self.sandbox.files.list("/workspace")
              output_files = [f for f in files if f.name.endswith(('.csv', '.json', '.png'))]
              
              exports = {}
              for file in output_files:
                  content = self.sandbox.files.read(file.path)
                  exports[file.name] = {
                      "size": len(content),
                      "type": file.name.split('.')[-1]
                  }
              
              return {
                  "success": True,
                  "files": exports
              }
          except Exception as e:
              return {
                  "success": False,
                  "error": str(e)
              }
      
      def cleanup(self):
          """Clean up pipeline resources"""
          if self.sandbox:
              self.sandbox.kill()
              self.sandbox = None

  # Usage
  pipeline = DataAnalysisPipeline(api_key=os.getenv("HOPX_API_KEY"))
  pipeline.initialize()

  # Define pipeline steps
  pipeline.add_step(PipelineStep(
      step_id="load_data",
      code="""
  import pandas as pd
  df = pd.read_csv('/workspace/input.csv')
  print(f"Loaded {len(df)} rows")
  """
  ))

  pipeline.add_step(PipelineStep(
      step_id="analyze",
      code="""
  import pandas as pd
  df = pd.read_csv('/workspace/input.csv')
  summary = df.describe()
  summary.to_csv('/workspace/summary.csv')
  print("Analysis complete")
  """
  ))

  # Execute
  result = pipeline.execute_pipeline(input_data="name,age\nAlice,25\nBob,30")
  print(result)

  pipeline.cleanup()
  ```

  ```javascript JavaScript theme={null}
  import { Sandbox } from '@hopx-ai/sdk';

  class PipelineStep {
      constructor(stepId, code, dependencies = []) {
          this.stepId = stepId;
          this.code = code;
          this.dependencies = dependencies;
      }
  }

  class DataAnalysisPipeline {
      constructor(apiKey) {
          this.apiKey = apiKey;
          this.sandbox = null;
          this.steps = [];
      }
      
      async initialize() {
          this.sandbox = await Sandbox.create({
              template: 'code-interpreter',
              apiKey: this.apiKey,
              timeoutSeconds: 3600
          });
      }
      
      addStep(step) {
          this.steps.push(step);
      }
      
      async executePipeline(inputData = null) {
          const results = [];
          
          // Upload input data if provided
          if (inputData) {
              await this.sandbox.files.write('/workspace/input.csv', inputData);
          }
          
          // Execute steps in order
          for (const step of this.steps) {
              const result = await this.sandbox.runCode(step.code, { timeout: 60 });
              results.push({
                  stepId: step.stepId,
                  success: result.success,
                  stdout: result.stdout,
                  stderr: result.stderr
              });
              
              if (!result.success) {
                  break;  // Stop on failure
              }
          }
          
          // Export results
          const exportResult = await this.exportResults();
          
          return {
              success: results.every(r => r.success),
              stepsExecuted: results.length,
              stepResults: results,
              export: exportResult
          };
      }
      
      async exportResults() {
          try {
              // Check for output files
              const files = await this.sandbox.files.list('/workspace');
              const outputFiles = files.filter(f => 
                  f.name.endsWith('.csv') || 
                  f.name.endsWith('.json') || 
                  f.name.endsWith('.png')
              );
              
              const exports = {};
              for (const file of outputFiles) {
                  const content = await this.sandbox.files.read(file.path);
                  exports[file.name] = {
                      size: content.length,
                      type: file.name.split('.').pop()
                  };
              }
              
              return {
                  success: true,
                  files: exports
              };
          } catch (error) {
              return {
                  success: false,
                  error: error.message
              };
          }
      }
      
      async cleanup() {
          if (this.sandbox) {
              await this.sandbox.kill();
              this.sandbox = null;
          }
      }
  }

  // Usage
  const pipeline = new DataAnalysisPipeline(process.env.HOPX_API_KEY);
  await pipeline.initialize();

  // Define pipeline steps
  pipeline.addStep(new PipelineStep(
      'load_data',
      `
  import pandas as pd
  df = pd.read_csv('/workspace/input.csv')
  print(f"Loaded {len(df)} rows")
  `
  ));

  pipeline.addStep(new PipelineStep(
      'analyze',
      `
  import pandas as pd
  df = pd.read_csv('/workspace/input.csv')
  summary = df.describe()
  summary.to_csv('/workspace/summary.csv')
  print("Analysis complete")
  `
  ));

  // Execute
  const result = await pipeline.executePipeline('name,age\nAlice,25\nBob,30');
  console.log(result);

  await pipeline.cleanup();
  ```
</CodeGroup>

## Best Practices

1. **Step Isolation**: Each step should be independent and testable
2. **Error Handling**: Handle errors gracefully and provide recovery options
3. **Data Validation**: Validate data between steps
4. **Result Caching**: Cache intermediate results for efficiency

## Related Cookbooks

* [Cloud Jupyter Notebook](/cookbooks/data-science/cloud-jupyter) - Notebook execution
* [ML Model Training Service](/cookbooks/data-science/ml-training-service) - Machine learning workflows

## Next Steps

1. Implement pipeline scheduling
2. Add data validation between steps
3. Create pipeline visualization
4. Implement result caching
5. Add monitoring and alerting


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.