Bufio 流式解析 NDJSON

示例说明

NDJSON 逐行可解析的特性让它适合流式处理：Scanner 按换行切分并交付单行字节，配合 json.Unmarshal 即可逐条消费，内存与文件大小无关。默认缓冲六十四 KiB 超限即报错——显式调大到四兆并声明上限，超长行被确定性拒绝而非崩溃。坏行只计数不中断，统计结果量化数据质量。

技术要点

• Buffer 显式设定行长上限
• 坏行隔离计数维持管道韧性
• 每行检查 context 支持取消

代码解读

parseNDJSON 的循环里先查 ctx 取消——长流任务可被随时中止；空行跳过、解析失败与缺字段的行进 rejected 计数。stringsReader 是演示用的最小 Reader 实现，生产里直接传文件或 HTTP 响应体。返回值把好坏计数与扫描器错误一起交给调用方。

使用建议

行长上限按真实数据分布设定并监控 rejected 率。需要行级错误详情时收集行号与片段。超大文件配 gzip 解压 Reader 组成管道。

代码示例

package main

import (
	"bufio"
	"context"
	"encoding/json"
	"fmt"
	"io"
)

type Event struct {
	Kind string `json:"kind"`
	Data int    `json:"data"`
}

func parseNDJSON(ctx context.Context, r io.Reader) (valid int, rejected int, err error) {
	scanner := bufio.NewScanner(r)

	// 默认缓冲 64KiB，超长行会报 ErrTooLong，按业务上调
	buf := make([]byte, 0, 64*1024)
	scanner.Buffer(buf, 4*1024*1024)

	line := 0
	for scanner.Scan() {
		if err := ctx.Err(); err != nil {
			return valid, rejected, err
		}
		line++

		raw := scanner.Bytes()
		if len(raw) == 0 {
			continue
		}

		var event Event
		if jsonErr := json.Unmarshal(raw, &event); jsonErr != nil {
			rejected++ // 坏行隔离计数，不中断整条流
			continue
		}
		if event.Kind == "" {
			rejected++
			continue
		}
		valid++
	}

	return valid, rejected, scanner.Err()
}

func main() {
	stream := `{"kind":"view","data":1}
{"kind":"click","data":2}
not-json-at-all
{"kind":"purchase","data":3}
`

	valid, rejected, err := parseNDJSON(
		context.Background(),
		stringsReader(stream),
	)
	fmt.Println("valid:", valid, "rejected:", rejected, "err:", err)
}

func stringsReader(s string) io.Reader { return &stringReader{s: s} }

type stringReader struct{ s string }

func (r *stringReader) Read(p []byte) (int, error) {
	if len(r.s) == 0 {
		return 0, io.EOF
	}
	n := copy(p, r.s)
	r.s = r.s[n:]
	return n, nil
}