[
https://issues.apache.org/jira/browse/FLUME-2787?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14940488#comment-14940488
]
Tianji Li commented on FLUME-2787:
----------------------------------
We experienced similar issues in our work and figured out a way to fix it.
In essence, ElasticSearch is not really schema less, because the underlying
Lucene requires strict schemas. The schema-less of ElasticSearch is achieved by
it guessing what data format should be used when data is coming in. But the
guessing work sometimes fails, and fails silently, leaving people pulling lots
of hair off to understand.
To resolve this guessing work therefore is to explicitly tell ElasticSearch
what your schema is, before sending any data in.
There are two ways to achieve this.
First, using the index templates provided by ElasticSearch
(https://www.elastic.co/guide/en/elasticsearch/reference/current/indices-templates.html).
The idea of these templates is that one can set mappings (mapping means schema
in the ElasticSearch terms) for many indices before using them. This is not a
good option when using Flume and ElasticSearch at the same time. The reason is
Flume automately rotates the indices on a daily basis (which is a good thing
when indices are huge). To use the templates, one has to keep the rotation in
mind al the time, and change the templates at good timing.
Second, we prefer therefore to let Flume set the mappings/schemas when needed.
Again due to Flume's index rotation, setting mappings should be done in
EventSerializers. I implemented this as in the attached patch.
Now, let me give the results.
If not using agent.sinks.elastic-sink.serializer =
org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer, as
mentioned by [~goi.cto] above, depending on version of ElasticSearch used, the
actual mapping can vary. For example, using ElasticSearch 1.7.1, the latest
version, the field names I got are not 'body' and 'timestamp', but are shown
below (Please check the details in the noserializer file in the patch.):
{
"noserializer-2015-10-01": {
"mappings": {
"logs": {
"properties": {
"@fields": {
"properties": {
"timestamp": {
"type": "string"
}
}
},
"@message": {
"type": "string"
},
"@timestamp": {
"type": "date",
"format": "dateOptionalTime"
}
}
}
}
}
}
If using agent.sinks.elastic-sink.serializer =
org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer, again as
[~goi.cto] said, the 'body' field is a string, rather than a 'date_time'.
Please check the details in the dynamic.conf file in the patch.
{
"dynamic-2015-10-01": {
"mappings": {
"logs": {
"properties": {
"body": {
"type": "string"
},
"timestamp": {
"type": "string"
}
}
}
}
}
}
If using the manualSerializer I implemented, the mapping will be below. See the
manual.conf in the patch for how to use it.
{
"manual-2015-10-01": {
"mappings": {
"logs": {
"_timestamp": {
"enabled": true,
"store": true,
"path": "timestamp",
"format": "date_time",
"default": "1970-01-01T00:00:00.000Z"
},
"_index": {
"enabled": true
},
"properties": {
"body": {
"type": "string",
"index": "not_analyzed",
"doc_values": true
},
"timestamp": {
"type": "date",
"format": "dateOptionalTime"
}
}
}
}
}
}
In our work, and in many other use cases, manual mapping is needed on string
field such as people full names. Say we index FirstName and LastName in a same
field, and do a terms aggregation which supports to get distinct full names. If
we use the org.apache.flume.sink.elasticsearch.ElasticSearchDynamicSerializer,
the field will be 'analyzed' by default which means ElasticSearch will index
firstName and lastName separately. For example, if we index:
New York
New York
New York
New Mexico
New Jersey
in a field that is analyzed, then this term aggregation
{
"size" : 0,
"aggs" : {
"bodyTerms" : {
"terms" : {
"field" : "body"
}
}
}
}
will give us:
{
"took": 1,
"timed_out": false,
"_shards": {
"total": 5,
"successful": 5,
"failed": 0
},
"hits": {
"total": 5,
"max_score": 0,
"hits": []
},
"aggregations": {
"bodyTerms": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "new",
"doc_count": 5
},
{
"key": "york",
"doc_count": 3
},
{
"key": "jersey",
"doc_count": 1
},
{
"key": "mexico",
"doc_count": 1
}
]
}
}
}
To resolve this, we have to change the field to be 'not_analyzed' which will
give us (using the
org.apache.flume.sink.elasticsearch.ElasticSearchManualSerializer in the patch):
{
"took": 2,
"timed_out": false,
"_shards": {
"total": 5,
"successful": 5,
"failed": 0
},
"hits": {
"total": 5,
"max_score": 0,
"hits": []
},
"aggregations": {
"bodyTerms": {
"doc_count_error_upper_bound": 0,
"sum_other_doc_count": 0,
"buckets": [
{
"key": "New York",
"doc_count": 3
},
{
"key": "New Jersey",
"doc_count": 1
},
{
"key": "New Mexico",
"doc_count": 1
}
]
}
}
}
> 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.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)
