csv,grok,json,ngnix parser添加多线程 #709

redHJ · 2018-08-15T07:49:21Z

unknwon · 2018-08-16T02:41:08Z

logkit.go

@@ -239,8 +239,9 @@ func main() {
 		times.AddLayout(conf.TimeLayouts)
 	}
 	if conf.MaxProcs == 0 {
-		conf.MaxProcs = runtime.NumCPU()
+		conf.MaxProcs = NumCpu


s/NumCpu/NumCPU/

unknwon · 2018-08-16T02:43:05Z

parser/csv/csv.go

@@ -39,6 +41,7 @@ type Parser struct {
 	allmoreStartNUmber   int
 	allowNotMatch        bool
 	ignoreInvalid        bool
+	routineNumber        int


routineNumber 听起来像某个任务号，建议改叫 numRoutines，其它地方也是

unknwon · 2018-08-16T03:06:38Z

parser/json/json.go

@@ -9,6 +9,10 @@ import (

 	"github.com/qiniu/log"

+	"sync"
+
+	"sort"


分组搞一下

wonderflow · 2018-08-20T05:13:35Z

utils/models/models.go

@@ -60,6 +61,11 @@ const (
 	InputNumber = "inputNumber"
 )

+var (
+	MaxProcs = 0


这里初始化设置为1

wonderflow · 2018-08-20T05:16:28Z

utils/utils.go

@@ -10,3 +12,15 @@ func IsExist(path string) bool {
 	_, err := os.Stat(path)
 	return err == nil || os.IsExist(err)
 }
+
+func GetTestData(line string) []string {


这个方法本身的含义是复制数据到默认的maxBatchSize，应该以函数本身的含义命名函数。然后这个函数主要用于测试，这一点可以写在注释上。

wonderflow · 2018-08-20T05:20:43Z

parser/csv/csv.go

@@ -95,6 +98,10 @@ func NewParser(c conf.MapConf) (parser.Parser, error) {
 	}
 	allmoreStartNumber, _ := c.GetIntOr(parser.KeyCSVAllowMoreStartNum, 0)
 	ignoreInvalid, _ := c.GetBoolOr(parser.KeyCSVIgnoreInvalidField, false)
+	numRoutine := MaxProcs
+	if numRoutine == 0 {


理论上MaxProcs不可能是0，如果是0，那么代表哪里出bug了，那我建议这里的numRoutine也不用跑飞，只要设置为1保持稳妥就好了

wonderflow · 2018-08-20T05:23:52Z

parser/csv/csv.go

@@ -451,6 +463,15 @@ func (p *Parser) Rename(datas []Data) []Data {
 	return newData
 }

+func (p *Parser) RenameData(data Data) Data {


上面的Rename应该调用你这个RenameData方法。另外这个方法不用挂到(p *Parser)下面，还能方便一下测试

wonderflow · 2018-08-20T05:28:46Z

parser/json/json.go

+	return dataSlice, nil
+}
+
+func (p *Parser) parseLine(sendChan chan parser.ParseInfo, resultChan chan parser.ParseResult, wg *sync.WaitGroup) {


你这个sendChan 应该叫 dataPipeline，那一头是send，这一头实际上是receive，另外channel类型也可以设置为receive only ： <-chan

另外这里为什么要单独写一个，我看parser包里有一个

这个和别的parse不一样，这边返回的是数组，别的不是，json会返回数组

unknwon · 2018-08-21T03:12:55Z

LGTM

wonderflow · 2018-08-21T04:56:40Z

parser/json/json.go

-		// label 不覆盖数据，其他parser不需要这么一步检验，因为Schema固定，json的Schema不固定
-		if _, ok := data[l.Name]; ok {
-			continue
+func (p *Parser) parse(line string) (dataSlice []Data, err error) {


为什么要返回数组

json可能是数组啊

redHJ · 2018-08-27T08:15:27Z

Unbuntu Parse Test

GROK, data: 1GB ParseTime(ns)
Routine: 1, ParseTime: 1226077872156
Routine: 2, ParseTime: 713335285905
Routine: 3, ParseTime: 503750868591
Routine: 4, ParseTime: 395376283339

CSV, data: 1GB ParseTime(ns)
Routine: 1, ParseTime: 262033208580
Routine: 2, ParseTime: 183709294532
Routine: 3, ParseTime: 141773371734
Routine: 4, ParseTime: 121622516601

NGNIX data: 1GB ParseTime(ns)
Routine: 1, ParseTime: 258334012559
Routine: 2, ParseTime: 188298916476
Routine: 3, ParseTime: 147130715494
Routine: 4, ParseTime: 126473431218

JSON data: 1GB ParseTime(ns)
Routine: 1, ParseTime: 112820240001
Routine: 2, ParseTime: 69955550437
Routine: 3, ParseTime: 51142855868
Routine: 4, ParseTime: 43761123498

wonderflow · 2018-08-28T02:06:32Z

parser/csv/csv.go

+	for resultInfo := range resultChan {
+		parseResultSlice = append(parseResultSlice, resultInfo)
+	}
+	sort.Stable(parseResultSlice)


所有sort的地方加个判断，如果 routine==1，不做这个sort

wonderflow · 2018-08-28T03:16:07Z

parser/csv/csv.go

+	for resultInfo := range resultChan {
+		parseResultSlice = append(parseResultSlice, resultInfo)
+	}
+	if len(parseResultSlice) > 1 {


这个条件判断的不对

应该是判断Routine而不是slice，slice表达的是解析的这批数据量大小，单个Routine出来的slice可以很长，但是已经有序

wonderflow · 2018-08-28T03:53:25Z

parser/grok/grok.go

-		if err != nil {
+
+		if parseResult.Err != nil {
+			log.Debug(parseResult.Err)


这个删掉吧

wonderflow · 2018-08-28T04:04:12Z

parser/json/json.go

-			se.AddSuccess()
+
+		if parseResult.Err != nil {
+			log.Debug(parseResult.Err)


这个debug删掉吧

wonderflow · 2018-08-28T04:06:57Z

parser/parser.go

+				Line:  parseInfo.Line,
+				Index: parseInfo.Index,
+			}
+			continue


空行还需要传吗？是为了index吗

对， DatasourceSkipIndex 需要记录

wonderflow · 2018-08-28T05:51:05Z

LGTM

redHJ force-pushed the pdr-6966 branch 3 times, most recently from 7fe4d33 to cec6521 Compare August 15, 2018 08:45

redHJ changed the title ~~[WIP] csv,grok,json,ngnix,syslog parser添加多线程~~ [WIP] csv,grok,json,ngnix parser添加多线程 Aug 15, 2018

redHJ changed the title ~~[WIP] csv,grok,json,ngnix parser添加多线程~~ csv,grok,json,ngnix,syslog parser添加多线程 Aug 15, 2018

redHJ changed the title ~~csv,grok,json,ngnix,syslog parser添加多线程~~ csv,grok,json,ngnix parser添加多线程 Aug 15, 2018

redHJ force-pushed the pdr-6966 branch 2 times, most recently from f1ff457 to 706b030 Compare August 15, 2018 09:09

unknwon suggested changes Aug 16, 2018

View reviewed changes

redHJ force-pushed the pdr-6966 branch 3 times, most recently from 2f2d5a6 to 8911b8b Compare August 17, 2018 08:53

wonderflow reviewed Aug 20, 2018

View reviewed changes

unknwon approved these changes Aug 21, 2018

View reviewed changes

wonderflow reviewed Aug 21, 2018

View reviewed changes

redHJ force-pushed the pdr-6966 branch from 8911b8b to b0ad6a3 Compare August 23, 2018 05:18

redHJ changed the title ~~csv,grok,json,ngnix parser添加多线程~~ [WIP] csv,grok,json,ngnix parser添加多线程 Aug 23, 2018

redHJ force-pushed the pdr-6966 branch 2 times, most recently from 5cde0ce to 1f0b5d4 Compare August 27, 2018 08:15

redHJ changed the title ~~[WIP] csv,grok,json,ngnix parser添加多线程~~ csv,grok,json,ngnix parser添加多线程 Aug 27, 2018

redHJ force-pushed the pdr-6966 branch 4 times, most recently from 93eadfd to 8939a44 Compare August 27, 2018 09:56

wonderflow reviewed Aug 28, 2018

View reviewed changes

redHJ force-pushed the pdr-6966 branch 2 times, most recently from 8752cdb to 5cb09b8 Compare August 28, 2018 03:20

wonderflow reviewed Aug 28, 2018

View reviewed changes

redHJ force-pushed the pdr-6966 branch from 5cb09b8 to d3a5a56 Compare August 28, 2018 04:46

李红 added 5 commits August 28, 2018 12:51

change parse to multiple goroutine

65920d6

grok goroutine

8536899

json goroutine

63bb6a7

nginx goroutinue

8246c82

test

b7fbfb5

redHJ force-pushed the pdr-6966 branch from d3a5a56 to b7fbfb5 Compare August 28, 2018 04:54

wonderflow merged commit b110855 into qiniu:master Aug 28, 2018

redHJ deleted the pdr-6966 branch January 24, 2019 09:01

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

csv,grok,json,ngnix parser添加多线程 #709

csv,grok,json,ngnix parser添加多线程 #709

redHJ commented Aug 15, 2018 •

edited

Loading

unknwon Aug 16, 2018

unknwon Aug 16, 2018

unknwon Aug 16, 2018

wonderflow Aug 20, 2018

wonderflow Aug 20, 2018

wonderflow Aug 20, 2018

wonderflow Aug 20, 2018

wonderflow Aug 20, 2018

wonderflow Aug 21, 2018

redHJ Aug 23, 2018

unknwon commented Aug 21, 2018

wonderflow Aug 21, 2018

redHJ Aug 23, 2018

redHJ commented Aug 27, 2018

wonderflow Aug 28, 2018

wonderflow Aug 28, 2018

wonderflow Aug 28, 2018

wonderflow Aug 28, 2018

wonderflow Aug 28, 2018

wonderflow Aug 28, 2018

redHJ Aug 28, 2018

wonderflow commented Aug 28, 2018

csv,grok,json,ngnix parser添加多线程 #709

csv,grok,json,ngnix parser添加多线程 #709

Conversation

redHJ commented Aug 15, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

unknwon commented Aug 21, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

redHJ commented Aug 27, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wonderflow commented Aug 28, 2018

redHJ commented Aug 15, 2018 •

edited

Loading