当前位置：首页 > news >正文

flink1.19.2+cdc-3.2.1遇到的问题及解决方案

news 来源：原创 2025/6/14 12:03:28

No operators defined in streaming topology. Cannot generate StreamGraph

原因是博主想通过在flinkjar中写flinksql，以jar的方式提交application模式的作业(因为无法直接使用yarn-application提交flinksql)。
代码中有create和insert intosql，但是运行时报上面错误，各方面查找资料，其原因是flink没有识别insert操作，解决方案就是显示提交insert
错误代码

public static void execute(String content, TableEnvironment tEnv) {List<String> insertSql = new ArrayList<>();// 分割SQL语句Arrays.stream(content.split(";")).map(String::trim).filter(s -> !s.isEmpty()).forEach(sql -> {sql = sql.replaceAll(SEMICOLON, ";").replaceAll(SY, "\"").replaceAll(SDY, "`");if (!sql.endsWith(";")) {sql = sql + ";"; // 加上分号}System.out.println("执行SQL>>>" + sql);tEnv.executeSql(sql);});
}

修改后

public static void execute(String content, TableEnvironment tEnv) {List<String> insertSql = new ArrayList<>();// 分割SQL语句Arrays.stream(content.split(";")).map(String::trim).filter(s -> !s.isEmpty()).forEach(sql -> {if (!sql.endsWith(";")) {sql = sql + ";"; // 加上分号}if (sql.toUpperCase().startsWith("INSERT")) {insertSql.add(sql);} else {System.out.println("执行SQL>>>" + sql);tEnv.executeSql(sql);}});// 统一执行insert// 显式触发, 否则会报错The main method caused an error: No operators defined in streaming topology. Cannot execute.if (insertSql.size() > 0) {StatementSet stmtSet = tEnv.createStatementSet();for (String insert : insertSql) {// 添加多个 INSERT 语句System.out.println("执行SQL>>>" + insert);stmtSet.addInsertSql(insert);}try {// 执行并设置作业名称TableResult result = stmtSet.execute();JobClient jobClient = result.getJobClient().orElseThrow(() -> new RuntimeException("无法获取JobClient"));System.out.println("等待作业启动...");// 等待作业进入运行状态JobStatus status = jobClient.getJobStatus().get(STATUS_TIMEOUT, TimeUnit.SECONDS);if (status == JobStatus.RUNNING) {System.out.println("作业已正常运行");// 添加2秒延迟确保状态同步Thread.sleep(2000);} else {throw new IllegalStateException("作业启动失败，状态: " + status);}} catch (InterruptedException | ExecutionException e) {e.printStackTrace();throw new RuntimeException(e);} catch (TimeoutException e) {throw new RuntimeException(e);}}}

ERROR org.apache.flink.shaded.curator5.org.apache.curator.ConnectionState - Authentication failed

在flink的conf.yml添加

zookeeper:sasl:disable: true

hadoop日志里中文显示成了问号

在flink的conf.yml添加

env:java:opts:all: -Dfile.encoding=UTF-8

Invalid event: APP_UPDATE_SAVED at ACCEPTED

flink/lib下添加flink-shaded-hadoop-3-uber-3.1.1.7.2.9.0-173-9.0.jar

java.lang.NoSuchMethodError: org.apache.commons.cli.Option.builder(Ljava/lang/String;)Lorg/apache/commons/cli/Option$Builder;

flink/lib下添加commons-cli-1.5.0.jar

yarn启动容器使用随机端口的问题

需要节点间端口互通，否则经常会因为端口不通而导致容器启动不起来
#假设要允许192.168.0.66和192.168.0.67访问本机所有端口

sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.0.66" accept'
sudo firewall-cmd --permanent --add-rich-rule='rule family="ipv4" source address="192.168.0.67" accept'
sudo firewall-cmd --reload