[FLINK-15400] elasticsearch sink support dynamic index.

• Type: New Feature
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.11.0
• Fix Version/s: 1.11.0
• Component/s: Connectors / ElasticSearch, Table SQL / Ecosystem
• Labels: pull-request-available usability
• Assignee: Leonard Xu
• Reporter: ouyangwulin
• Created: 26/Dec/19 08:54
• Updated: 18/Apr/20 14:43
• Resolved: 18/Apr/20 14:43

Description

From [e-mail](https://lists.apache.org/thread.html/ac4e0c068baeb3b070f0213a2e1314e6552b226b8132a4c49d667ecd%40%3Cuser-zh.flink.apache.org%3E),  Becuase the es 6/7 not support ttl. so User need clean the index by timestamp. Add dynamic index is a useful function.  Add with properties 'dynamicIndex' as a switch for open dynamicIndex. Add with  properties 'indexField'  for the extract time field, Add properties 'indexInterval' for change cycle mode.

 

With property	discribe	default	Required
dynamicIndex	Dynamic or not	false(true/false)	false
indexField	extract index field	none	dynamicIndex is true , then indexField is required,only supported type "timestamp","date","long"
indexInterval	mode for  cycle	d	ddynamicIndex is true , this field is required ，the value optional is： d：day m：mouth w：week

 

After discussion, the final design looks as following :

CREATE TABLE es_sink_table (
  log_source varchar ,
  log_content varchar ,
  log_level bigint ,
  log_ts timestamp,
) WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '6',
'connector.index'='my-log-{log_ts|yyyy-MM-dd}'， 
# elasticsearch index name, Flink support create index based on field at 
# runtime dynamically, the index value comes from the dynamicIndex 
# pattern when the field type is varchar, eg:'my-log-{log_source}',the # dynamicIndex pattern support format and parse date by Java 
# SimpleDataFormat when the field type is timestamp/date, 
# eg:'my-log-{log_ts|yyyy-MM-dd}'.
'connector.index-alias'='my-log'，       
# index alias name, the alias name mapping to all indies that 
# creatd from 'connector.index'. 
…
)

 

 

 

Issue Links

• Is contained by: FLINK-16146 Improve end-to-end usability of Flink Table API & SQL
• is related to: FLINK-16701 Elasticsearch sink support alias for indices.
• links to: GitHub Pull Request #11466