139 lines
3.0 KiB
Go
139 lines
3.0 KiB
Go
package main
|
|
|
|
import (
|
|
"fmt"
|
|
"iter"
|
|
"math/rand"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const (
|
|
numExtractors int = 2
|
|
numTransformers int = numExtractors
|
|
numLoaders int = 4
|
|
chunkSize int = 20
|
|
totalRecords int = 1000
|
|
extractorsQueueSize int = 10
|
|
transformersQueueSize int = 10
|
|
recordsPerExtractor int = totalRecords / numExtractors
|
|
)
|
|
|
|
type Record struct {
|
|
Id int
|
|
Data string
|
|
}
|
|
|
|
var idCounter int = 0
|
|
|
|
func generateData(max int) iter.Seq[Record] {
|
|
time.Sleep(randomDurationMs(500, 1500))
|
|
|
|
return func(yield func(Record) bool) {
|
|
for i := range max {
|
|
record := Record{
|
|
Id: idCounter,
|
|
Data: fmt.Sprintf("Data-%d", i),
|
|
}
|
|
idCounter++
|
|
|
|
if !yield(record) {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func randomDurationMs(min int, max int) time.Duration {
|
|
if min < 0 {
|
|
panic(fmt.Sprintf(`Invalid negative value for "min" argument: (min=%d)`, min))
|
|
}
|
|
|
|
if max <= min {
|
|
panic(fmt.Sprintf(`Invalid value for "max" argument: (max=%d), "max" should be greater than "min"`, max))
|
|
}
|
|
|
|
return time.Duration(min+rand.Intn(max-min)) * time.Millisecond
|
|
}
|
|
|
|
func Extractor(id int, chunkSize int, out chan<- []Record) {
|
|
chunk := make([]Record, 0, chunkSize)
|
|
|
|
for record := range generateData(recordsPerExtractor) {
|
|
chunk = append(chunk, record)
|
|
|
|
if len(chunk) == chunkSize {
|
|
out <- chunk
|
|
chunk = make([]Record, 0, chunkSize)
|
|
fmt.Printf("[Extractor %d] Lote enviado\n", id)
|
|
}
|
|
}
|
|
|
|
if len(chunk) > 0 {
|
|
out <- chunk
|
|
fmt.Printf("[Extractor %d] Lote enviado (residuos=%d)\n", id, len(chunk))
|
|
}
|
|
}
|
|
|
|
func Transformer(id int, in <-chan []Record, out chan<- []Record) {
|
|
for records := range in {
|
|
fmt.Printf("[Transformer %d] Transformando lote de %d registros...\n", id, len(records))
|
|
time.Sleep(randomDurationMs(50, 200))
|
|
|
|
for i, record := range records {
|
|
records[i].Data = record.Data + "-transformed"
|
|
}
|
|
|
|
out <- records
|
|
}
|
|
}
|
|
|
|
func Loader(id int, in <-chan []Record) {
|
|
for chunk := range in {
|
|
fmt.Printf("[Loader %d] Procesando lote de %d registros...\n", id, len(chunk))
|
|
time.Sleep(randomDurationMs(100, 2000))
|
|
}
|
|
}
|
|
|
|
func main() {
|
|
chChunksExtract := make(chan []Record, extractorsQueueSize)
|
|
chChunksTransform := make(chan []Record, transformersQueueSize)
|
|
|
|
var wgExtractors sync.WaitGroup
|
|
for i := 1; i <= numExtractors; i++ {
|
|
wgExtractors.Go(func() {
|
|
Extractor(i, chunkSize, chChunksExtract)
|
|
})
|
|
}
|
|
|
|
go func() {
|
|
wgExtractors.Wait()
|
|
close(chChunksExtract)
|
|
fmt.Println("--- Todos los extractores terminaron. Canal cerrado (chChunksExtract). ---")
|
|
}()
|
|
|
|
var wgTransformers sync.WaitGroup
|
|
for i := 1; i <= numTransformers; i++ {
|
|
wgTransformers.Go(func() {
|
|
Transformer(i, chChunksExtract, chChunksTransform)
|
|
})
|
|
}
|
|
|
|
go func() {
|
|
wgTransformers.Wait()
|
|
close(chChunksTransform)
|
|
fmt.Println("--- Todos los transformadores terminaron. Canal cerrado (chChunksTransform). ---")
|
|
}()
|
|
|
|
var wgLoaders sync.WaitGroup
|
|
for i := 1; i <= numLoaders; i++ {
|
|
wgLoaders.Go(func() {
|
|
Loader(i, chChunksTransform)
|
|
})
|
|
}
|
|
|
|
start := time.Now()
|
|
wgLoaders.Wait()
|
|
fmt.Printf("ETL Finalizado en %v\n", time.Since(start))
|
|
}
|