heka从kalka中读取数据的示例分析

共计 3846 个字符，预计需要花费 10 分钟才能阅读完成。

这期内容当中丸趣 TV 小编将会给大家带来有关 heka 从 kalka 中读取数据的示例分析，文章内容丰富且以专业的角度为大家分析和叙述，阅读完这篇文章希望大家可以有所收获。

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

结果如下，

上述就是丸趣 TV 小编为大家分享的 heka 从 kalka 中读取数据的示例分析了，如果刚好有类似的疑惑，不妨参照上述分析进行理解。如果想知道更多相关知识，欢迎关注丸趣 TV 行业资讯频道。