问题一：flink 1.11运算结果存mysql出错
flink 1.11运算结果存mysql出错
参考回答：
简单看了下代码应该没啥问题，alarm_test_g 这个kafka topic里有数据吗？可以检查下是否有脏数据，直接用./bin/kafka-console-consumer.sh 检查下？我有点怀疑这点
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372155
问题二：如何在sql运行时动态修改kafka的scan.startup.mode
现在有个需求，就是一段用sql-client提交的sql任务需要动态修改kafka的scan.startup.mode，以支持不同的消费需求。请问有什么好的办法吗？
参考回答：
可以尝试下1.11中引入的LIKE[1]或者是Table Hint[2]
[1]
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/create.html#create-table
[2]
https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sql/hints.html
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372157
问题三：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
参考回答：
看起来是format找不到实现，你可以添加一下flink-json的依赖看一下。
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372158
问题四：flink-benchmarks使用求助
如题，最近在新机器上跑flink-benchmarks验证下机器性能，但是不太会对跑出的结果进行分析，不知是否有大神也用过这个，可否指点一二
参考回答：
你说的 flink-benchmarks 是指 这个仓库[1]的代码吗？ 是这个仓库的代码的话，你按照 readme 能跑出一个结果（csv
文件，或者终端能看到最终的结果），这个结果就是 JMH 的的结果，具体的可以阅读 JMH 的相关文档[2]
[1] https://github.com/dataArtisans/flink-benchmarks
[2] http://openjdk.java.net/projects/code-tools/jmh/
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372160
问题五：代码中如何取消正在运行的Flink Streaming作业
这个问题之前看到过有人在问，但是没有看到答案，我想问一下，Flink Streaming API有没有提供类似的接口，调用后就能停止这个Stream作业呢？
参考回答：
可以通过 StreamExecutionEnvironment#executeAsync 提交作业，返回 JobClient [1], 通过
JobClient 可以 cancel 作业，获取 job status。
[1]
https://cwiki.apache.org/confluence/display/FLINK/FLIP-74%3A+Flink+JobClient+API
关于本问题的更多回答可点击原文查看：https://developer.aliyun.com/ask/372162
