介绍
Flink Table API和SQL程序可以连接到其他外部系统，用于 读取 和 写入 批处理和流式处理 的 Table。
source 可以是 外部系统（如数据库、键值存储、消息队列或文件系统）中的数据的访问。
sink 可以是外部存储系统。
根据 source 和 sink 的类型，支持不同的格式，例如 CSV、Avro
官网：
https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/table/overview/
一句话解释
Connector 连接器可以像 建表 那样，连接外部系统
支持的 Connectors
| Name | Version | Source | Sink |
|---|---|---|---|
| Filesystem | 有界和无界扫描，查找 | Streaming Sink, Batch Sink | |
| Elasticsearch | 6.x & 7.x | 不支持 | Streaming Sink, Batch Sink |
| Apache Kafka | 0.10+ | 无界扫描 | Streaming Sink, Batch Sink |
| Amazon Kinesis Data Streams | 无界扫描 | Streaming Sink | |
| JDBC | 有界扫描，查找 | Streaming Sink, Batch Sink | |
| Apache HBase | 1.4.x & 2.2.x | 有界扫描，查找 | Streaming Sink, Batch Sink |
| Apache Hive | Supported Versions | 有界和无界扫描，查找 | Streaming Sink, Batch Sink |
注意
- kafka 只支持 append 方式，所以使用 table sql 时，不能执行 聚合函数
SQL connector
使用 SQL 语句来注册表。可以定义表名称、表结构以及用于连接到外部系统的表选项
例子
连接 kafka，详细参数解释见 链接
CREATE TABLE MyUserTable (
`user` BIGINT,
`message` STRING,
`rowtime` TIMESTAMP(3) METADATA FROM 'timestamp', -- use a metadata column to access Kafka's record timestamp
`proctime` AS PROCTIME(), -- use a computed column to define a proctime attribute
WATERMARK FOR `rowtime` AS `rowtime` - INTERVAL '5' SECOND -- use a WATERMARK statement to define a rowtime attribute
) WITH (
-- declare the external system to connect to
'connector' = 'kafka', -- 连接 kafka
'topic' = 'topic_name', -- 主题
'scan.startup.mode' = 'earliest-offset', -- 消费者启动模式
'properties.bootstrap.servers' = 'localhost:9092', -- 服务器ip、port
'format' = 'json' -- 序列化和反序列化格式
)
table api connector
tableEnviroment
.connect(...) //指定Table Connector Descriptor
.withFormat(...) //指定数据格式
.withSchema(...) //指定表结构
.inAppendMode() //指定更新模式
.registerTableSource("MyTable") //注册TableSource
更加见链接：https://blog.csdn.net/wwwzydcom/article/details/103749927