User-defined Sources & sinks     Dynamic tables 是Flink Table & SQL API的核心概念，对于处理有界与无界数据采用了统一的方式。     Dynamic tables 是一个逻辑概念，Flink自己不拥有数据。相反，dynamic table是被存储在外部系统（databases，key-value，消息队列）或者文件     Dynamic sources 从外部读取数据，和 Dynamic sink 被用于写数据到外部系统。在这个文档中，sources 和 sinks被归为connector.     Flink 提供了预定义的 connector，如 kafka ,hive 与不同的文件系统。     这章节，聚焦如何开发一个通用的，用户自定义的connector     *** 新的 table Sources 和 table sink接口已经介绍在Flink1.11 part of FLIP-95。Factory 接口已经重新构建了。如果必要，查看 old table sources 和 sinks页。         这些接口后向支持兼容

Overview 在很多案例中，实现者不需要创建一个新的connector从scratch，而可能是去显示的修改一个存在的connector或者hook一个存在stack中。在其他的案例中，实现者可能去创建一个特定的connector。 这个部分帮助解决这两种案例。解析了通用的table connector的结构，从 api的定义到 集群中执行的代码 这个箭头展示了对象如何转换成其他对象，从一个stage到下一个stage在转换处理期间

Metadata      Table API 与 SQL是定义的API。Metadata包含声明的tables。因此，执行 create table语法时，将导致目标catalog上的metadate更新。     对于大多catalog实现来说，在外部系统的物理数据不会因为这些操作进行修改。Connector-specific依赖也没有出现在classpath中。定义在with语法中的，即不进行验证，也不进行解释     dynamic tables的metadata(通过DDL创建或者被catalog提供的)代表 catalogTable实例。表名将被解析进catalogTable内部当必要的时候。

Planning     当进入到planning和优化table程序时，catalogTable需要被解析成 DynamicTableSource(查询时,select )和DynamicTableSink(写入时，insert into)     DynamicTableSourceFactory 和 DynamicTableSinkFactory 提供 connector-specific逻辑，把catalogTable的metadata转换成 DynamicTableSource和DynamicTableSink。     在多数这类案例中，factory的主要功能是去验证 options(如 'port'='5022')，配置编码与解码格式(如果需要)，创建参数化的table connector实例。     默认，DynamicTableSourceFactory 和 DynamicTableSinkFactory是通过Java的SPI去发现的。Connector的 option('connector'='custom')必须满足factory的验证符     虽然可能不出现在类名中，DynamicTableSource和DynamicTableSink也能够被看成有状态的factorise，最终产生真实的运行环境来读取与写入数据     Planner使用source和sink实例去执行connetor-specific 双向连接直到一个理想的逻辑计划被发现。依靠可选的声明的接口（SupportsProjectPushDown or SupportsOverwrite）,     planner可能对实例进行修改，变动这个生成的运行环境的实现      Runtime