kafka
使用sink kafka将日志数据发送至下游Kafka。
Example
sink:
type: kafka
brokers: ["127.0.0.1:6400"]
topic: "log-${fields.topic}"
brokers
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| brokers |
string数组 |
必填 |
无 |
发送日志至Kafka的brokers地址 |
topic
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| topic |
string |
非必填 |
loggie |
发送日志至Kafka的topic |
可使用${a.b}的方式,获取event里的字段值作为具体的topic名称。
比如,一个event为:
{
"topic": "loggie",
"hello": "world"
}
可配置topic: ${topic},此时该event发送到Kafka的topic为"loggie"。
同时支持嵌套的选择方式:
{
"fields": {
"topic": "loggie"
},
"hello": "world"
}
可配置topic: ${fields.topic},同样也会发送到topic "loggie"。
balance
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| balance |
string |
非必填 |
roundRobin |
负载均衡策略,可填hash、roundRobin、leastBytes |
compression
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| compression |
string |
非必填 |
gzip |
日志发送至Kafka的压缩策略,可填gzip、snappy、lz4、zstd |
maxAttempts
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| maxAttempts |
int |
非必填 |
10 |
发送最多重试次数 |
batchSize
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| batchSize |
int |
非必填 |
100 |
发送时每个batch最多包含的数据个数 |
batchBytes
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| batchBytes |
int |
非必填 |
1048576 |
每个发送请求包含的最大字节数 |
batchTimeout
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| batchTimeout |
time.Duration |
非必填 |
1s |
形成每个发送batch的最长时间 |
readTimeout
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| readTimeout |
time.Duration |
非必填 |
10s |
读取超时时间 |
writeTimeout
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| writeTimeout |
time.Duration |
非必填 |
10s |
写入超时时间 |
requiredAcks
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| requiredAcks |
int |
非必填 |
0 |
等待ack参数,可为0、1、-1 |
0: 不要求ack
1: 等待leader partition ack
-1: 等待ISR中所有replica ack
sasl
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| sasl |
|
非必填 |
|
SASL authentication |
| sasl.type |
string |
必填 |
|
SASL类型,可为:plain、scram |
| sasl.userName |
string |
必填 |
|
用户名 |
| sasl.password |
string |
必填 |
|
密码 |
| sasl.algorithm |
string |
type=scram时必填 |
|
type=scram时使用的算法,可选sha256、sha512 |
partitionKey
字段 |
类型 |
是否必填 |
默认值 |
含义 |
| partitionKey |
string |
非必填 |
无 |
控制发送至topic下哪个分区 |
与topic相似,可使用${a.b}的方式,获取event里的字段值作为具体的topic名称。