你好,请问下我修改后的语句是 insert into print_sink select game_id,count(id) from mysql_source group by game_id 然后在执行的时候如果选择的是streamMode他会打印出Changelog,如下 2> +I(12,1) 5> +I(12555,1) 1> +I(122,1) 3> +I(13,1) 6> +I(1,1) 6> -U(1,1) 6> +U(1,2) 6> -U(1,2) 6> +U(1,3) 6> -U(1,3) 6> +U(1,4) 6> -U(1,4)
然后如果我使用的是batchMode,他就报错了 org.apache.flink.util.FlinkException: Error while shutting the TaskExecutor down. at org.apache.flink.runtime.taskexecutor.TaskExecutor.handleOnStopException(TaskExecutor.java:440) at org.apache.flink.runtime.taskexecutor.TaskExecutor.lambda$onStop$2(TaskExecutor.java:425) ... Caused by: java.util.concurrent.CompletionException: java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper at java.util.concurrent.CompletableFuture.encodeThrowable(CompletableFuture.java:273) ... at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) ... 21 more Suppressed: org.apache.flink.util.FlinkException: Could not properly shut down the TaskManager services. at org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:236) at org.apache.flink.runtime.taskexecutor.TaskExecutor.stopTaskExecutorServices(TaskExecutor.java:462) at... ... 21 more Caused by: org.apache.flink.util.FlinkException: Could not close resource. at org.apache.flink.util.AutoCloseableAsync.close(AutoCloseableAsync.java:42) at org.apache.flink.runtime.taskexecutor.TaskManagerServices.shutDown(TaskManagerServices.java:204) ... 37 more Caused by: java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper at org.apache.flink.runtime.memory.UnsafeMemoryBudget.reserveMemory(UnsafeMemoryBudget.java:94) at org.apache.flink.runtime.memory.UnsafeMemoryBudget.verifyEmpty(UnsafeMemoryBudget.java:64) ... at org.apache.flink.runtime.taskexecutor.TaskExecutor.onStop(TaskExecutor.java:422) ... 21 more [CIRCULAR REFERENCE:java.lang.NoClassDefFoundError: Could not initialize class org.apache.flink.util.JavaGcCleanerWrapper] 不知道您是否知道原因 在 2020-08-04 12:11:32，"godfrey he" <[email protected]> 写道： >逻辑上批产生的结果是Table，流产生的结果是Changelog。 >你在例子中Table的结果和changelog的结果是一样的，所以你感觉差不多。 >最简单的方式可以将query改为带group by的，再看结果的差异。 >更多关于Table和Changelog的概念可以参考 [1] > >[1] >https://ci.apache.org/projects/flink/flink-docs-release-1.11/zh/dev/table/streaming/dynamic_tables.html > >chenxuying <[email protected]> 于2020年8月4日周二 上午11:44写道： > >> hi : >> flink table sql 1.11.0 >> 在EnvironmentSettings中可以设置BatchMode或StreamingMode >> >> >> EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() >> // .inStreamingMode() >> .inBatchMode() >> .build(); >> >> >> 如果使用mysql作为source , 使用这两种模式都可以运行 , 效果都一样 , 感觉对批和流的理解还不够 , 不知道其中区别 , >> 不知道大佬们有没有例子可以比较容易理解 >> 我的代码 >> EnvironmentSettings environmentSettings = EnvironmentSettings.newInstance() >> // .inStreamingMode() >> .inBatchMode() >> .build(); >> TableEnvironment tableEnvironment = >> TableEnvironment.create(environmentSettings); >> tableEnvironment.executeSql("CREATE TABLE mysql_source ( " + >> " id bigint, " + >> " game_id varchar, " + >> " PRIMARY KEY (id) NOT ENFORCED " + >> " ) " + >> " with ( " + >> "'connector' = 'jdbc', " + >> " 'url' = 'jdbc:mysql://47.99.181.86:3306/flinksql?useSSL=false' , " + >> " 'username' = 'root' , " + >> " 'password' = 'root', " + >> " 'table-name' = 'mysqlsink' , " + >> " 'driver' = 'com.mysql.cj.jdbc.Driver' , " + >> " 'sink.buffer-flush.interval' = '2s', " + >> " 'sink.buffer-flush.max-rows' = '300' " + >> " )"); >> tableEnvironment.executeSql("CREATE TABLE print_sink ( " + >> " id bigint, " + >> " game_id varchar, " + >> " PRIMARY KEY (id) NOT ENFORCED " + >> " ) " + >> " with ( " + >> "'connector' = 'print' " + >> " )"); >> tableEnvironment.executeSql("insert into print_sink select id,game_id from >> mysql_source");
