I have Glow running on 1 machine just fine, but when trying to simulate the glow cluster system on my local machine, via: glow master --address 0.0.0.0:8930 glow agent --dir="/Users/andrew/Desktop/GlowFolder" --port=8931 --master="0.0.0.0:8930" --memory=4096 --clean.restart --cpu.level=4

And start the app via: myapp -glow -glow.leader="0.0.0.0:8930"

1. If I don't have my executable in the Desktop/GlowFolder, i get an issue saying Failed to start command ./myapp under /Users/andrewt/Desktop/GlowFolder: fork/exec ./myapp: no such file or directory I thought the --dir flag was just for temp documents, do I need to copy the app binary to that folder as well?

2. Read size: If I run from a folder containing myapp's binary, then I can run, but the glow agent outputs the following error: 2017/03/21 09:41:24 Read size from -ct-0-ds-0-shard-4 offset 1054782852: read /Users/andrew/Desktop/GlowFolder/-ct-0-ds-0-shard-4-8931.dat: invalid argument How is read size determined and expected?

////////////////////////////Distributed Mode Conf//////////////////////////// ./glow master ./glow agent --dir data --max.executors=16 --memory=2048 --master="localhost:8930" --port 8931 ./glow agent --dir data1 --max.executors=16 --memory=2048 --master="localhost:8930" --port 8932 go run word_count.go -glow -glow.leader="localhost:8930" -glow.related.files="passwd"

///////////////////////////////word_count.go///////////////////////////////////////// `package main

import ( "flag" "fmt" "strings" "strconv" "time" "sync" "encoding/gob" _ "github.com/chrislusf/glow/driver" "github.com/chrislusf/glow/flow" )

type WordCountResult struct { Addr string Info MemInfo }

type MemInfo struct { Addr string Size int Count int }

func init() { gob.Register(MemInfo{}) }

func goStart(wg *sync.WaitGroup, fn func()) { wg.Add(1) go func() { defer wg.Done() fn() }() }

func testWordCount1() {

println("testWordCount1")
flowOut1 := make(chan WordCountResult)  
f1       := flow.New()  
f1.TextFile(
    "passwd", 2,
).Map(func(line string, ch chan MemInfo) {
    words:=strings.Split(line, ":")
    if s, err := strconv.ParseInt(words[1], 16, 0); err == nil {
        ch <- MemInfo{words[0], int(s), 1}
    }       
}).Map(func(ws MemInfo) (string, MemInfo) {
    return ws.Addr, ws
}).ReduceByKey(func(x MemInfo, y MemInfo) (MemInfo) {
    return MemInfo{x.Addr,x.Size+y.Size,x.Count+y.Count}
}).AddOutput(flowOut1)

flow.Ready()

startTime := time.Now().UnixNano()
var wg sync.WaitGroup
goStart(&wg, func() {
    f1.Run()
})

goStart(&wg, func() {
    for t := range flowOut1 {
        fmt.Printf("%s size:%-8d count:%-8d\n",
            t.Info.Addr,t.Info.Size,t.Info.Count)
    }
})  

wg.Wait()

endTime := time.Now().UnixNano()
fmt.Printf("UseTime:%d\n",(endTime - startTime) / 1000000)

}

func main() { flag.Parse() testWordCount1() }`

/////////////////////////////////console output/////////////////////// [::1]:8931>2016/02/28 10:34:06 receive error:read tcp [::1]:46225->[::1]:8931: read: connection reset by peer

/////////////////////////////////passwd/////////////////////////////////// 0x001b8aa0:00000012 0x001b8aa0:00000012 0x001b8aa0:00000020 0x001b8aa0:00000012 0x001b8aa0:00000400 0x001b8aa0:00000096 0x001b8aa0:00000064 0x001b8aa0:00000012 0x001b8aa0:00000020 0x001b8aa0:00000008 0x001b8aa0:00000012 0x001b8aa0:00000020 0x001b8aa0:00000016 0x001b8aa0:00000012 0x001b8aa0:00000020 0x001b8aa0:00000016 0x001b8aa0:00000021 0x76fb9640:00000008 0x001b8aa0:00000020 0x001b8aa0:00000020 0x001b8aa0:00000020 0x001b8aa0:00000032 0x001b8aa0:00000016 0x001b8aa0:00000012 0x001b8aa0:00000020 0x001b8ab8:00000512 0x76fb9640:00000008 0x76fb9640:00000008 0x76fb9640:00000008 0x76fb9640:00000008 0x7688b540:00000057 0x76fb9640:00000008 .......

Hi,

I'm reopening this issue as glow/flow didn't install properly this time neither. A regression might be introduced.

As glow was installed (first step go get github.com/chrislusf/glow was successful), I tried installing only flow ( go get github.com/chrislusf/glow/flow ), but this produced the same result as before.

Then, I manually uninstalled glow and reinstalled it with success. But this time, the second step yields:

> go get github.com/chrislusf/glow/flow
# github.com/chrislusf/glow/flow
C:\Go\src\github.com\chrislusf\glow\flow\signal_handling.go:21: undefined: syscall.SIGINFO
C:\Go\src\github.com\chrislusf\glow\flow\signal_handling.go:29: undefined: syscall.SIGINFO
C:\Go\src\github.com\chrislusf\glow\flow\signal_handling_windows.go:11: OnInterrupt redeclared in this block
C:\Go\src\github.com\chrislusf\glow\flow\signal_handling_windows.go:26: OnInterrupt.func1 redeclared in this block
previous declaration at C:\Go\src\github.com\chrislusf\glow\flow\signal_handling.go:26
previous declaration at C:\Go\src\github.com\chrislusf\glow\flow\signal_handling.go:11
C:\Go\src\github.com\chrislusf\glow\flow\signal_handling_windows.go:29: undefined: syscall.SIGINFO

Hope this is useful.

just use consol http://blog.scottlowe.org/2015/02/06/quick-intro-to-consul/

it incorporates allot of what you need and is written in go. i use it for micro services, and for this project its great because it wil tel you the appress of all the things you depend on too. Like the file servr or mongodb or any other data source or sink.

Minor changes in test code to make them compile and green. data_store_test:TestReadOffsets is deactivated as it is constantly failing with deadlocks. I don't think the test setup is correct. Please correct me if I'm wrong but I couldn't find anything writing to the store for the test.

This isn't a great example but I'm looking for a way to do something like this:

stream := flow.New().Source(source, parts).Map(mapper)
flow_a, flow_b := stream.Tee() // <-- this is what i really want
go flow_a.Filter(a_filter).
  Reduce(a_reduce).
  Run()
flow_a.Filter(b_filter).
  Reduce(b_reduce).
  Run()

basically, I want to be able to filter the same stream multiple ways without iterating the stream multiple times.

Right now, the Travis CI builds for 12 environments, based a combination of OS and Go versions. The config shows:

os:

• linux
• osx
• windows
• freebsd

go:

• 1.5
• 1.6
• tip

Maybe we can reduce the build combinations to only for Linux, and only build on the previous major go version + tip; that reduces the builds from 12 to 2.

The benefit is that the build time will be reduce dramatically. The down side is that we lose the testing coverage on some OSes. IMHO, this trade-off favors reduced test time; mainly because the extra OSes does not seem have much presence for Glow's intended use cases, i.e., parallel batch processing.

WDYT?

The problem is at line 73 in file master/resource/service_discovery/master/master_allocation.go. It stores pointer of the loop variable request, instead of storing pointer to the i-th element of the req.Requests.

Provides docker-compose example as quick start to use the example.

Don't merge, yet.

TODO:

• [ ] Seed mongoDB with test data

@chrislusf can you provide me the example data?

In https://github.com/chrislusf/glow/blob/master/driver/rsync/http_server.go#L91 is a random port selected which makes it very hard when access is restricted via firewalls or in a container environment where ports are exposed. The remote address which is used with the heartbeat can be a local one. Then it's impossible for any client outside to connect. I had these problems when I tried to connect to a local docker cluster from the host system.

There's no "go get" command to show how to install, but I guessed with:

go get -u github.com/chrislusf/glow

which returned no errors, but when I did a go run on the first sample in README.md, I get this:

$ go run glow1.go 
../src/github.com/chrislusf/glow/flow/dataset_sort.go:7:2: cannot find package "github.com/psilva261/timsort" in any of:
    /usr/local/go/src/github.com/psilva261/timsort (from $GOROOT)
    /home/cecil/Workspace/src/github.com/psilva261/timsort (from $GOPATH)
$ 

Looks like there may be dependencies? Could you update the README to provide install instructions?

flow.New().Source(func(chVipDataGlow chan [][]string) {
	defer close(chVipDataGlow)
	file, err := os.Open(fileName)
	fileInfo, err := os.Stat(fileName)
	if err != nil || fileInfo.Size() == 0 {
		return
	}
	defer func() {
		file.Close()
		os.Remove(fileName)
	}()
	if err != nil {
		log.Info(cc).Msgf("企业%d, 打开SFTP文件失败: %v 路径是:%s", enterpriseId, err, fileName)
		return
	}
	buf := bufio.NewReader(file)
	num := 2000
	// var tmp = make([][]string, 0, num
	var tmp [][]string
	var i = 0
	for {
		b, err := buf.ReadString('\n')
		if err != nil && err != io.EOF {
			log.Info(cc).Msgf("数据读取报错了%+v\n", err)
			break
		}
		i++
		cell := strings.Split(b, ",")
		if i == 1 && cell[0] == "数据更新时间" {
			continue
		}
		if len(cell) <= 1 {
			fileCountNumbers += 1
			continue
		}
		tmp = append(tmp, cell)
		if len(tmp) == num {
			fileCountNumbers += num
			chVipDataGlow <- tmp
			tmp = [][]string{}
		}
		if err != nil || err == io.EOF {
			log.Info(cc).Msgf("数据读完了或者报错了%+v", err)
			break
		}
	}
	if len(tmp) > 0 {
		fileCountNumbers += len(tmp)
		chVipDataGlow <- tmp
		tmp = [][]string{}
	}
}, 10).Map(func(params [][]string) [][]string {
	defer func() {
		if errs := recover(); errs != nil {
			log.Error(cc).Msgf("数据库插入执行异常了%+v", errs)
			errDefer = append(errDefer, params...)
			return
		}
	}()
	if len(params) > 0 {
		sucn, ern, _, errData := c.readDataToTableOneByOne(cc, params, enterpriseId)
		errNums += ern
		successNums += sucn
		return errData
	}
	return params
}).Map(func(errItems [][]string) {
        fmt.println(errItems)
}).Run()

I am interesting in this project. The project is surprising. It turns out that golang has a similar implementation similar to Hadoop and Flink. Has been used in the commercial production environment so far? Is there a company using supporting and developing it? As we known，Hadoop and Flink are very mature. Do I want to know that the Glow can provide the same stable function? Or this project is a toy project based your interest？

Hello.

I apologize if I have missed something obvious, but I am using glow to map and reduce time series. I would like to do a reduce or reduceByKey on every time slices (for instance, reduceByKey on all events received in the last minute.

Right now, I am setting the code up to be distributed and, following the tutorial, have put my flows in the func init() section so that they are "statically -- (instantiated only once the same way)" on every nodes.

The data is coming from an unlimited stream (i.e, not from a bounded file). So I have something like this:

func init() {
	mapRecordsToMetadata.
		Channel(mapRecordsToMetadataInput).
		Map(mapTimeSeriesToTimeSliceFunc).
		Map(mapConvertColumnValuesFunc).
	        
                // ... some more maps and filters
                
                ReduceByKey(reduceByFlowKey).
		AddOutput(mapRecordsToMetadataOutput)
}

// letsDoIt uses mapRecordsToMetadata to map and reduces all events for a given key during a time slice
func letsDoIt(streamedEvents chan []string) chan GroupedEventsByKeyChan{

  out := make (chan GroupedEventsByKeyChan)
  go func() {
      for evt := range streamedEvents {
          mapRecordsToMetadataInput <- evt
     }
  }()

  go func() {
      for evt := range mapRecordsToMetadataOutput {
          out <- evt
      }
  }()

 return out
}

I have simplified a bit, but hopefully this is enough to get the idea. Now, reduceByKey is blocking until I close mapRecordsToMetadataInput input channel (makes sense). However, if I do this, I can't really use my flow mapRecordsToMetadata anymore (is there a way to replace the input channel and restart it?).

Conceptually, I would "close" my input flow (mapRecordsToMetadataInput every "time slices" where I want the aggregate to run (i.e every 30 seconds) so that my reduceByKey would run on that intervals of inputs.

My only option seems to make the "map" operations in the init() section (i.e mapRecordsToMetadataInput) and the reduceByKey() operation in a dynamic flow, recreating the dynamic flow every 30 seconds in my case.

Something like this:

func init() {
	mapRecordsToMetadata.
		Channel(mapRecordsToMetadataInput).
		Map(mapTimeSeriesToTimeSliceFunc).
		Map(mapConvertColumnValuesFunc).
	        
                // ... some more maps and filters
                // Removed the Reduce By Key 
		AddOutput(mapRecordsToMetadataOutput)
}

func letsDoIt(streamedEvents chan []string) chan GroupedEventsByKeyChan{

  out := make (chan GroupedEventsByKeyChan)
  go func() {
      for evt := range streamedEvents {
          mapRecordsToMetadataInput <- evt
     }
  }()

  go func() {
      nextInterval := time.Now().Add(30 * time.SECOND)
      for {
         
         reduceFlow := flow.New()
         reduceInChan := make(chan EventsByKeychan)
         reduceFlow.
            Channel(reduceInChan).
            ReduceByKey(reduceByFlowKey).
            AddOutput(out)

         for evt := range mapRecordsToMetadataOutput {
            reduceInChan  <- evt

            if (evt.Time.After(nextInterval) {
                //flush and reduce for that interval
                close(ReduceInChan)
                nextInterval := nextInterval.Add(30 * time.SECOND)
            }
         }
      }
  }()

 return out
}

Is this the "right" canonical way to proceed? Does that scale? Or are we missing a small feature that would allow to "flush" our static flows at fixed intervals or on demand so that we can operate on streaming use cases in a more streamline fashion?

Frequent test timing out happened in the Travis CI. The error message usually looks like:

cogroup testing start...
panic: test timed out after 1m30s

Find the root cause and eliminate the flakiness.

The lack of unit tests greatly slows down the pace to understand, discuss, and modify the code. I think the lack of unit tests blocks my work of integrating Glow with Kubernets: I have little idea of how code suppose to work, and there is virtually no check on the correctness of changes. It also greatly increase the effort to understand the code base.

I think it makes sense to add unit tests to cover the relatively obvious/common use cases, which should be able to finish quickly and brings a lot of benefits.

I plan to go through the code base, and add test along the way. Hopefully it also accelerate my rate of ramping up with the code base.

Any thoughts, comments, or objections?