Etl Laravel Package

Getting Started

Minimal Steps

1. Installation:

   composer require flow-php/etl
   

   Publish the config file:

   php artisan vendor:publish --provider="Flow\ETL\ETLServiceProvider" --tag="config"
   

2. First Use Case: Define a simple ETL pipeline to extract data from a CSV, transform it, and load it into a database table.

   use Flow\ETL\ETL;
   use Flow\ETL\Extractor\CsvExtractor;
   use Flow\ETL\Loader\DatabaseLoader;
   use Flow\ETL\Transformer\ArrayTransformer;
   
   $etl = app(ETL::class);
   $etl->run(
       new CsvExtractor('path/to/data.csv'),
       new ArrayTransformer(fn($row) => ['name' => $row['first_name'], 'email' => $row['email']]),
       new DatabaseLoader('users', connection: 'mysql')
   );
   

3. Where to Look First:

   • Documentation for core concepts and API reference.
   • config/etl.php for adapter configurations.
   • vendor/flow-php/etl/src/ for source code examples (extractors, loaders, transformers).

Implementation Patterns

Usage Patterns

1. Pipeline Composition: Chain extractors, transformers, and loaders in a fluent manner.

   $etl->run(
       new ApiExtractor('https://api.example.com/users'),
       new JsonPathTransformer('$.data[*].{name,email}'),
       new DatabaseLoader('users')
   );
   

2. Generator-Based Processing: Leverage generators for memory efficiency with large datasets.

   $etl->run(
       new DatabaseExtractor('SELECT * FROM large_table'),
       new ChunkTransformer(1000), // Process in chunks
       new S3Loader('bucket-name')
   );
   

3. Adapter Customization: Extend built-in adapters or create custom ones.

   class CustomDatabaseExtractor implements ExtractorInterface {
       public function extract(): Generator {
           yield ['id' => 1, 'name' => 'John'];
           yield ['id' => 2, 'name' => 'Jane'];
       }
   }
   

4. Integration with Laravel Queues: Dispatch ETL jobs asynchronously.

   use Flow\ETL\ETLJob;
   
   ETLJob::dispatch(
       new CsvExtractor('data.csv'),
       new ArrayTransformer(...),
       new DatabaseLoader('users')
   )->onQueue('etl');
   

5. Error Handling and Retries: Use Laravel’s queue retries or custom middleware.

   $etl->run(
       new ApiExtractor('https://api.example.com/users'),
       new RetryTransformer(3, fn($row) => $row['status'] === 'failed')
   );
   

Workflows

1. Data Migration:

   • Extract: Pull data from a legacy system (e.g., LegacyDatabaseExtractor).
   • Transform: Clean and normalize data (e.g., ArrayTransformer).
   • Load: Insert into a new database (e.g., DatabaseLoader).

2. Real-Time Analytics:

   • Extract: Fetch data from a webhook (e.g., WebhookExtractor).
   • Transform: Aggregate or enrich data (e.g., AggregateTransformer).
   • Load: Store in a time-series database (e.g., InfluxDBLoader).

3. Automated Reporting:

   • Extract: Pull data from multiple APIs (e.g., MultiApiExtractor).
   • Transform: Apply business logic (e.g., ReportTransformer).
   • Load: Export to CSV or PDF (e.g., CsvLoader, PdfLoader).

Integration Tips

1. Laravel Service Container: Bind adapters as Laravel services for dependency injection.

   $this->app->bind(
       ExtractorInterface::class,
       fn($app) => new ApiExtractor('https://api.example.com')
   );
   

2. Configuration: Use Laravel’s config system to manage adapter settings.

   // config/etl.php
   'adapters' => [
       'api' => [
           'base_url' => env('API_BASE_URL'),
           'timeout' => 30,
       ],
   ],
   

3. Artisan Commands: Create CLI commands for running ETL pipelines.

   class RunDataSync extends Command {
       public function handle() {
           $etl = app(ETL::class);
           $etl->run(new DataSyncPipeline());
       }
   }
   

4. Event Listeners: Emit events for job lifecycle management.

   // Listen for ETL job started/failed events
   event(new EtlJobStarted($job));
   

5. Testing: Use Laravel’s testing tools to mock adapters and pipelines.

   $this->mock(ExtractorInterface::class, function ($mock) {
       $mock->shouldReceive('extract')->andReturn([['name' => 'Test']]);
   });
   

Gotchas and Tips

Pitfalls

1. Memory Leaks:

   • Issue: Generators are memory-efficient, but improperly closed resources (e.g., database connections, file handles) can cause leaks.
   • Fix: Ensure all adapters properly close resources. Use Laravel’s Connection facade for database operations to manage connections.

2. Type Safety:

   • Issue: Transformers may not enforce strict typing, leading to runtime errors.
   • Fix: Use PHP 8.0+ typed properties and assert statements.

     $transformer = new ArrayTransformer(fn(array $row): array => [
         'name' => $row['first_name'] ?? '',
         'email' => assert($row['email'] ?? null, 'string'),
     ]);
     

3. Queue Blocking:

   • Issue: Long-running ETL jobs can block queue workers.
   • Fix: Process data in chunks or smaller batches. Use Laravel’s chunk() method for database operations.

4. Adapter Dependencies:

   • Issue: Some adapters (e.g., S3, AWS SDK) introduce heavy dependencies.
   • Fix: Audit composer.json for unused dependencies. Use Laravel’s config/etl.php to disable unused adapters.

5. Debugging Generators:

   • Issue: Generator-based flows can be hard to debug with traditional tools.
   • Fix: Use tap() to inspect intermediate data.

     $etl->run(
         new CsvExtractor('data.csv'),
         new ArrayTransformer(fn($row) => tap($row, fn($r) => Log::debug('Row:', $r))),
         new DatabaseLoader('users')
     );
     

6. Idempotency:

   • Issue: Retrying failed jobs may cause duplicate data.
   • Fix: Implement idempotent loaders (e.g., UpsertDatabaseLoader).

7. Performance Bottlenecks:

   • Issue: Slow extractors or loaders can bottleneck the pipeline.
   • Fix: Profile with Laravel Telescope or Blackfire. Optimize queries or use batch processing.

Tips

1. Leverage Laravel Facades: Use Laravel’s facades (e.g., DB, Cache, Log) within adapters for consistency.

   class DatabaseExtractor implements ExtractorInterface {
       public function extract(): Generator {
           foreach (DB::table('users')->cursor() as $user) {
               yield $user;
           }
       }
   }
   

2. Custom Middleware: Add middleware for cross-cutting concerns (e.g., logging, validation).

   $etl->run(
       new CsvExtractor('data.csv'),
       new LoggingMiddleware(),
       new ValidationTransformer(fn($row) => validator($row, ['email' => 'required|email'])),
       new DatabaseLoader('users')
   );
   

3. Environment-Specific Configs: Use Laravel’s .env for environment-specific settings.

   // .env
   ETL_API_TIMEOUT=60
   ETL_S3_BUCKET=my-bucket
   

4. Monitoring: Integrate with Laravel Horizon or custom dashboards to track job progress.

   // Emit progress events
   event(new EtlProgressEvent(50, 'Processing users...'));
   

5. Testing Strategies:

   • Unit Tests: Mock adapters and test transformers in isolation.
   • Integration Tests: Test full pipelines with a test database.
   • Feature Tests: Simulate real-world usage with HTTP requests or commands.

6. Extending Adapters: Create reusable adapter wrappers for common use cases.

   class EloquentLoader implements LoaderInterface {
       public function load(iterable $data): void {
           foreach ($data as $item) {
               User::create($item);
           }
       }
   }
   

7. Documentation: Document complex pipelines with PHPDoc or