Flink：无法在类路径中为'org.apache.flink.table.factories.DeserializationSchemaFactory'找到合适的表工厂

Question

我正在使用flink的表api，我从kafka接收数据，然后将其注册为表，然后我使用sql语句进行处理，最后将结果转换回流，写入目录，代码如下所示：

def main(args: Array[String]): Unit = { 

    val sEnv = StreamExecutionEnvironment.getExecutionEnvironment 
    sEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 

    val tEnv = TableEnvironment.getTableEnvironment(sEnv) 

   tEnv.connect( 
      new Kafka() 
        .version("0.11") 
        .topic("user") 
        .startFromEarliest() 
        .property("zookeeper.connect", "") 
        .property("bootstrap.servers", "") 
    ) 
      .withFormat( 
        new Json() 
          .failOnMissingField(false) 
          .deriveSchema()   //使用表的 schema 
      ) 
      .withSchema( 
        new Schema() 
          .field("username_skey", Types.STRING) 
      ) 
      .inAppendMode() 
      .registerTableSource("user") 
     val userTest: Table = tEnv.sqlQuery( 
      """ 
       select ** form ** join **"".stripMargin) 
    val endStream = tEnv.toRetractStream[Row](userTest) 
    endStream.writeAsText("/tmp/sqlres",WriteMode.OVERWRITE) 
    sEnv.execute("Test_New_Sign_Student") 
 } 

我在本地测试中取得了成功，但是当我在集群中提交以下命令时，出现以下错误：

=======================================================

org.apache.flink.client.program.ProgramInvocationException: The main method 
caused an error. 
        at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:546) 
        at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:421) 
        at 
org.apache.flink.client.program.ClusterClient.run(ClusterClient.java:426) 
        at 
org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:804) 
        at 
org.apache.flink.client.cli.CliFrontend.runProgram(CliFrontend.java:280) 
        at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:215) 
        at 
org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:1044) 
        at 
org.apache.flink.client.cli.CliFrontend.lambda$main$11(CliFrontend.java:1120) 
        at java.security.AccessController.doPrivileged(Native Method) 
        at javax.security.auth.Subject.doAs(Subject.java:422) 
        at 
org.apache.hadoop.security.UserGroupInformation.doAs(UserGroupInformation.java:1692) 
        at 
org.apache.flink.runtime.security.HadoopSecurityContext.runSecured(HadoopSecurityContext.java:41) 
        at 
org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1120) 
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could 
not find a suitable table factory for 
'org.apache.flink.table.factories.DeserializationSchemaFactory' in 
the classpath. 

Reason: No factory implements 
'org.apache.flink.table.factories.DeserializationSchemaFactory'. 

The following properties are requested: 
connector.properties.0.key=zookeeper.connect 
.... 
schema.9.name=roles 
schema.9.type=VARCHAR 
update-mode=append 

The following factories have been considered: 
org.apache.flink.table.sources.CsvBatchTableSourceFactory 
org.apache.flink.table.sources.CsvAppendTableSourceFactory 
org.apache.flink.table.sinks.CsvBatchTableSinkFactory 
org.apache.flink.table.sinks.CsvAppendTableSinkFactory 
org.apache.flink.streaming.connectors.kafka.Kafka011TableSourceSinkFactory 


        at 
org.apache.flink.table.factories.TableFactoryService$.filterByFactoryClass(TableFactoryService.scala:176) 
        at 
org.apache.flink.table.factories.TableFactoryService$.findInternal(TableFactoryService.scala:125) 
        at 
org.apache.flink.table.factories.TableFactoryService$.find(TableFactoryService.scala:100) 
        at 
org.apache.flink.table.factories.TableFactoryService.find(TableFactoryService.scala) 
        at 
org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase.getDeserializationSchema(KafkaTableSourceSinkFactoryBase.java:259) 
        at 
org.apache.flink.streaming.connectors.kafka.KafkaTableSourceSinkFactoryBase.createStreamTableSource(KafkaTableSourceSinkFactoryBase.java:144) 
        at 
org.apache.flink.table.factories.TableFactoryUtil$.findAndCreateTableSource(TableFactoryUtil.scala:50) 
        at 
org.apache.flink.table.descriptors.ConnectTableDescriptor.registerTableSource(ConnectTableDescriptor.scala:44) 
        at 
org.clay.test.Test_New_Sign_Student$.main(Test_New_Sign_Student.scala:64) 
        at 
org.clay.test.Test_New_Sign_Student.main(Test_New_Sign_Student.scala) 
        at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
        at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 
        at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
        at java.lang.reflect.Method.invoke(Method.java:498) 
        at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:529) 

===================================

有人能告诉我是什么造成的吗？我对此很困惑........

Answer 1

您必须将您正在使用的连接器（Kafka）和格式（JSON）的JAR依赖项添加到程序的类路径中，即，构建包含它们的胖JAR或者将它们提供给Flink集群的类路径。将它们复制到./lib文件夹中。

检查Flink documentation以获取下载相应依赖项的链接。

Answer 2

我遇到了同样的问题，只需在运行应用程序时添加参数--connector.type kafka即可解决此问题。见enter link description here