transform

Documentation

Overview

Package transform 提供字段转换处理器

Index

• type Config
• type Processor
  • func New() *Processor
  • func (p *Processor) ConcurrencyCap() dataflow.ConcurrencyCap
  • func (p *Processor) Init(config []byte) error
  • func (p *Processor) Process(ctx context.Context, in <-chan types.Record, out chan<- types.Record) error

Types

type Config

type Config struct {
	// Mapping 字段映射: 原字段名 -> 新字段名
	Mapping map[string]string `json:"mapping"`
	// Add 要添加的字段
	Add map[string]interface{} `json:"add"`
	// Remove 要删除的字段
	Remove []string `json:"remove"`

	// Extract 提取指定字段的值作为新记录（支持点号分隔的嵌套路径）
	Extract string `json:"extract"`
	// ExtractKeep 提取时保留的原始字段列表
	ExtractKeep []string `json:"extract_keep"`
	// ExtractFlatten 当提取的字段是数组时，展开为多条记录
	ExtractFlatten bool `json:"extract_flatten"`
}

Config Processor 的配置

type Processor

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

Processor 转换记录的字段

func New

func New() *Processor

New 创建新的 Processor

func (*Processor) ConcurrencyCap

func (p *Processor) ConcurrencyCap() dataflow.ConcurrencyCap

ConcurrencyCap 声明支持并发

func (*Processor) Init

func (p *Processor) Init(config []byte) error

Init 初始化 Processor

func (*Processor) Process

func (p *Processor) Process(ctx context.Context, in <-chan types.Record, out chan<- types.Record) error

Process 处理数据，转换记录