New ElasticSearch()…xxx...registerTableSink();
ElasticSearch是一个ConnectorDescriptor: 最关键的就是一个toConnectorProperties就是一个配置的map,传给factory使用，他其实是一个配置收集器.
TableFactoryUtils : 生成一个TableFactory的工具类，
findAndCreateTableSink( ) 通过一个描述来生成一个工厂类.//描述里有必要的配置.
先找到合适的工厂类，然后创建一个TableSink.
第一步，A先找到合适的工厂类，通过properties里的配置来寻找,
TableFactoryService.find(TableSinkFactory.class, properties)
findSingleInternal()里的两个代码是关键代码
A.1 : List<TableFactory> tableFactories = discoverFactories(classLoader); 通过classLoader(可以传入, 否则使用默认)找到所有的Factory.
A.2: 然后开始过滤， List<T> filtered = filter(tableFactories, factoryClass, properties)
A.2.1 : 首先检查类型, TableSinkFactory.class filterByFactoryClass(factoryClass,properties,foundFactories)
A.2.2 : 然后检查必须的配置是不是设置了, List<T> contextFactories = filterByContext(factoryClass,properties,foundFactories,classFactories);
其中必备的配置是在factory里配置和定义的，必须不能为空.这个是有TableFactory的requiredContext()方法定义的.
A.2.3 : 经过以上两轮的过滤，然后再继续检查，依然是TableFactoryService.java，这里面的
private static <T extends TableFactory> List<T> filterBySupportedProperties(
Class<T> factoryClass,
Map<String, String> properties,
List<TableFactory> foundFactories,
List<T> classFactories) 方法
这里主要是TableFactory里的两个方法定义的属性
Map<String, String> requiredContext();
List<String> supportedProperties();
然后和用户Descriptor（如 ElasticSearch）里设置的属性匹配，其中有字符忽略大小写，星号匹配等细节，不再研究.
然后这就找到了工厂类，接下来要生成TableSink了。例如 ElasticSearch找到的就是Elasticsearch6UpsertTableSinkFactory.
第二步，生成TableSink.在ElasticSearch中，他是直接的new了一个Elasticsearch6UpsertTableSink.而这个对象又是一个StreamTableSink，这样就生成了一个StreamTableSink, 这个sink可以交给flink直接使用了。在ElasticSearch这个TableSink里，最关键的是实现了consumeDataStream，这个决定了sink的写逻辑.注意区分的是一个是StreamTableSink是给Table接口使用的，另一个是DataStreamSink是给DataStream使用的。这里是连接Table和Stream接口的地方。所以说，Table的API其实还是基于Stream的API的。
生成了Elasticsearch的TableSink之后，可以看到到这个类的用途是为了向es写入数据，他的作用是生成一个createSinkFunction。
createSinkFunction是具体的连接es，向es写入数据的逻辑，具体的实现是ElasticsearchUpsertSinkFunction的process的方法。是由ElasticsearchSinkBase发起的调用。
在写入ElasticSearch的时候，如何让_id使用数据里的指定的field的值？
在ES2.0之后，mapping里设置_id就被移除了这个特性。目的是为了让业务方不需要关心es的实现细节。在es connector里，他会根据你的table的primary key来生成这个_id的值。一般情况下，在append模式下，因为只是insert，所以他都是随机的。只有在update的时候，才需要根据primary key查找到元素并且进行更新。具体的实现在processUpsert方法。
private void processUpsert(Row row, RequestIndexer indexer) {
final byte[] document = serializationSchema.serialize(row);
if (keyFieldIndices.length == 0) {
final IndexRequest indexRequest = requestFactory.createIndexRequest(
index,
docType,
contentType,
document);
indexer.add(indexRequest);
} else {
final String key = createKey(row);
final UpdateRequest updateRequest = requestFactory.createUpdateRequest(
index,
docType,
key,
contentType,
document);
indexer.add(updateRequest);
}
}
这个方法的关键是keyFieldIndices这个属性。如果这个属性没值，那么久不指定key，让他自动生成_id，否则就从元数据里拿到这个字段值，再拼接成_id.而这个值是由方法
public void setKeyFields(String[] keyNames) {} 来设置的。这个就是tablesink接口里定义的方法。这个方法可不是给业务方调用的，这个是flink框架调用的。所以你显示的调用这个方法也没用，他flink最终会覆盖掉你的配置。
/**
* Configures the unique key fields of the {@linkTable} to write.
* The method is called after {@linkTableSink#configure(String[], TypeInformation[])}.
*
*<p>The keys array might be empty, if the table consists of a single (updated) record.
* If the table does not have a key and is append-only, the keys attribute is null.
*
*@paramkeysthe field names of the table's keys, an empty array if the table has a single
* row, and null if the table is append-only and has no key.
*/
voidsetKeyFields(String[] keys);
说明在这里。说的很清楚了，如果是append-only的模式，传入的keys就为空，否则就用table 的 primary key. 所以你可以定义一个table，指定主键。
找了一圈发现并没有定义table 的 primary的语法，翻看flink的源码发现他是自动判断的，具体的是根据你的sql语句来判断的。
UpdatingPlanChecker.scala是一个scala的不太好看明白，具体的在.private class UniqueKeyExtractor {} 这个类里面。他的寻找方式是，如果你是group by 语句，那么group by 就是你的key.如果你是join，那就看你左边的表和你右边的表来分别递归计算。
// Output of join must have keys if left and right both contain key(s).
// Key groups from both side will be merged by join equi-predicates
val lInNames: Seq[String] = j.getLeft.getRowType.getFieldNames
val rInNames: Seq[String] = j.getRight.getRowType.getFieldNames
val joinNames = j.getRowType.getFieldNames
// if right field names equal to left field names, calcite will rename right
// field names. For example, T1(pk, a) join T2(pk, b), calcite will rename T2(pk, b)
总之就是，只有在聚合计算的时候才会有key, 也就是group by 后面的内容。那这不就扯了吗？ ElasticSearch connector 不就只支持插入操作了吗？
Ps： flink后续会支持在table schema里定义primary key, 已经提交了pr，https://github.com/apache/flink/pull/8736 可惜现在还没有。pat ~