问题一：Flink catalog的几个疑问
Hi Flink社区：
有几个疑问希望社区小伙伴们帮忙解答一下：
1.个人感觉Flink很有必要提供一个官方的catalog，用来支持各种connector，比如：kafka，jdbc，hbase等等connector。不知道社区有没有这个打算，目前没有看到对应的flip
2.社区对hive catalog的定位是什么，后续有可能转正为flink 默认的catalog实现吗？
3.hive catalog是不支持大小写敏感的（字段名都是小写），这个后续会带来哪些问题？想征集下大家的意见避免我们以后踩大坑。
*来自志愿者整理的flink邮件归档
参考答案：
- Flink 提供了一套catalog的接口，提提供了几个内置的实现：in-memory catalog, hive catalog,
postgres catalog,
可以根据自己的需求选择。也可以实现自定义的catalog。参考 [1]
- hive catalog 主要是对接 hive，方便读取现有的hive catalog的meta信息。当然也可以往hive
catalog写新的meta。
是否会转为默认catalog，据我所知，目前没有。
- 一般没什么问题。在和其他区分大小写的db对接的时候，可能有问题。
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370308?spm=a2c6h.12873639.article-detail.42.29d04378ApxdqJ
问题二：flink-1.11 ddl kafka-to-hive问题
hive-1.2.1
chk 已经成功了（去chk目录查看了的确有chk数据，kafka也有数据），但是hive表没有数据，我是哪里缺少了什么吗？
String hiveSql = "CREATE TABLE stream_tmp.fs_table (\n" +
" host STRING,\n" +
" url STRING," +
" public_date STRING" +
") partitioned by (public_date string) " +
"stored as PARQUET " +
"TBLPROPERTIES (\n" +
" 'sink.partition-commit.delay'='0 s',\n" +
" 'sink.partition-commit.trigger'='partition-time',\n" +
" 'sink.partition-commit.policy.kind'='metastore,success-file'" +
")";
tableEnv.executeSql(hiveSql);
tableEnv.executeSql("INSERT INTO stream_tmp.fs_table SELECT host, url, DATE_FORMAT(public_date, 'yyyy-MM-dd') FROM stream_tmp.source_table");
*来自志愿者整理的flink邮件归档
参考答案：
rolling 策略配一下？ https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/filesystem.html#sink-rolling-policy-rollover-interval
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370306?spm=a2c6h.12873639.article-detail.43.29d04378ApxdqJ
问题三：FlinkKafkaConsumer API 维表关联
各位好：
请教一下，用FlinkKafkaConsumer API的话，如何支持SQL的方式，和维表关联。（之前用Kafka API&StreamTableDescriptor API是可以的） "select a.id,b.name from kafka_table a " + "join dim_table FOR SYSTEM_TIME AS OF a.proctime as b on a.id = b.user_id";
*来自志愿者整理的flink邮件归档
参考答案：
```你需要用 DDL 去声明这张 kafka 表[1], 目前不建议使用 Kafka 和 StreamTableDescriptor API。[1]:
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/kafka.html
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370305?spm=a2c6h.12873639.article-detail.44.29d04378ApxdqJ
问题四：Flink sql中可以使用自定义窗口触发器吗
rt
*来自志愿者整理的flink邮件归档
参考答案：
目前是不支持的。不过有个实验性功能可以指定提前输出的策略和迟到处理的策略 [1]，可能可以满足你的需求。
[1]:
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370304?spm=a2c6h.12873639.article-detail.45.29d04378ApxdqJ
问题五：flink 1.11 cdc: kafka中存了canal-json格式的多张表信息，需要按表解析做
mysql表： CREATE TABLE test ( id int(11) NOT NULL, name varchar(255) NOT NULL, time datetime NOT NULL, status int(11) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8
CREATE TABLE status ( id int(11) NOT NULL, name varchar(255) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8
kafka中数据： // 表test 中insert事件 {"data":[{"id":"1745","name":"jindy1745","time":"2020-07-03 18:04:22","status":"0"}],"database":"ai_audio_lyric_task","es":1594968168000,"id":42,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)","time":"datetime","status":"int(11)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12,"time":93,"status":4},"table":"test","ts":1594968168789,"type":"INSERT"}
//表status 中的事件 {"data":[{"id":"10","name":"status"}],"database":"ai_audio_lyric_task","es":1595305259000,"id":589240,"isDdl":false,"mysqlType":{"id":"int(11)","name":"varchar(255)"},"old":null,"pkNames":["id"],"sql":"","sqlType":{"id":4,"name":12},"table":"status","ts":1595305259386,"type":"INSERT"}
如何由于kafka中的json动态的变化的，比如新增一个表，如何能转成应对的RowData， 感觉无法直接用JsonRowDeserializationSchema或CanalJsonDeserializationSchema来做处理。
*来自志愿者整理的flink邮件归档
参考答案：
这个邮件里提到了类似的问题。
https://issues.apache.org/jira/browse/FLINK-18002 这个issue完成后(1.12)，你可以将
“data”，“mysqlType”等格式不确定的字段定义为String类型，
下游通过udf自己再解析对应的json*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370303?spm=a2c6h.12873639.article-detail.46.29d04378ApxdqJ
