Documentation
¶
Index ¶
- Constants
- Variables
- type StreamAggregatorNode
- func (x *StreamAggregatorNode) Def() types.ComponentForm
- func (x *StreamAggregatorNode) Destroy()
- func (x *StreamAggregatorNode) Init(ruleConfig types.Config, configuration types.Configuration) error
- func (x *StreamAggregatorNode) New() types.Node
- func (x *StreamAggregatorNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)
- func (x *StreamAggregatorNode) Type() string
- type StreamAggregatorNodeConfiguration
- type StreamTransformNode
- func (x *StreamTransformNode) Desc() string
- func (x *StreamTransformNode) Destroy()
- func (x *StreamTransformNode) Init(ruleConfig types.Config, configuration types.Configuration) error
- func (x *StreamTransformNode) New() types.Node
- func (x *StreamTransformNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)
- func (x *StreamTransformNode) Type() string
- type StreamTransformNodeConfiguration
Constants ¶
const RelationTypeWindowEvent = "window_event"
RelationTypeWindowEvent 表示窗口事件关系类型，用于聚合结果的链路传递
const WindowEventMsgType = "window_event"
WindowEventMsgType 表示窗口事件消息类型，用于标识聚合结果消息
Variables ¶
var ( ErrAggregatorSQLEmpty = errors.New("aggregator SQL query is required") ErrNotAggregatorQuery = errors.New("SQL does not contain aggregation functions, use x/streamTransform instead") ErrAggregatorSQLExecution = errors.New("failed to execute aggregator SQL") ErrAggregatorChainCtxNil = errors.New("chain context is nil") ErrAggregatorNodeIdEmpty = errors.New("self node id is empty") ErrAggregatorChainIdEmpty = errors.New("chain id is empty") )
错误定义
var ( ErrTransformSQLEmpty = errors.New("transform SQL query is required") ErrNotTransformQuery = errors.New("SQL contains aggregation functions, use x/streamAggregator instead") ErrTransformSQLExecution = errors.New("failed to execute transform SQL") ErrNotMatchWhereCondition = errors.New("not match WHERE condition") ErrStreamsqlInstanceNil = errors.New("streamsql instance is nil") ErrArrayProcessingFailed = errors.New("failed to process array data") ErrUnsupportedDataType = errors.New("only JSON data type is supported") ErrDataProcessingFailed = errors.New("failed to process message data") // 元数据键名，用于标识数据是否匹配转换条件 Match = "match" MatchTrue = "true" MatchFalse = "false" )
错误定义和常量
Functions ¶
This section is empty.
Types ¶
type StreamAggregatorNode ¶
type StreamAggregatorNode struct {
// 节点配置
Config StreamAggregatorNodeConfiguration
// contains filtered or unexported fields
}
StreamAggregatorNode 流聚合器节点
功能说明： - 专门处理聚合查询，如窗口聚合、分组聚合、统计计算等 - 支持单条数据和数组数据输入，数组数据会被逐条添加到聚合流中 - 聚合结果通过 `window_event` 关系链传递到下一个节点，而不是通过普通的Success链 - 原始输入数据（无论单条还是数组）都会通过 `Success` 链继续传递，保持数据流的连续性
数据流向： - 输入数据 -> 添加到聚合流 -> 原始数据通过Success链传递 - 聚合触发 -> 聚合结果通过window_event链传递
注意事项： - 聚合结果通过全局`Config.OnEnd`回调返回，而不是通过消息处理上下文的OnEnd回调返回 - 聚合计算是异步进行的，不会阻塞原始数据的流转 - 窗口触发时机由StreamSQL引擎根据时间窗口或数据量自动决定
func (*StreamAggregatorNode) Def ¶ added in v0.36.0
func (x *StreamAggregatorNode) Def() types.ComponentForm
Def returns the component form definition
func (*StreamAggregatorNode) Init ¶
func (x *StreamAggregatorNode) Init(ruleConfig types.Config, configuration types.Configuration) error
Init 初始化节点 该方法在节点被加载时调用，用于验证配置和初始化StreamSQL实例
func (*StreamAggregatorNode) OnMsg ¶
func (x *StreamAggregatorNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)
OnMsg 处理消息 支持单条数据和数组数据： - 单条数据：直接添加到聚合流中 - 数组数据：遍历每个元素并逐条添加到聚合流中 无论哪种情况，原始消息都会通过Success链继续传递
type StreamAggregatorNodeConfiguration ¶
type StreamAggregatorNodeConfiguration struct {
// SQL is the aggregation query statement (must contain GROUP BY, aggregation or window functions).
// Example: SELECT AVG(temperature) as avg_temp FROM stream GROUP BY TumblingWindow('5s')
SQL string `` /* 183-byte string literal not displayed */
}
StreamAggregatorNodeConfiguration 流聚合器节点配置
type StreamTransformNode ¶
type StreamTransformNode struct {
// 节点配置
Config StreamTransformNodeConfiguration
// contains filtered or unexported fields
}
StreamTransformNode 流转换器节点
功能说明： - 专门处理非聚合查询，如数据过滤、字段转换、格式变换等 - 支持单条数据和数组数据输入：
- 单条数据：直接进行转换处理
- 数组数据：遍历转换每个元素，将成功转换的结果合并成数组输出
- 数据符合WHERE条件并转换成功，则通过`Success`链输出，否则通过`Failure`链输出 - 对于数组输入，只要有任何元素转换成功，就会通过Success链输出合并结果
数据流向： - 单条数据：输入 -> SQL转换 -> Success/Failure输出 - 数组数据：输入数组 -> 逐个转换 -> 合并成功结果 -> Success输出（如有成功项）/ Failure输出（全部失败）
注意事项： - 转换处理是同步的，会阻塞当前消息的处理 - WHERE条件不匹配的数据会被过滤掉，不包含在输出结果中 - 对于数组输入，部分元素转换失败不会影响整体结果，只影响最终数组的元素数量
func (*StreamTransformNode) Desc ¶ added in v0.36.0
func (x *StreamTransformNode) Desc() string
Desc returns the component description
func (*StreamTransformNode) Init ¶
func (x *StreamTransformNode) Init(ruleConfig types.Config, configuration types.Configuration) error
Init 初始化节点 该方法在节点被加载时调用，用于验证配置和初始化StreamSQL实例
func (*StreamTransformNode) OnMsg ¶
func (x *StreamTransformNode) OnMsg(ctx types.RuleContext, msg types.RuleMsg)
OnMsg 处理消息 支持单条数据和数组数据：
- 单条数据：直接进行SQL转换，成功则通过Success链输出，失败则通过Failure链输出
- 数组数据：遍历每个元素进行转换，将所有成功的结果合并成数组输出 如果至少有一个元素转换成功，则通过Success链输出；如果全部失败，则通过Failure链输出
type StreamTransformNodeConfiguration ¶
type StreamTransformNodeConfiguration struct {
// SQL is the non-aggregation query statement (filter, transform, field selection).
// Example: SELECT temperature, humidity FROM stream WHERE temperature > 20
SQL string `` /* 156-byte string literal not displayed */
}
StreamTransformNodeConfiguration 流转换器节点配置
Source Files