各位好，写了个demo，代码如下，在本地跑没有问题，提交到yarn session上报错： Caused by: org.apache.flink.table.api.ValidationException: Could not find any factory for identifier 'kafka' that implements 'org.apache.flink.table.factories.DynamicTableSourceFactory' in the classpath. 请问是什么原因导致的呢？

代码如下：

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); StreamTableEnvironment tenv = StreamTableEnvironment.create(env, settings);

tenv.executeSql("CREATE TABLE test_table (\n" + " id INT,\n" + " name STRING,\n" + " age INT,\n" + " create_at TIMESTAMP(3)\n" + ") WITH (\n" + " 'connector' = 'kafka',\n" + " 'topic' = 'test_json',\n" + " 'properties.bootstrap.servers' = 'localhost:9092',\n" + " 'properties.group.id' = 'testGroup',\n" + " 'format' = 'json',\n" + " 'scan.startup.mode' = 'latest-offset'\n" + ")"); Table table = tenv.sqlQuery("select * from test_table"); tenv.toRetractStream(table, Row.class).print(); env.execute("flink 1.11.0 demo");

pom 文件如下：

<scala.binary.version>2.11</scala.binary.version> <flink.version>1.11.0</flink.version> org.apache.flink

flink-table-planner-blink_${scala.binary.version} ${flink.version} org.apache.flink

flink-table-runtime-blink_${scala.binary.version} ${flink.version} org.apache.flink flink-json ${flink.version} org.apache.flink

flink-connector-kafka_${scala.binary.version} ${flink.version} org.apache.flink

flink-connector-kafka-0.11_${scala.binary.version} ${flink.version} org.apache.flink

flink-connector-kafka_${scala.binary.version} ${flink.version} org.apache.flink flink-core ${flink.version} org.apache.flink flink-clients_${scala.binary.version} ${flink.version} org.apache.flink flink-table-common ${flink.version}

*来自志愿者整理的flink邮件归档