go 줄별로 데이터를 읽는 구덩이

소스 주소:https://github.com/ma6174/blog/issues/10
먼저 로그 분석을 하는 Go 프로그램에서 말하자면 기본적인 기능은 한 줄 한 줄 데이터를 읽고 처리하는 것이다.코드는 대체로 다음과 같습니다.

func main() {
    scanner := bufio.NewScanner(os.Stdin)
    for scanner.Scan() {
        line := scanner.Text()
        if err := deal(line); err != nil {
            log.Println(line)
        }
    }
}

데이터는 Hadoop에서 직접cat으로 나온 것으로 파이프를 통해 로그 처리 프로그램에 전송되기 때문에 여기에 입력된 원본은 stdin입니다.
실제 처리를 할 때 비교적 이상한 현상을 발견했다. 로그 파일의 총 줄 수는 차이가 크지 않지만 어떤 파일의 처리 시간은 다른 것보다 현저히 짧고 이런 로그 정보를 볼 수 있다. cat: Unable to write to output stream..이 로그는 Go에 있는 것이 아닙니다. Hadoop이 이 오류를 보고한 것으로 확인되었습니다. 처음에는 Hadoop이 어떤 상황에서 이 오류가 발생할 수 있다고 생각했습니다. 이 오류가 발생하면 일부 파일이 프로그램을 처리하지 않고 종료된 것을 의미하기 때문입니다.이것은 용납할 수 없다.그래서 운비에게 버그를 보고했다.
윈웨이 측은 이 오류가hadoop에서 보고된 것이라고 인정하지만, 이 오류가 발생한 것은 보통stream이 끊겼기 때문이다. 예를 들어head 조작이 있으면 이 오류가 발생한다.

$ hdfs dfs -cat  0.log | head -1
*****log contents*****
cat: Unable to write to output stream.

다음 처리 프로그램의 문제이지 Hadoop의 문제가 아니라는 것을 증명하기 위해 운위 쪽에서 다음과 같은 조작을 했습니다.
Hadoop의 파일을 로컬로 다운로드하기 UNIXcat 명령으로 직접 데이터를 프로세서로 리디렉션 cat와 프로세서의 반환값 보기

$ cat 0.log | go_program
$ echo ${PIPESTATUS[@]}
141 0

cat은 0이 아닙니다. 다음 프로그램은 0으로 되돌아옵니다. UNIX 기본 프로그램은 일반적으로 문제가 없습니다. 뒤에 있는 Go 프로그램에 문제가 있는 것 같습니다.검사 검증을 시작합니다.
먼저 확인해야 할 질문은 두 가지입니다.
Go 프로그램이 정말 데이터를 처리하지 못하고 종료되었습니까?
비정상적으로 탈퇴하면 어느 줄을 처리할 때 탈퇴하는 거예요?
이 문제를 조회하는 것도 비교적 쉽다. 처리된 줄마다 로그를 출력하고 줄 번호를 출력하면 어느 줄로 처리되었는지 알 수 있고 어느 줄에서 오류가 발생했는지 알 수 있다.
프로그램을 수정한 후에 다시 실행합니다. 확실히 Go 프로그램이 앞당겨 종료되었고 처리하지 않은 다음 줄의 유일한 특이점은 이 줄이 매우 길다는 것입니다.설마 Go가 긴 줄을 처리할 수 없단 말인가?이것은 발생할 수 없을 것 같지만, 그래도 한번 검증해야 한다.
우선 리뷰 코드를 다시 만듭니다. 이치대로 말하면 프로그램이 잘못되면 오류 로그가 발생합니다. 왜 Go 프로그램은 오류를 보고하지 않았을 뿐만 아니라 정상적으로 종료되었습니까?우선review 데이터 처리 코드입니다. 문제를 발견하지 못하고bufio라고 의심하기 시작했습니다.Scanner의 질문에 Go 문서를 다시 봤는데 홈페이지에 example가 이렇게 적혀 있습니다.

scanner := bufio.NewScanner(os.Stdin)
for scanner.Scan() {
    fmt.Println(scanner.Text()) // Println will add back the final '
'
}
if err := scanner.Err(); err != nil {
    fmt.Fprintln(os.Stderr, "reading standard input:", err)
}

원래는 scanner에 Err() 방법이 하나 더 있었다. 공식적으로는 Err returns the first non-EOF error that was encountered by the Scanner.라고 했다. 즉, scanner.Scan()가 잘못되면 오류 정보는 Err() 방법을 통해 얻을 수 있다. 나의 go 프로그램은 이Err를 무시했다. 오류가 발생한 후에 for 순환을 종료하면 바로 종료된다. 도대체 무슨 오류가 있는 거야?코드가 완전하게 보충된 후에 이러한 오류를 보았습니다: bufio.Scanner: token too long.
지금까지 확인된 것은 Go 프로그램 문제입니다. 데이터를 읽을 때 오류가 발생했습니다. 위의 오류는 무슨 뜻입니까?Go 코드를 뒤집습니다.

~/go$ grep -n ` find -name "*.go"` -e "bufio.Scanner: token too long"
./src/pkg/bufio/scan.go:61:     ErrTooLong         = errors.New("bufio.Scanner: token too long")

확실히 이 오류가 있다.go, 이 오류를 어디서 보고했는지 봐라ErrTooLong:

~/go$ grep -n ` find -name "*.go"` -e "ErrTooLong"
./src/pkg/bufio/scan.go:61:     ErrTooLong         = errors.New("bufio.Scanner: token too long")
./src/pkg/bufio/scan.go:147:                            s.setErr(ErrTooLong)
./src/pkg/bufio/scan_test.go:244:       if err != ErrTooLong {
./src/pkg/bufio/scan_test.go:245:               t.Fatalf("expected ErrTooLong; got %s", err)

scan.go 147 줄에 발생한 오류는 구체적인 코드를 보십시오:

   144                  // Is the buffer full? If so, resize.
   145                  if s.end == len(s.buf) {
   146                          if len(s.buf) >= s.maxTokenSize {
   147                                  s.setErr(ErrTooLong)
   148                                  return false
   149                          }
   150                          newSize := len(s.buf) * 2
   151                          if newSize > s.maxTokenSize {
   152                                  newSize = s.maxTokenSize
   153                          }
   154                          newBuf := make([]byte, newSize)
   155                          copy(newBuf, s.buf[s.start:s.end])
   156                          s.buf = newBuf
   157                          s.end -= s.start
   158                          s.start = 0
   159                          continue
   160                  }

Scanner 모듈의 코드를 봤는데 원래 이랬다. Scanner가 초기화할 때 maxTokenSize를 설정했는데 이 값은 기본값MaxScanTokenSize = 64 * 1024으로 한 줄의 길이가 64*1024 즉 65536보다 크면 ErrTooLong 오류가 발생한다.이렇게 하면 이전에 왜 오랫동안 잘못 보고한 문제를 처리했는지 설명할 수 있다.
일지 한 줄이 65536보다 길지 않을 수도 있으니 문제는 두 가지 방안이 있다.
maxTokenSize의 값을 높이다다른 함수로 바꾸기방안 1에 대해 Scnner를 자세히 보았지만 Go 원본 코드를 수정하고 Go를 다시 컴파일하지 않으면 이 값을 수정할 수 있는 인터페이스가 없다는 것을 발견하지 못했다. 이것은 너무 번거롭다.공식 문서에서 다음과 같은 문장을 발견했다.
Scanning stops unrecoverably at EOF, the first I/O error, or a token too large to fit in the buffer. When a scan stops, the reader may have advanced arbitrarily far past the last token. Programs that need more control over error handling or large tokens, or must run sequential scans on a reader, should use bufio.Reader instead.
이것은 Token이 너무 크면 (줄이 너무 길면) 사용해야 한다고 알려준다. bufio.Reader여기가 첫 번째 구덩이인데 처음에 사용했을 때 문서를 자세히 보지 않고 Go 코드를 자세히 연구하지 못했기 때문이다.여기서 우리는 줄 길이가 65536을 초과하지 않는다는 것을 확정하지 않으면bufio를 사용하지 말라는 교훈을 얻을 수 있다.Scanner!
그럼 이제 착실하게 bufio.Reader하자.그럼 bufio.Reader 비슷한 질문이 있나요?bufio.Scanner 코드를 봤는데 이것도 버퍼 크기 제한이 있고 기본 버퍼 크기는 bufio.Reader입니다. 다행히 함수4096가 이 버퍼 크기를 조절할 수 있습니다.
아니면 이전의 수요, 어떻게 줄별로 데이터를 읽습니까?NewReaderSize는 bufio.Reader 함수를 제공했는데 문서에서 이렇게 말했다.
ReadLine is a low-level line-reading primitive. Most callers should use ReadBytes('') or ReadString('') instead or use a Scanner.
이 함수가 아래쪽을 비교한다는 뜻으로 ReadBytes나 ReadString 또는 Scanner를 사용하는 것을 권장합니다.설명서를 계속 보려면 다음과 같이 하십시오.
ReadLine tries to return a single line, not including the end-of-line bytes. If the line was too long for the buffer then isPrefix is set and the beginning of the line is returned. The rest of the line will be returned from future calls. isPrefix will be false when returning the last fragment of the line. The returned buffer is only valid until the next call to ReadLine. ReadLine either returns a non-nil line or it returns an error, never both.
여기서 버퍼 설정의 역할을 알 수 있습니다. ReadLine은 가능한 한 전체 줄을 읽고 되돌려줍니다. 그러나 줄이 너무 길어서 버퍼가 가득 차면 전체 줄이 아니라 버퍼 안의 내용을 되돌려주고 isPrefix를true로 설정합니다.한 줄을 다 읽을 때까지 ReadLine을 계속 호출해야 합니다.그리고 외부 호출 프로그램은 이 블록을 조합해야만 완전한 줄을 구성할 수 있다.isPrefix뿐만 아니라 접두사도 처리해야 해요. 귀찮아요!우리가 주동적으로 매우 큰 버퍼를 설정하지 않으면, 전제는 가장 긴 줄의 길이를 알아야 한다는 것이다. 대부분의 경우, 이것은 미리 알 수 없는 것이다.어쩐지 ReadBytes나 ReadString 또는 Scanner를 사용하는 것을 권장합니다.
앞서 논의한 바와 같이 Scanner는 줄이 너무 길 때 문제가 있다. ReadBytes와 ReadString은 원리적으로 똑같다. 여기서 우리는 ReadString을 예로 들어 이 함수를 살펴본다.ReadString의 원리는 매우 간단하고 사용도 편리하다. 기본적인 방법은 하나의 구분자를 지정하는 것이다. 예를 들어 우리가 한 줄을 읽으면 구분자를 ReadLine()로 지정하면 ReadString은 구분자
를 발견하거나 오류가 발생할 때까지 데이터를 계속 읽는다.홈페이지에서는 이렇게 말했다.
ReadString reads until the first occurrence of delim in the input, returning a string containing the data up to and including the delimiter. If ReadString encounters an error before finding a delimiter, it returns the data read before the error and the error itself (often io.EOF). ReadString returns err != nil if and only if the returned data does not end in delim. For simple uses, a Scanner may be more convenient.
(ps: 여기 또 Scanner 추천...)
ReadString은 버퍼 크기 제한이 있습니까?이것도 사실은 여전히 있지만, 코드 안에서 이 버퍼 구역의 문제를 잘 처리했다. 즉, 되돌아오는 한 줄이 틀림없이 완전할 것이라는 것을 보장할 수 있다.적어도 ReadLine 함수보다 사용하기 편합니다.
그럼 ReadString으로 줄별로 읽으면 구덩이가 있을까요?답은 긍정적이다.

package main

import (
    "bufio"
    "fmt"
    "io"
    "strings"
)

func main() {
    s := "a
b
c"
    reader := bufio.NewReader(strings.NewReader(s))
    for {
        line, err := reader.ReadString('
')
        if err != nil {
            if err == io.EOF {
                break
            }
            panic(err)
        }
        fmt.Printf("%#v
", line)
    }
}

위 코드의 출력은 다음과 같습니다.

"a
"
"b
"

왜
출력이 없습니까?이곳은 구덩이라고 할 수 있다.앞서 논의한 바와 같이 줄별로 읽으면 ReadString 함수는 c를 분할해야 한다. 위의 특수한 상황은 데이터 끝에
가 없을 때 EOF에 구분자
가 없을 때까지 EOF 오류를 되돌려주지만 라인 안에 데이터가 있어 처리하지 않으면 마지막 줄이 빠진다.간단하게 수정:

package main

import (
    "bufio"
    "fmt"
    "io"
    "strings"
)

func main() {
    s := "a
b
c"
    reader := bufio.NewReader(strings.NewReader(s))
    for {
        line, err := reader.ReadString('
')
        if err != nil {
            if err == io.EOF {
                fmt.Printf("%#v
", line)
                break
            }
            panic(err)
        }
        fmt.Printf("%#v
", line)
    }
}

이렇게 하면 다음과 같이 출력됩니다.

"a
"
"b
"
"c"

마지막 줄 뒤에 없기 때문에 이렇게 처리할 때도 조심해야 한다
.
그에 비해python 처리는 간단하고 뚜렷하며 생성기가 있어 for 순환을 직접 반복하면 된다.

import StringIO

s = "a
b
c"

sio = StringIO.StringIO(s)

for line in sio:
    print line,