How golang handles big data

Golang has proven to be very suitable for concurrent programming. Goroutine is more readable, elegant and efficient than asynchronous programming. This article proposes a Pipeline execution model suitable for implementation by Golang, which is suitable for batch processing of large amounts of data (ETL).

imagined such application scenarios: (Recommended Learning: Go )

## Load user reviews from Database A (Cassandra) (huge quantity, for example, for example 1 billion); associate user information from database B (MySQL) based on the user ID of each comment; call the NLP service (natural language processing) to process each comment; write the processing results to database C (ElasticSearch).

Due to various problems encountered in the application, these requirements are summarized:

Requirement 1: Data should be processed in batches, for example, 100 items per batch are specified. When a problem occurs (such as any database failure), it will be interrupted, and checkpoint will be used to resume from the interruption the next time the program starts.
Requirement 2: Set a reasonable number of concurrencies for each process, so that the database and NLP services have a reasonable load (without affecting other businesses, occupy as many resources as possible to improve ETL performance). For example, steps (1)-(4) set the concurrency numbers to 1, 4, 8, and 2 respectively.

This is a typical Pipeline execution model. Think of each batch of data (for example, 100 items) as a product on the assembly line. The 4 steps correspond to the 4 processing procedures on the assembly line. After each process is completed, the semi-finished product is handed over to the next process. The number of products that can be processed simultaneously in each process varies.

You may first think of enabling 1 4 8 2 goroutines and using channels to transfer data. I have done this before, and the conclusion is that doing this will make programmers crazy: the process concurrency control code is very complicated, especially when you have to deal with exceptions, execution times exceeding expectations, controllable interruptions, etc., you have to add a bunch of channels, Until you don't even remember the use.

Reusable Pipeline module

In order to complete the ETL work more efficiently, I abstracted Pipeline into modules. I'll paste the code first and then analyze the meaning. The module can be used directly, and the main interfaces used are: NewPipeline, Async, and Wait.

Using this Pipeline component, our ETL program will be simple, efficient, and reliable, freeing programmers from cumbersome concurrent process control:

package main
 
import "log"
 
func main() {
    //恢复上次执行的checkpoint，如果是第一次执行就获取一个初始值。
    checkpoint := loadCheckpoint()
    
    //工序(1)在pipeline外执行，最后一个工序是保存checkpoint
    pipeline := NewPipeline(4, 8, 2, 1) 
    for {
        //(1)
        //加载100条数据，并修改变量checkpoint
        //data是数组，每个元素是一条评论，之后的联表、NLP都直接修改data里的每条记录。
        data, err := extractReviewsFromA(&checkpoint, 100) 
        if err != nil {
            log.Print(err)
            break
        }
        
        //这里有个Golang著名的坑。
        //“checkpoint”是循环体外的变量，它在内存中只有一个实例并在循环中不断被修改，所以不能在异步中使用它。
        //这里创建一个副本curCheckpoint，储存本次循环的checkpoint。
        curCheckpoint := checkpoint
        
        ok := pipeline.Async(func() error {
            //(2)
            return joinUserFromB(data)
        }, func() error {
            //(3)
            return nlp(data)
        }, func() error {
            //(4)
            return loadDataToC(data)
        }, func() error {
            //(5)保存checkpoint
            log.Print("done:", curCheckpoint)
            return saveCheckpoint(curCheckpoint)
        })
        if !ok { break }
        
        if len(data) < 100 { break } //处理完毕
    }
    err := pipeline.Wait()
    if err != nil { log.Print(err) }
}

The above is the detailed content of How golang handles big data. For more information, please follow other related articles on the PHP Chinese website!