gostream

README

gostream

go Map Reduce Sorted Filter Limit stream , a simple library
本库可用于科学计算，数据分析等方向，欢迎贡献或提需求

进度 已实现

• Limit 支持流截断
• Combine 支持流之间的拼接
• Filter 支持流过滤
• Sorted 支持流排序
• Reduce 支持流 Reduce
• Map 支持流 Map
• MultipartMap 支持并发流
• Copy 流拷贝
• GroupByToStream 流分流
• GroupByToMap 流分组
• Distinct 排除重复元素 拟定适用技术 bloom 过滤器 / Cuckoo Filter 过滤器
• 流是否可以用反射形成新流从而替换之前的流，然后使用新流进行下一步的计算(是可以的，不过每个流都需要是[]interface的，如果不是，替换时会发生类型不一致，但为了一般性和适用性，采用将期望转换类型传入的方式，返回转换后的期望类型给用户,已实现，采用传入目标数组的形式来获取期望的数组类型)

进度 待实现

• Manager 流管理器 需要构思需求 暂时只需要存储流即可
• 流之间拼接允许按照优先级进行排列（和优先级队列相似，用于优先级重要的场景）

Documentation here: https://godoc.org/github.com/felix0080/gostream

Usage

package main
import (
   "fmt"
   "strconv"
   . "github.com/felix0080/gostream"
)
type Student struct {
   Name string
}
type IntSlice []interface{}

func (s IntSlice) Len() int { return len(s) }
func (s IntSlice) Swap(i, j int){ s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
   return s[i].(int) < s[j].(int)
} 
type Item int
func (s Item) HashCode() []byte {
   return []byte(strconv.Itoa(int(s)))
}
func main() {
  //调用BuildStream 方法，传入实现了sort 接口的slice，若不需要排序，则无需实现sort
   //定义一个slice.填充元素
   a:=[]interface{}{4,3,2,1}
   //使用实现了sort的type 包装，并调用调用BuildStream
   s:=BuildStream(IntSlice(a))
   //定义一个slice.填充元素
   a1:=[]interface{}{6,5,8,7}
   //使用实现了sort的type 包装,并调用调用BuildStream
   s1:=BuildStream(IntSlice(a1))
  //指定流的类型
   var types []int
   //将两个流合并为1个，并且以s为主流
   value:=s.Combine(s1).
   //排序，对流的排序，若没有实现sort接口，将不会进行排序
       Sorted().
   //只保留前6个元素
       Limit(6).
   //map 将函数作用到每一个流元素上
       Map(types,func(i interface{}) interface{} {
           return i.(int)+1
       }).
   //Filter 过滤函数返回值为true的元素，形成新流
       Filter(func(i interface{}) bool {
           return i.(int)==7
       }).
   //Reduce 将前两个元素计算后的结果和第三个元素计算，以此类推
       Reduce(func(i interface{}, i2 interface{}) interface{} {
           return i.(int)+i2.(int)
       })//==20
   //s.Limit(10)
   fmt.Println("after arr: ", s.Collect())
   fmt.Println("after value: ", value)
  
   s2:=s.Copy()
   s2.Map(types,func(i interface{}) interface{} {
       return i.(int)+1
   }).Sorted()
   fmt.Println(s2.Collect())
   fmt.Println(s.Collect())
   a3 := []interface{}{Item(1), Item(2), Item(4), Item(4)}
   s3 := BuildStream(IntSlice(a3))
   s3.Distinct()
   fmt.Println(s3.Collect())
   
//流分流
   var stus []Student
   for i := 0; i < 5; i++ {
   	stus=append(stus,Student{strconv.Itoa(i)})
   }
   stus=append(stus, Student{strconv.Itoa(1)})
   ss:=BuildStream(stus)
   sss:=ss.GroupByToStream("Name")
   fmt.Println(sss)
   for index,value:=range sss{
   	fmt.Println(index,value.Collect())
   }
}
  

Documentation

Overview

Package gostream implements a stream slice.

Usage

调用BuildStream 方法，传入实现了sort 接口的slice，若不需要排序，则无需实现sort

//定义一个slice.填充元素
a:=[]interface{}{4,3,2,1}
//使用实现了sort的type 包装，并调用调用BuildStream
s:=BuildStream(IntSlice(a))
//定义一个slice.填充元素
a1:=[]interface{}{6,5,8,7}
//使用实现了sort的type 包装,并调用调用BuildStream
s1:=BuildStream(IntSlice(a1))
//将两个流合并为1个，并且以s为主流
value:=s.Combine(s1).
//排序，对流的排序，若没有实现sort接口，将不会进行排序
	Sorted().
//只保留前6个元素
	Limit(6).
//map 将函数作用到每一个流元素上
	Map(func(i interface{}) interface{} {
		return i.(int)+1
	}).
//Filter 过滤函数返回值为true的元素，形成新流
	Filter(func(i interface{}) bool {
		return i.(int)==7
	}).
//Reduce 将前两个元素计算后的结果和第三个元素计算，以此类推
Reduce(func(i interface{}, i2 interface{}) interface{} {
		return i.(int)+i2.(int)
	})//==20
//s.Limit(10)
fmt.Println("after arr: ", s.Collect())
fmt.Println("after value: ", value)

可使用copy,将流深度拷贝成新的Stream

a:=[]interface{}{4,3,2,1}
s:=BuildStream(IntSlice(a))
s1:=s.Copy()
s1.Map(func(i interface{}) interface{} {
	return i.(int)+1
}).Sorted()
fmt.Println(s1.Collect())
fmt.Println(s.Collect())

去除重复元素样例

func (s Item) HashCode() []byte {
	return []byte(strconv.Itoa(int(s)))
}
func (s IntSlice) Len() int      { return len(s) }
func (s IntSlice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
func TestStream_Distinct(t *testing.T) {
	a := []interface{}{Item(1), Item(2), Item(4), Item(4)}
	s := BuildStream(IntSlice(a))
	s.Distinct()
	fmt.Println(s.Collect())
}

Index

• type Stream
  • func BuildStream(array interface{}) *Stream
  • func (stream *Stream) Collect() interface{}
  • func (stream *Stream) Combine(anotherStream *Stream) *Stream
  • func (stream *Stream) Copy() *Stream
  • func (stream *Stream) Distinct() *Stream
  • func (stream *Stream) Filter(f func(interface{}) bool) *Stream
  • func (stream *Stream) GroupByToMap(FieldName string) map[string][]interface{}
  • func (stream *Stream) GroupByToStream(FieldName string) map[string]*Stream
  • func (stream *Stream) Limit(len int) *Stream
  • func (stream *Stream) Map(types interface{}, f func(interface{}) interface{}) *Stream
  • func (stream *Stream) MultipartMap(worknum int, types interface{}, f func(interface{}) interface{}) *Stream
  • func (stream *Stream) Reduce(f func(interface{}, interface{}) interface{}) interface{}
  • func (stream *Stream) Sorted() *Stream
• type StreamManager
• type Unique

Types

type Stream

type Stream struct {
	// contains filtered or unexported fields
}

定义一个流，内部为 实现了Sort 接口的 Slice(interfalce{}). 若没有实现Sort接口，sort方法将不可使用 使用方法类似于java8 stream 欢迎贡献，大家一起完善此库

func BuildStream

func BuildStream(array interface{}) *Stream

Example:

type IntSlice []interface{}

func (s IntSlice) Len() int { return len(s) }
func (s IntSlice) Swap(i, j int){ s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
a:=[]interface{}{4,3,2,1}
s:=BuildStream(IntSlice(a))

func (*Stream) Collect

func (stream *Stream) Collect() interface{}

收集流，将流的内部数组返回。

func (*Stream) Combine

func (stream *Stream) Combine(anotherStream *Stream) *Stream

组合流，将两个stream 简单组合在一起

func (*Stream) Copy

func (stream *Stream) Copy() *Stream

流拷贝，拷贝一个新流，返回与以前的流相同大小，容量的新流，且将内容拷贝进去

s2:=s.Copy()
s2.Map(func(i interface{}) interface{} {
	return i.(int)+1
}).Sorted()
fmt.Println(s2.Collect())
fmt.Println(s.Collect())

func (*Stream) Distinct

func (stream *Stream) Distinct() *Stream

此方法将需要流中的元素实现 unique接口 hashCode（） 算法采用布谷鸟算法

Example
func (s Item) HashCode() []byte {
	return []byte(strconv.Itoa(int(s)))
}
func (s IntSlice) Len() int      { return len(s) }
func (s IntSlice) Swap(i, j int) { s[i], s[j] = s[j], s[i] }
func (s IntSlice) Less(i, j int) bool {
	return s[i].(int) < s[j].(int)
}
func TestStream_Distinct(t *testing.T) {
	a := []interface{}{Item(1), Item(2), Item(4), Item(4)}
	s := BuildStream(IntSlice(a))
	s.Distinct()
	fmt.Println(s.Collect())
}

func (*Stream) Filter

func (stream *Stream) Filter(f func(interface{}) bool) *Stream

设置条件满足的被过滤

func (*Stream) GroupByToMap

func (stream *Stream) GroupByToMap(FieldName string) map[string][]interface{}

GroupByToMap 不同于java stream 只求每种类别的数量， GroupByToMap 主要是分解出多个流， 返回一个流map结构 key为FieldName 即结构体内的元素

func (*Stream) GroupByToStream

func (stream *Stream) GroupByToStream(FieldName string) map[string]*Stream

//设置条件满足的被过滤

func (stream *Stream) ConvertToMap(f func(interface{}) (string,interface{})) *Stream {
	v := reflect.ValueOf(stream.array)
	len := v.Len()
	for i := 0; i < len; i++ {
		key,value:=f(v.Index(i).Interface())
	}
	v = v.Slice(0, len)
	stream.array = v.Interface()
	return stream
}

GroupByToStream 不同于java stream 只求每种类别的数量， GroupByToStream 主要是分解出多个流， 返回一个流map结构 key为FieldName 即结构体内的元素

func (*Stream) Limit

func (stream *Stream) Limit(len int) *Stream

用户获取指定数量的流

func (*Stream) Map

func (stream *Stream) Map(types interface{}, f func(interface{}) interface{}) *Stream

用于映射每个元素到对应的结果 返回的数组的类型需要指定，传入第一个参数reflect.Type

func (*Stream) MultipartMap

func (stream *Stream) MultipartMap(worknum int, types interface{}, f func(interface{}) interface{}) *Stream

func (stream *Stream) Copy()*Stream {
	a1:=[]interface{}{6,5,"asd",7}
	a2:=make([]interface{},len(a1))
	fmt.Println(fmt.Sprintf("%p %p",a1,a2))
	copy(a2,a1)
	fmt.Println(a2)
}

用于映射每个元素到对应的结果 返回的数组的类型需要指定，传入第一个参数reflect.Type

func (*Stream) Reduce

func (stream *Stream) Reduce(f func(interface{}, interface{}) interface{}) interface{}

当数组==0时，返回Nil 当数组==1时，返回Nil

func (*Stream) Sorted

func (stream *Stream) Sorted() *Stream

用户对流进行排序 需要自行实现sort接口才可使用此方法

type StreamManager

type StreamManager struct {
}

type Unique

type Unique interface {
	HashCode() []byte
}