報錯如下:

Exception in thread "main" org.apache.flink.table.api.TableException: findAndCreateTableSink failed.
	at org.apache.flink.table.factories.TableFactoryUtil.findAndCreateTableSink(TableFactoryUtil.java:92)
	at org.apache.flink.table.factories.TableFactoryUtil.lambda$findAndCreateTableSink$0(TableFactoryUtil.java:117)
	at java.util.Optional.orElseGet(Optional.java:267)
	at org.apache.flink.table.factories.TableFactoryUtil.findAndCreateTableSink(TableFactoryUtil.java:117)
	at org.apache.flink.table.planner.delegation.PlannerBase.getTableSink(PlannerBase.scala:341)
	at org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:204)
	at org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translate$1(PlannerBase.scala:163)
	at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
	at scala.collection.Iterator.foreach(Iterator.scala:937)
	at scala.collection.Iterator.foreach$(Iterator.scala:937)
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
	at scala.collection.IterableLike.foreach(IterableLike.scala:70)
	at scala.collection.IterableLike.foreach$(IterableLike.scala:69)
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
	at scala.collection.TraversableLike.map(TraversableLike.scala:233)
	at scala.collection.TraversableLike.map$(TraversableLike.scala:226)
	at scala.collection.AbstractTraversable.map(Traversable.scala:104)
	at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:163)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1264)
	at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:700)
	at org.apache.flink.table.api.internal.TableImpl.executeInsert(TableImpl.java:565)
	at org.apache.flink.table.api.internal.TableImpl.executeInsert(TableImpl.java:549)
	at InsertInto.main(InsertInto.java:53)
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.TableSinkFactory' in
the classpath.

Reason: Required context properties mismatch.

The following properties are requested:
connector.password=appleyuchi
connector.table=table_api
connector.type=jdbc
connector.url=jdbc:mysql://Desktop:3306/flink
connector.username=appleyuchi
connector.write.flush.max-rows=5000
schema.0.data-type=VARCHAR(2147483647)
schema.0.name=user
schema.1.data-type=VARCHAR(2147483647)
schema.1.name=product
schema.2.data-type=BIGINT
schema.2.name=amount

The following factories have been considered:
org.apache.flink.table.sinks.CsvBatchTableSinkFactory
org.apache.flink.table.sinks.CsvAppendTableSinkFactory
org.apache.flink.streaming.connectors.kafka.Kafka011TableSourceSinkFactory
org.apache.flink.streaming.connectors.kafka.Kafka010TableSourceSinkFactory
org.apache.flink.streaming.connectors.kafka.Kafka09TableSourceSinkFactory
org.apache.flink.table.filesystem.FileSystemTableFactory
	at org.apache.flink.table.factories.TableFactoryService.filterByContext(TableFactoryService.java:322)
	at org.apache.flink.table.factories.TableFactoryService.filter(TableFactoryService.java:190)
	at org.apache.flink.table.factories.TableFactoryService.findSingleInternal(TableFactoryService.java:143)
	at org.apache.flink.table.factories.TableFactoryService.find(TableFactoryService.java:96)
	at org.apache.flink.table.factories.TableFactoryUtil.findAndCreateTableSink(TableFactoryUtil.java:89)
	... 22 more

Process finished with exit code 1

maven工程解決方案:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-jdbc_2.12</artifactId>
    <version>1.12.0</version>
</dependency>

Flink SQL Client工程解決方案:

对于Flink SQL Client如果碰到这种问题,对于Flink1.12版本,确保$FLINK_HOME/lib下面有flink-connector-jdbc_2.12-1.12.0.jar

commons-cli-1.4.jar
flink-connector-hive_2.12-1.12.0.jar
flink-connector-jdbc_2.12-1.12.0.jar
flink-connector-kafka_2.12-1.12.0.jar
flink-csv-1.12.0.jar
flink-dist_2.12-1.12.0.jar
flink-json-1.12.0.jar
flink-shaded-hadoop-3-uber-3.1.1.7.0.3.0-79-7.0.jar
flink-shaded-zookeeper-3.4.14.jar
flink-sql-connector-hive-3.1.2_2.12-1.12.0.jar
flink-table_2.12-1.12.0.jar
flink-table-blink_2.12-1.12.0.jar
flink-table-planner_2.12-1.12.0.jar
flink-table-planner-blink_2.12-1.12.0.jar
hadoop-yarn-api-3.1.2.jar
hive-common-3.1.2.jar
hive-exec-3.1.2.jar
javax.ws.rs-api-2.0.jar
kafka-clients-2.5.0.jar
log4j-1.2-api-2.12.1.jar
log4j-api-2.12.1.jar
log4j-core-2.12.1.jar
log4j-slf4j-impl-2.12.1.ja

另外,还有可能出现下面这种错误

[ERROR] Could not execute SQL statement. Reason:
java.lang.NoSuchMethodError: org.apache.flink.table.api.TableColumn.isGenerated()Z
这种错误是因为使用了错误的jdbc的jar包（jar包版本错误）导致的.

########################################################################################################################################

注意TableSourceFactory的处理办法[1]与TableSinkFactory完全不同

Reference:

[1]Could not find a suitable table factory for ‘org.apache.flink.table.factories.TableSourceFactory‘