[
https://issues.apache.org/jira/browse/FLUME-2787?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14941281#comment-14941281
]
Tianji Li edited comment on FLUME-2787 at 10/2/15 3:45 PM:
-----------------------------------------------------------
Great point!
The default serializer is ElasticSearchLogStashEventSerializer, which uses the
following mapping. I am not sure if this is what you meant 'works well' :-). In
our case, we want our field names to be exactly the ones we set, which will be
used in our ElasticSearch queries.
{
"noserializer-2015-10-01":{
"mappings":{
"logs":{
"properties":{
"@fields":{
"properties":{
"timestamp":{
"type":"string"
}
}
},
"@message":{
"type":"string"
},
"@timestamp":{
"type":"date",
"format":"dateOptionalTime"
}
}
}
}
}
The above mapping is generated with this config file:
// Step 0: Define the agent
agent.sources = execSource
agent.channels = dynamicChannel
agent.sinks = dynamicSink
// Step 1: Define the exec source
agent.sources.execSource.type = exec
agent.sources.execSource.command = tail /tmp/some.log
agent.sources.execSource.interceptors = timestampInterceptor
agent.sources.execSource.interceptors.timestampInterceptor.type =
org.apache.flume.interceptor.TimestampInterceptor$Builder
agent.sources.execSource.interceptors.timestampInterceptor.preserveExisting =
true
// Step 2: Define the channel which is the memory, i.e., we can lost data in
case this agent crashes.
agent.channels.dynamicChannel.type = memory
agent.channels.dynamicChannel.capacity = 1000
// Step 3: Define the sink
agent.sinks.dynamicSink.type =
org.apache.flume.sink.elasticsearch.ElasticSearchSink
agent.sinks.dynamicSink.indexName = noserializer
agent.sinks.dynamicSink.indexType = logs
agent.sinks.dynamicSink.clusterName = elasticsearch
agent.sinks.dynamicSink.hostNames = localhost:9300
// Step 4: Bind the source and the sink to the channel
agent.sources.execSource.channels = dynamicChannel
agent.sinks.dynamicSink.channel = dynamicChannel
was (Author: skyahead):
Great point!
The default serializer is ElasticSearchLogStashEventSerializer, which uses the
following mapping. I am not sure if this is what you meant 'works well' :-). In
our case, we want our field names to be exactly the ones we set, which will be
used in our ElasticSearch queries.
{
"noserializer-2015-10-01":{
"mappings":{
"logs":{
"properties":{
"@fields":{
"properties":{
"timestamp":{
"type":"string"
}
}
},
"@message":{
"type":"string"
},
"@timestamp":{
"type":"date",
"format":"dateOptionalTime"
}
}
}
}
}
The above mapping is generated with this config file:
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# ./bin/flume-ng agent -n agent -c conf -f ./conf/dynamic.conf
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# Step 0:
#
# Define the agent.
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
agent.sources = execSource
agent.channels = dynamicChannel
agent.sinks = dynamicSink
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# Step 1:
#
# Define the exec source
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
agent.sources.execSource.type = exec
agent.sources.execSource.command = tail /tmp/some.log
agent.sources.execSource.interceptors = timestampInterceptor
agent.sources.execSource.interceptors.timestampInterceptor.type =
org.apache.flume.interceptor.TimestampInterceptor$Builder
agent.sources.execSource.interceptors.timestampInterceptor.preserveExisting =
true
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# Step 2:
#
# Define the channel which is the memory, i.e., we can lost data in case this
agent crashes.
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
agent.channels.dynamicChannel.type = memory
agent.channels.dynamicChannel.capacity = 1000
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# Step 3:
#
# Define the sink
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
agent.sinks.dynamicSink.type =
org.apache.flume.sink.elasticsearch.ElasticSearchSink
agent.sinks.dynamicSink.indexName = noserializer
agent.sinks.dynamicSink.indexType = logs
agent.sinks.dynamicSink.clusterName = elasticsearch
agent.sinks.dynamicSink.hostNames = localhost:9300
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
#
# Step 4:
#
# Bind the source and the sink to the channel
#
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
# # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # # #
agent.sources.execSource.channels = dynamicChannel
agent.sinks.dynamicSink.channel = dynamicChannel
> org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer does not
> serialize @timestamp correctly
> ----------------------------------------------------------------------------------------------------------
>
> Key: FLUME-2787
> URL: https://issues.apache.org/jira/browse/FLUME-2787
> Project: Flume
> Issue Type: Bug
> Components: Sinks+Sources
> Reporter: Eran W
> Priority: Minor
> Attachments: FLUME-2787.2.patch, FLUME-2787.patch
>
>
> When using
> agent.sinks.elastic-sink.serializer =
> org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
> the event timestamp is stored as string and not dateOptionalTime if the
> agent.sinks.elastic-sink.serializer is not set the code works as expected.
> REPRO:
> 1) use the following config
> agent.channels.channel-elastic.type = memory
> agent.channels.channel-elastic.capacity = 1000
> agent.channels.channel-elastic.transactionCapacity = 100
> # Define a source on agent and connect to channel.
> agent.sources.tail-source.type = exec
> agent.sources.tail-source.command = tail -4000 /home/cto/hs_err_pid11679.log
> agent.sources.tail-source.channels = channel-elastic
> #####INTERCEPTORS
> agent.sources.tail-source.interceptors = timestampInterceptor
> agent.sources.tail-source.interceptors.timestampInterceptor.type =
> org.apache.flume.interceptor.TimestampInterceptor$Builder
> agent.sources.tail-source.interceptors.timestampInterceptor.preserveExisting
> = true
> agent.sinks.elastic-sink.channel = channel-elastic
> agent.sinks.elastic-sink.type =
> org.apache.flume.sink.elasticsearch.ElasticSearchSink
> agent.sinks.elastic-sink.hostNames = 127.0.0.1:9300
> agent.sinks.elastic-sink.indexName = flume_index
> agent.sinks.elastic-sink.indexType = logs_type
> agent.sinks.elastic-sink.clusterName = elasticsearch
> agent.sinks.elastic-sink.batchSize = 10
> agent.sinks.elastic-sink.ttl = 5d
> agent.sinks.elastic-sink.serializer =
> org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer
> # Finally, activate.
> agent.channels = channel-elastic
> agent.sources = tail-source
> agent.sinks = elastic-sink
> 2) run it
> 3) look at the elastic index which was created:
> {
> "state": "open",
> "settings": {
> "index": {
> "creation_date": "1441728286466",
> "number_of_shards": "5",
> "number_of_replicas": "1",
> "version": {
> "created": "1070199"
> },
> "uuid": "9u-OCPxoQHWwURHyxh15lA"
> }
> },
> "mappings": {
> "logs_type": {
> "properties": {
> "body": {
> "type": "string"
> },
> "timestamp": {
> "type": "string"
> }
> }
> }
> },
> "aliases": [ ]
> }
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
