Golang高性能分布式Map/Reduce系统插件Gleam的使用

Gleam是一个高性能、高效的分布式执行系统，它简单、通用、灵活且易于定制。Gleam是用Go语言构建的，用户定义的可以用Go、Unix管道工具或任何流式程序来编写计算逻辑。

主要特性

高性能

• 纯Go的mapper和reducer具有高性能和并发性
• 数据通过内存流动，可选择性地写入磁盘
• 多个map reduce步骤会合并在一起以获得更好的性能

内存高效

• 每个执行器运行在独立的OS进程中，由OS管理内存
• Gleam master和agent服务器内存高效，消耗约10MB内存
• 根据数据大小提示自动调整所需内存大小

灵活

• Gleam流可以独立运行或分布式运行
• 可调整为内存模式或磁盘模式

易于定制

• Go代码比Scala、Java、C++更简单易读

使用示例

单词计数示例

package main

import (
	"flag"
	"strings"

	"github.com/chrislusf/gleam/distributed"
	"github.com/chrislusf/gleam/flow"
	"github.com/chrislusf/gleam/gio"
	"github.com/chrislusf/gleam/plugins/file"
)

var (
	isDistributed   = flag.Bool("distributed", false, "run in distributed or not")
	Tokenize  = gio.RegisterMapper(tokenize)
	AppendOne = gio.RegisterMapper(appendOne)
	Sum = gio.RegisterReducer(sum)
)

func main() {

	gio.Init()   // 如果命令行调用mapper或reducer，执行并退出
	flag.Parse() // 可选，因为gio.Init()也会调用这个

	f := flow.New("top5 words in passwd").
		Read(file.Txt("/etc/passwd", 2)).  // 读取txt文件并分成2个分片
		Map("tokenize", Tokenize).    // 调用注册的"tokenize" mapper函数
		Map("appendOne", AppendOne).  // 调用注册的"appendOne" mapper函数
		ReduceByKey("sum", Sum).         // 调用注册的"sum" reducer函数
		Sort("sortBySum", flow.OrderBy(2, true)).
		Top("top5", 5, flow.OrderBy(2, false)).
		Printlnf("%s\t%d")

	if *isDistributed {
		f.Run(distributed.Option())
	} else {
		f.Run()
	}
}

func tokenize(row []interface{}) error {
	line := gio.ToString(row[0])
	for _, s := range strings.FieldsFunc(line, func(r rune) bool {
		return !('A' <= r && r <= 'Z' || 'a' <= r && r <= 'z' || '0' <= r && r <= '9')
	}) {
		gio.Emit(s)
	}
	return nil
}

func appendOne(row []interface{}) error {
	row = append(row, 1)
	gio.Emit(row...)
	return nil
}

func sum(x, y interface{}) (interface{}, error) {
	return gio.ToInt64(x) + gio.ToInt64(y), nil
}

使用Unix管道工具的单词计数

package main

import (
	"fmt"

	"github.com/chrislusf/gleam/flow"
	"github.com/chrislusf/gleam/gio"
	"github.com/chrislusf/gleam/gio/mapper"
	"github.com/chrislusf/gleam/plugins/file"
	"github.com/chrislusf/gleam/util"
)

func main() {

	gio.Init()

	flow.New("word count by unix pipes").
		Read(file.Txt("/etc/passwd", 2)).
		Map("tokenize", mapper.Tokenize).
		Pipe("lowercase", "tr 'A-Z' 'a-z'").
		Pipe("sort", "sort").
		Pipe("uniq", "uniq -c").
		OutputRow(func(row *util.Row) error {
			fmt.Printf("%s\n", gio.ToString(row.K[0]))
			return nil
		}).Run()
}

连接两个CSV文件

假设文件"a.csv"有字段"a1, a2, a3, a4, a5"，文件"b.csv"有字段"b1, b2, b3"。我们想连接a1 = b2的行，输出格式为"a1, a4, b3"。

package main

import (
	. "github.com/chrislusf/gleam/flow"
	"github.com/chrislusf/gleam/gio"
	"github.com/chrislusf/gleam/plugins/file"
)

func main() {

	gio.Init()

	f := New("join a.csv and b.csv by a1=b2")
	a := f.Read(file.Csv("a.csv", 1)).Select("select", Field(1,4)) // a1, a4
	b := f.Read(file.Csv("b.csv", 1)).Select("select", Field(2,3)) // b2, b3

	a.Join("joinByKey", b).Printlnf("%s,%s,%s").Run()  // a1, a4, b3
}

分布式计算

本地设置Gleam集群

# 在服务器上启动"gleam master"
> go get github.com/chrislusf/gleam/distributed/gleam
> gleam master --address=":45326"

# 在不同的服务器或端口上启动"gleam agent"
> gleam agent --dir=2 --port 45327 --host=127.0.0.1
> gleam agent --dir=3 --port 45328 --host=127.0.0.1

在Kubernetes上设置Gleam集群

安装Kubernetes工具后，可以使用Skaffold运行最新版本：

cd ./k8s
skaffold run --profile base

或者构建和运行本地版本：

cd ./k8s
skaffold dev --profile dev

更改执行模式

定义流后，Run()函数可以在本地模式或分布式模式下执行：

f := flow.New("")
...
// 1. 本地模式
f.Run()

// 2. 分布式模式
import "github.com/chrislusf/gleam/distributed"
f.Run(distributed.Option())
f.Run(distributed.Option().SetMaster("master_ip:45326"))

重要特性

• 容错的OnDisk()
• 从本地、HDFS或S3读取数据
• 多种数据源支持：Cassandra、Kafka、Parquet文件、ORC文件、CSV文件、TSV文件、TXT文件等

状态

Gleam仍在发展中，欢迎任何帮助！特别是现在需要帮助实现Go读取Parquet文件的功能。