请教一个 clickhouse 和 filebeat 的问题
proxychains · 2023-04-16 01:58:37 +08:00 via Android · 401 次点击这是一个创建于 378 天前的主题，其中的信息可能已经有所发展或是发生改变。
如题，当前使用 kafka filebeat clickhouse 收集网站访问日志入库. nginx 日志换成 json 格式扔到 kafka ，再在 clickhouse 分别创建实体表和 kafka 引擎表消费 kafka 中的 json 以及视图表来将 nginx 访问日志的按照 字段 分别存入 clickhouse 实体表.
filebeat 中配置了 field_under_root.
kafka 中对应的 topic 数据大概是这样的
{"@timestamp":"timestamp","uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}
入库后实体表字段大概是这样的:
uri http_code remote_ip
查询的时候大概是这样的:
select uri, http_code, remote_ip from ngx_acc_log_entry;
/abc.html 404 10.2.2.3
/index.php 200 10.5.6.6
...
但是 iis 的日志不支持 json ,他的日志各项值是以空格为分隔符. 所以我在 filebeat 配置文件中使用 dissect 将日志转化为了 json. 在 kafka 中大概是这样的:
{"@timestamp":"timestamp","dissect":{"uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}
}
实际上我想要的数据在 dissect 这个字段内，然而在 clickhouse 中创建 kafka 引擎表的时候解析不到 dissect 这个对象中的值了.
我尝试了两个办法：
1. 在 clickhouse 创建 kafka 引擎表 的时候，将字段 dissect 类型设置为 Nested. 这种情况下从读到的数据是 {} ，为空.
2. 尝试配置 filebeat 把日志需要入库的值从 "dissect" 下挪出来，也就是变成
{"@timestamp":"timestamp","uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}.
没找到解决方案， 不知道 v 友是否有做过这种的？
filebeat 中配置了 field_under_root.
kafka 中对应的 topic 数据大概是这样的
{"@timestamp":"timestamp","uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}
入库后实体表字段大概是这样的:
uri http_code remote_ip
查询的时候大概是这样的:
select uri, http_code, remote_ip from ngx_acc_log_entry;
/abc.html 404 10.2.2.3
/index.php 200 10.5.6.6
...
但是 iis 的日志不支持 json ,他的日志各项值是以空格为分隔符. 所以我在 filebeat 配置文件中使用 dissect 将日志转化为了 json. 在 kafka 中大概是这样的:
{"@timestamp":"timestamp","dissect":{"uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}
}
实际上我想要的数据在 dissect 这个字段内，然而在 clickhouse 中创建 kafka 引擎表的时候解析不到 dissect 这个对象中的值了.
我尝试了两个办法：
1. 在 clickhouse 创建 kafka 引擎表 的时候，将字段 dissect 类型设置为 Nested. 这种情况下从读到的数据是 {} ，为空.
2. 尝试配置 filebeat 把日志需要入库的值从 "dissect" 下挪出来，也就是变成
{"@timestamp":"timestamp","uri":"/abc.html","http_code":"404","remote_ip":"10.2.2.3"}.
没找到解决方案， 不知道 v 友是否有做过这种的？
2 条回复
 |
1
tyit 312 天前
CREATE TABLE ngx_acc_log_entry (
"@timestamp" DateTime, dissect String, uri String, http_code String, remote_ip String ) ENGINE = Kafka SETTINGS kafka_broker_list = 'kafka:9092', kafka_topic_list = 'your_topic', kafka_group_name = 'your_group', kafka_format = 'JSONEachRow', kafka_row_format = 'FlatJson'; INSERT INTO ngx_acc_log_entry SELECT "@timestamp", dissect, JSONExtractString(dissect, 'uri') AS uri, JSONExtractString(dissect, 'http_code') AS http_code, JSONExtractString(dissect, 'remote_ip') AS remote_ip FROM kafka; |
 |
2
proxychains OP @tyit 哇，时隔两个月，虽然已经解决了，但仍然十分感谢回复🥺
|
Select Language