Go를 활용한 간단 데이터 파이프라인 만들기

안녕하세요.

갈비 만두를 좋아하는 개발자 임태빈입니다.

​

Go에서 Channel을 활용한 간단한 파이프라인을 만들어보려고 합니다.

​

데이터를 생성하고 이를 Channel을 통해 전달하여 출력하고 총합을 구하는 코드를 구현해보곘습니다.

​

코드 구현에 앞서 그림을 간단하게 그려보면 다음과 같습니다.

그러면 이제 코드 구현을 해보겠습니다.

먼저 전체 코드를 공유하겠습니다.

package main

import "log"

func GenerateData() <-chan int {
	oc := make(chan int)
	go func() {
		i := 0
		for {
			oc <- i
			i = i + 1
			if i == 15 {
				break
			}
		}
		close(oc)
	}()
	return oc
}

func PrintData(ic <-chan int) <-chan int {
	oc := make(chan int)
	go func() {
		for data := range ic {
			log.Println(data)
			oc <- data
		}
		close(oc)
	}()
	return oc
}

func TotalSumOfData(ic <-chan int) {
	total := 0
	for data := range ic {
		total += data
	}
	log.Println("total: ", total)
}

func main() {
	TotalSumOfData(PrintData(GenerateData()))
}

첫번째로 볼 코드는 GenerateData입니다. 이 함수에서는 데이터를 생성합니다.

func GenerateData() <-chan int {
	oc := make(chan int)
	go func() {
		i := 0
		for {
			oc <- i
			i = i + 1
			if i == 15 {
				break
			}
		}
		close(oc)
	}()
	return oc
}

여기서 저는 sync.WaitGroup를 사용하지 않았습니다.

그 이유는 main함수 부분에서 TotalSumOfData(PrintData(GenerateData())) 이 함수가 엮이면서 달라지기 때문입니다.

먼저 Channel을 사용하면 Channel에 값을 넣어주고 바로 소모를 시켜줘야 합니다. GenerateData에서는 그런 부분이 없지만 PrintData에서 이를 Range를 통해서 진행해줍니다. 그래서 저는 sync.WaitGroup를 사용하지 않았습니다.

​

두번째 코드는 PrintData입니다. 이 함수는 데이터가 잘들어왔는지를 보기 위해 작성했습니다.

func PrintData(ic <-chan int) <-chan int {
	oc := make(chan int)
	go func() {
		for data := range ic {
			log.Println(data)
			oc <- data
		}
		close(oc)
	}()
	return oc
}

이 부분에서는 위에서 말한 Range를 통해 GenerateData에서 생긴 Channel에 값을 소모시켜줍니다. 그런 다음 다시 TotalSumOfData를 위해 다시 새로운 Channel을 만들어서 진행합니다.

​

마지막으로 봐야하는 코드인 TotalSumOfData는 Channel에 있는 값들을 모두 더하여 최종값을 만들어줍니다.

func TotalSumOfData(ic <-chan int) {
	total := 0
	for data := range ic {
		total += data
	}
	log.Println("total: ", total)
}

저는 이 파이프라인을 만들어보면서 실시간 데이터를 처리할 수 있는 간단한 코드를 만들 수 있겠다고 생각했습니다. 다음번에는 fluentd를 통해 들어온 로그 데이터들을 처리해서 만들어보면 좋겠다는 생각이 들었습니다.

​

이 포스팅을 읽어주셔서 감사합니다:) 혹시나 궁금하신 부분이나 이상한 부분이 있다면 댓글 남겨주시면 감사드리겠습니다~~

https://github.com/tae2089/golang/blob/main/datapipeline/main.go