Capillaries - Golang分布式批量数据处理框架插件库

Capillaries是一个用于构建分布式批量数据处理管道的Golang框架，它提供了插件化的架构，可以方便地扩展各种数据源、转换器和目的地。

核心概念

Capillaries基于以下几个核心概念构建：

1. 工作流(Workflow)：定义数据处理流程
2. 节点(Node)：工作流中的处理单元
3. 插件(Plugin)：可扩展的组件，用于实现特定功能

基本使用

安装

go get github.com/capillariesio/capillaries

简单示例

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/capillariesio/capillaries/pkg/capillaries"
	"github.com/capillariesio/capillaries/pkg/env"
	"github.com/capillariesio/capillaries/pkg/wfmodel"
)

func main() {
	// 初始化环境
	envConfig := env.Config{
		RunnerID: "local_runner",
	}
	env.Init(&envConfig)

	// 创建工作流定义
	workflowDef := wfmodel.WorkflowDef{
		Name: "demo_workflow",
		Nodes: []*wfmodel.NodeDef{
			{
				Name: "source_node",
				Type: "file_reader",
				Params: map[string]interface{}{
					"file_path": "input.csv",
					"format":    "csv",
				},
			},
			{
				Name: "transform_node",
				Type: "sql_transformer",
				Params: map[string]interface{}{
					"query": "SELECT col1, col2 FROM source_node WHERE col1 > 100",
				},
				DependsOn: []string{"source_node"},
			},
			{
				Name: "target_node",
				Type: "file_writer",
				Params: map[string]interface{}{
					"file_path": "output.csv",
					"format":    "csv",
				},
				DependsOn: []string{"transform_node"},
			},
		},
	}

	// 创建工作流运行器
	runner, err := capillaries.NewRunner(context.Background(), &workflowDef)
	if err != nil {
		log.Fatal(err)
	}

	// 执行工作流
	status, err := runner.Run()
	if err != nil {
		log.Fatal(err)
	}

	fmt.Printf("Workflow completed with status: %s\n", status)
}

主要功能

1. 数据源插件

Capillaries支持多种数据源：

// 文件读取器示例
fileReaderNode := &wfmodel.NodeDef{
	Name: "csv_reader",
	Type: "file_reader",
	Params: map[string]interface{}{
		"file_path": "data/input.csv",
		"format":    "csv",
		"has_header": true,
	},
}

// 数据库读取器示例
dbReaderNode := &wfmodel.NodeDef{
	Name: "db_reader",
	Type: "db_reader",
	Params: map[string]interface{}{
		"dsn":      "user:password@tcp(localhost:3306)/dbname",
		"query":    "SELECT * FROM table",
		"db_type":  "mysql",
	},
}

2. 数据转换插件

// SQL转换器示例
sqlTransformerNode := &wfmodel.NodeDef{
	Name: "transformer",
	Type: "sql_transformer",
	Params: map[string]interface{}{
		"query": "SELECT col1, UPPER(col2) as col2_upper FROM source_node",
	},
	DependsOn: []string{"source_node"},
}

// 自定义转换器示例
customTransformerNode := &wfmodel.NodeDef{
	Name: "custom_transformer",
	Type: "custom_processor",
	Params: map[string]interface{}{
		"processor_name": "my_processor",
	},
	DependsOn: []string{"previous_node"},
}

3. 数据目的地插件

// 文件写入器示例
fileWriterNode := &wfmodel.NodeDef{
	Name: "csv_writer",
	Type: "file_writer",
	Params: map[string]interface{}{
		"file_path": "data/output.csv",
		"format":    "csv",
		"has_header": true,
	},
	DependsOn: []string{"transformer"},
}

// 数据库写入器示例
dbWriterNode := &wfmodel.NodeDef{
	Name: "db_writer",
	Type: "db_writer",
	Params: map[string]interface{}{
		"dsn":      "user:password@tcp(localhost:3306)/dbname",
		"table":    "target_table",
		"db_type":  "mysql",
		"batch_size": 1000,
	},
	DependsOn: []string{"transformer"},
}

高级特性

1. 分布式执行

Capillaries支持分布式执行模式：

// 分布式配置
envConfig := env.Config{
	RunnerID:        "cluster_runner_1",
	CoordinatorHost: "coordinator.example.com",
	CoordinatorPort: 8080,
	WorkerCount:     4,
}
env.Init(&envConfig)

2. 自定义插件开发

创建自定义处理器：

package myprocessor

import (
	"context"

	"github.com/capillariesio/capillaries/pkg/l"
	"github.com/capillariesio/capillaries/pkg/processor"
	"github.com/capillariesio/capillaries/pkg/sc"
)

type MyProcessor struct {
	processor.DefaultProcessor
}

func (p *MyProcessor) Run(
	ctx context.Context,
	script *sc.ScriptContext,
	logger *l.Logger,
	params map[string]interface{},
	recordsBatch *sc.ProcessorRecordsBatch) error {
	
	// 处理逻辑
	for i := range recordsBatch.Records {
		// 对每条记录进行处理
		recordsBatch.Records[i]["processed"] = true
	}
	
	return nil
}

func init() {
	processor.RegisterProcessor("my_processor", func() processor.Processor {
		return &MyProcessor{}
	})
}

3. 监控和日志

// 配置日志
logger := l.NewLogger(l.VerboseLevel)
logger.Info("Starting workflow execution")

// 获取工作流状态
status, err := runner.GetStatus()
if err != nil {
	logger.Error("Failed to get workflow status: %s", err.Error())
}
logger.Info("Workflow status: %s", status)

最佳实践

1. 合理设计工作流：将复杂处理分解为多个节点
2. 使用批处理：对于大数据量，设置适当的批处理大小
3. 错误处理：实现健壮的错误处理和重试机制
4. 资源管理：监控资源使用情况，避免OOM
5. 测试：为自定义插件编写单元测试

Capillaries提供了灵活且强大的框架来构建分布式数据处理管道，通过其插件化架构，可以轻松扩展以满足各种数据处理需求。