文档章节

KafkaInput & esoutput

水东流
 水东流
发布于 2016/07/21 17:44
字数 692
阅读 9
收藏 0

heka从kalka中读取数据。

配置:

[hekad]
maxprocs = 2

[KafkaInputExample]
type = "KafkaInput"
topic = "test"
addrs = ["localhost:9092"]

[RstEncoder]

[LogOutput]
message_matcher = "TRUE"
encoder = "RstEncoder"

上述配置只有从kalfka中读取数据并显示到console,写到kalfka中数据,

结果

:Timestamp: 2016-07-21 09:39:46.342093657 +0000 UTC
:Type: heka.kafka
:Hostname: master
:Pid: 0
:Uuid: 501b0a0e-63a9-4eee-b9ca-ab572c17d273
:Logger: KafkaInputExample
:Payload: {"msg":"Start Request","event":"artemis.web.ensure-running1","userid":"12","extra":{"workspace-id":"cN907xLngi"},"time":"2015-05-06T    20:40:05.509926234Z","severity":1}
:EnvVersion: 
:Severity: 7
:Fields:
    | name:"Key" type:bytes value:
    | name:"Topic" type:string value:"test"
    | name:"Partition" type:integer value:0
    | name:"Offset" type:integer value:8

 

读取出来的数据放到了payload中,而fileds中存放了读取kalkfa中的一些信息。那么可以使用jsondecoder进行解析。

[hekad]
maxprocs = 2

[KafkaInputExample]
type = "KafkaInput"
topic = "test"
addrs = ["localhost:9092"]
decoder="JsonDecoder"

[JsonDecoder]
type = "SandboxDecoder"
filename = "lua_decoders/json.lua"

        [JsonDecoder.config]
        type = "artemis"
        payload_keep = true
        map_fields = true
        Severity = "severity"

[RstEncoder]

[LogOutput]
message_matcher = "TRUE"
encoder = "RstEncoder"

 

结果如下:

:Timestamp: 2016-07-21 09:42:34 +0000 UTC
:Type: artemis
:Hostname: master
:Pid: 0
:Uuid: 3965285c-70ac-4069-a1a3-a9bcf518d3e8
:Logger: KafkaInputExample
:Payload: {"msg":"Start Request","event":"artemis.web.ensure-running2","userid":"11","extra":{"workspace-id":"cN907xLngi"},"time":"2015-05-06T    20:40:05.509926234Z","severity":1}
:EnvVersion: 
:Severity: 1
:Fields:
    | name:"time" type:string value:"2015-05-06T    20:40:05.509926234Z"
    | name:"msg" type:string value:"Start Request"
    | name:"userid" type:string value:"11"
    | name:"event" type:string value:"artemis.web.ensure-running2"
    | name:"extra.workspace-id" type:string value:"cN907xLngi"

经过decoder解析之后,fileds发生了改变,但是我们可以看到Logger显示的还是KafkaInputExample,说明数据不是decoder产生,而是Input产生,只不过使用了decoder进行了解析,重写改写了fields而已。

接下来,把数据录入都es中吧。
[hekad]
maxprocs = 2

[KafkaInputExample]
type = "KafkaInput"
topic = "test"
addrs = ["localhost:9092"]
decoder="JsonDecoder"

[JsonDecoder]
type = "SandboxDecoder"
filename = "lua_decoders/json.lua"

        [JsonDecoder.config]
        type = "artemis"
        payload_keep = true
        map_fields = true
        Severity = "severity"

[ESJsonEncoder]
index = "%{Type}-%{%Y.%m.%d}"
es_index_from_timestamp = true
type_name = "%{Type}"
    [ESJsonEncoder.field_mappings]
    Timestamp = "@timestamp"
    Severity = "level"

[ElasticSearchOutput]
message_matcher = "TRUE"
encoder = "ESJsonEncoder"
flush_interval = 1

导入到es中,也需要json,所以使用ESJsonEncoder,同时指定索引名字和类型。执行结果如下,

可以看到,除了heka中元数据field之外,还有JsonDecoder生成field啊,其实是截取JsonDecoder中的fields属性中拿出。注意,Payload不解析。

:Fields:
    | name:"time" type:string value:"2015-05-06T    20:40:05.509926234Z"
    | name:"msg" type:string value:"Start Request"
    | name:"userid" type:string value:"11"
    | name:"event" type:string value:"artemis.web.ensure-running2"
    | name:"extra.workspace-id" type:string value:"cN907xLngi"

这些field当然随着数据不同而不同,那么称之为dynamic fileds。

入es的时候,可以指定提取哪些dynamic fields,

fields=["Timestamp","Uuid","Type","Logger","Pid","Hostname","DynamicFields"]
dynamic_fields=["msg","userid"]

只要使用dynamic_fileds,就必须要在fields中指定DynamicFields。

如果没有dynamic_fileds,那么fields只能列举几个固定的属性,参照官方文档即可。

 

完成的列子:

[hekad]
maxprocs = 2

[KafkaInputExample]
type = "KafkaInput"
topic = "test"
addrs = ["localhost:9092"]
decoder="JsonDecoder"

[JsonDecoder]
type = "SandboxDecoder"
[hekad]
maxprocs = 2

[KafkaInputExample]
type = "KafkaInput"
topic = "test"
addrs = ["localhost:9092"]
decoder="JsonDecoder"

[JsonDecoder]
type = "SandboxDecoder"
filename = "lua_decoders/json.lua"

        [JsonDecoder.config]
        type = "artemis"
        payload_keep = true
        map_fields = true
        Severity = "severity"

[ESJsonEncoder]
index = "%{Type}-%{%Y.%m.%d}"
es_index_from_timestamp = true
type_name = "%{Type}"
fields=["Timestamp","Uuid","Type","Logger","Pid","Hostname","DynamicFields"]
dynamic_fields=["msg","userid"]

raw_bytes_fields=["Payload"]
    [ESJsonEncoder.field_mappings]
    Timestamp = "@timestamp"
    Severity = "level"

[ElasticSearchOutput]
message_matcher = "TRUE"
encoder = "ESJsonEncoder"
flush_interval = 1

结果如下,

© 著作权归作者所有

水东流
粉丝 4
博文 51
码字总数 23858
作品 0
海淀
程序员
私信 提问

暂无文章

OSChina 周一乱弹 —— 人生,还真是到处是意外

Osc乱弹歌单(2019)请戳(这里) 【今日歌曲】 @这次装个文艺青年吧 :#今日歌曲推荐# 分享lil peep的单曲《High School》 《High School》- lil peep 手机党少年们想听歌,请使劲儿戳(这里...

小小编辑
20分钟前
1
0
Spring使用ThreadPoolTaskExecutor自定义线程池及实现异步调用

多线程一直是工作或面试过程中的高频知识点,今天给大家分享一下使用 ThreadPoolTaskExecutor 来自定义线程池和实现异步调用多线程。 一、ThreadPoolTaskExecutor 本文采用 Executors 的工厂...

CREATE_17
今天
6
0
CSS盒子模型

CSS盒子模型 组成: content --> padding --> border --> margin 像现实生活中的快递: 物品 --> 填充物 --> 包装盒 --> 盒子与盒子之间的间距 content :width、height组成的 内容区域 padd......

studywin
今天
7
0
修复Win10下开始菜单、设置等系统软件无法打开的问题

因为各种各样的原因导致系统文件丢失、损坏、被修改,而造成win10的开始菜单、设置等系统软件无法打开的情况,可以尝试如下方法解决 此方法只在部分情况下有效,但值得一试 用Windows键+R打开...

locbytes
昨天
8
0
jquery 添加和删除节点

本文转载于:专业的前端网站➺jquery 添加和删除节点 // 增加一个三和一节点function addPanel() { // var newPanel = $('.my-panel').clone(true) var newPanel = $(".triple-panel-con......

前端老手
昨天
8
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部