问题一：flink-1.11使用executeSql()执行DDL语句问题
环境信息：flink-1.11.0, blink-planner, 本地ide开发测试（IDEA）
问题描述：使用executeSql()方法执行DDL语句，控制台打印如下异常信息。（flink-connector-kafka_2.11依赖已添加）
我不确定是否还有某个必要的依赖没有添加，还是有其他的地方没有考虑完整，请大佬赐教。
--------------------------------------------------------------分割线-------------------------------------------------------------
Exception in thread "main" org.apache.flink.table.api.ValidationException: Unable to create a source for reading table 'default_catalog.default_database.kafka_out'.
Table options are:
'connector'='kafka'
'format'='json'
'properties.bootstrap.servers'='localhost:9092'
'properties.group.id'='flink-1.11'
'scan.startup.mode'='group-offsets'
'topic'='flink-kafka'
at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:125)
at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.buildTableScan(CatalogSourceTable.scala:135)
at org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.scala:78)
at org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3492)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2415)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2102)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2051)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:661)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:642)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3345)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2178)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2051)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:661)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:642)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3345)
at org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:568)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:164)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:151)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:773)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:745)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:238)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:527)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:204)
at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:678)
at example.Example.main(Example.java:77)
Caused by: org.apache.flink.table.api.ValidationException: Could not find any factories that implement 'org.apache.flink.table.factories.DeserializationFormatFactory' in the classpath.
at org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:229)
at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverOptionalFormatFactory(FactoryUtil.java:538)
at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverOptionalDecodingFormat(FactoryUtil.java:426)
at org.apache.flink.table.factories.FactoryUtil$TableFactoryHelper.discoverDecodingFormat(FactoryUtil.java:413)
at org.apache.flink.streaming.connectors.kafka.table.KafkaDynamicTableFactoryBase.createDynamicTableSource(KafkaDynamicTableFactoryBase.java:73)
at org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:122)
... 25 more
*来自志愿者整理的flink邮件归档
参考答案：
看起来是format找不到实现，你可以添加一下flink-json的依赖看一下。
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370199?spm=a2c6h.12873639.article-detail.88.6f9243783Lv0fl
问题二：flink-benchmarks使用求助
如题，最近在新机器上跑flink-benchmarks验证下机器性能，但是不太会对跑出的结果进行分析，不知是否有大神也用过这个，可否指点一二
*来自志愿者整理的flink邮件归档
参考答案：
你说的 flink-benchmarks 是指 这个仓库[1]的代码吗？ 是这个仓库的代码的话，你按照 readme 能跑出一个结果（csv
文件，或者终端能看到最终的结果），这个结果就是 JMH 的的结果，具体的可以阅读 JMH 的相关文档[2]
[1] https://github.com/dataArtisans/flink-benchmarks
[2] http://openjdk.java.net/projects/code-tools/jmh/
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370198?spm=a2c6h.12873639.article-detail.89.6f9243783Lv0fl
问题三：代码中如何取消正在运行的Flink Streaming作业
这个问题之前看到过有人在问，但是没有看到答案，我想问一下，Flink Streaming API有没有提供类似的接口，调用后就能停止这个Stream作业呢？
*来自志愿者整理的flink邮件归档
参考答案：
可以通过 StreamExecutionEnvironment#executeAsync 提交作业，返回 JobClient [1], 通过
JobClient 可以 cancel 作业，获取 job status。
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-74%3A+Flink+JobClient+API
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370179?spm=a2c6h.12873639.article-detail.90.6f9243783Lv0fl
问题四：State 0点清除的问题
大家好： 我想问下，在ProcessAllWindowFunction中，在每天的0点清除state如何清除？
*来自志愿者整理的flink邮件归档
参考答案：
如果你需要精确的控制每天 0 点清除 state 的话，或许你可以考虑使用 processFunction[1], 然后自己使用 timer
实现相关逻辑
[1]
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370176?spm=a2c6h.12873639.article-detail.91.6f9243783Lv0fl
问题五：flink1.9状态及作业迁移
flink1.9的检查点或保存点中会保留hadoop集群的nameservice数据吗？现在想将一个集群的flink作业迁移到另一个集群，但两个集群的nameservice名称不一样，迁移会有问题吗？如果有问题的话对应状态保存的nameservice可以修改吗？或者说迁移的时候还有哪些其他需要注意的问题？
*来自志愿者整理的flink邮件归档
参考答案：
Checkpoint meta中存储的是完整路径，所以一般会把hdfs的namespace存储起来，导致没办法直接迁移。
Flink-1.11 支持将savepoint（但是不支持Checkpoint）进行位置迁移 [1]，而对于Flink-1.9，二者均不支持。
[1] https://issues.apache.org/jira/browse/FLINK-5763
祝好
*来自志愿者整理的flink邮件归档
关于本问题的更多回答可点击进行查看：
https://developer.aliyun.com/ask/370175?spm=a2c6h.12873639.article-detail.92.6f9243783Lv0fl
