@[TOC]
完整的报错
Exception in thread "main" org.apache.flink.table.api.ValidationException: Unable to create a sink for writing table 'default_catalog.default_database.sink_ella_operation_log'.
Table options are:
'connector'='elasticsearch-6'
'hosts'=' http://bigdatanode01:9200; http://bigdatanode02:9200; http://bigdatanode03:9200'
'index'='ella_operation_log'
at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:171)
at org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:367)
at org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:201)
at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:162)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
at scala.collection.AbstractTraversable.map(Traversable.scala:104)
at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:162)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1518)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:740)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:856)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:730)
at com.medbook.foreign.mysql.ella.SyncEllaOperationToES.sink(SyncEllaOperationToES.scala:65)
at com.medbook.foreign.mysql.ella.Exec$.main(Exec.scala:6)
at com.medbook.foreign.mysql.ella.Exec.main(Exec.scala)Caused by: org.apache.flink.table.api.ValidationException: One or more required options are missing.
Missing required options are:
document-type
at org.apache.flink.table.factories.FactoryUtil.validateFactoryOptions(FactoryUtil.java:381)
at org.apache.flink.table.factories.FactoryUtil.validateFactoryOptions(FactoryUtil.java:354)
at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.validate(FactoryUtil.java:712)
at org.apache.flink.streaming.connectors.elasticsearch.table.Elasticsearch6DynamicSinkFactory.createDynamicTableSink(Elasticsearch6DynamicSinkFactory.java:92)
at org.apache.flink.table.factories.FactoryUtil.createTableSink(FactoryUtil.java:168)
... 20 more
如下图所示提取有用的信息
分析错误原因
我是在flink-sql创建es表的时候报的错，报错提示缺少对应的options，及document-type
，我连忙去flink官方文档查找答案：https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/table/elasticsearch/#document-type
从官方文档给出的连接参数可以看出document-type在es6中需要指定，而在es7中无需指定，原因大概是6.0版本及以前一个索引里可以创建多个类型（type），7.0版本开始废弃type，一般使用_doc代替了， 8.0版本会彻底废弃。
我这里使用的是6.x版本的es，所以自然要配置好document-type这个option。
