Flink sql 执行报错

[gbguanbo]

from pyalink.alink import *
#ExecutionEnvironment、BatchTableEnvironment、StreamExecutionEnvironment 和 StreamTableEnvironment。
benv, btenv, senv, stenv = useRemoteEnv("localhost", 8081, 1)

jars = "file:///home/guanbo/flink-1.13.3/lib/flink-csv-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-dist_2.11-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-json-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-shaded-zookeeper-3.4.14.jar"
+"file:///home/guanbo/flink-1.13.3/lib/flink-table_2.11-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-table-blink_2.11-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/alink_core_flink-1.13_2.11-1.6.2.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/alink_python_flink-1.13_2.11-1.6.2.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/log4j-1.2-api-2.12.1.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/log4j-api-2.12.1.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/log4j-core-2.12.1.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/log4j-slf4j-impl-2.12.1.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-connector-base-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-streaming-java_2.11-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-table-api-java-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-table-api-java-bridge_2.11-1.13.3.jar;"
+"file:///home/guanbo/flink-1.13.3/lib/flink-table-common-1.13.3.jar"
stenv.get_config().get_configuration().set_string("pipeline.jars", jars)
source_ddl = """
CREATE TABLE source_table(
id STRING,
price DOUBLE,
ts BIGINT
) WITH (
'connector' = 'kafka',
'topic' = 'test-topic',
'properties.bootstrap.servers' = '172.20.8.230:9092',
'properties.group.id' = 'test_3',
'scan.startup.mode' = 'latest-offset',
'format' = 'json'
)
"""

sink_ddl = """
CREATE TABLE sink_table(
id STRING
) WITH (
'connector' = 'kafka',
'topic' = 'sink_topic',
'properties.bootstrap.servers' = '172.20.8.230:9092',
'format' = 'json'
)
"""

stenv.execute_sql(source_ddl)
stenv.execute_sql(sink_ddl)

stenv.sql_query("SELECT id FROM source_table")
.execute_insert("sink_table")
StreamOperator.execute(）

Py4JJavaError: An error occurred while calling o4729.sqlQuery.
: org.apache.flink.table.api.ValidationException: SQL validation failed. findAndCreateTableSource failed.
at org.apache.flink.table.calcite.FlinkPlannerImpl.validateInternal(FlinkPlannerImpl.scala:147)
at org.apache.flink.table.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:107)
at org.apache.flink.table.sqlexec.SqlToOperationConverter.convert(SqlToOperationConverter.java:151)
at org.apache.flink.table.planner.ParserImpl.parse(ParserImpl.java:92)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:704)
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 py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
at py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
at py4j.Gateway.invoke(Gateway.java:282)
at py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
at py4j.commands.CallCommand.execute(CallCommand.java:79)
at py4j.GatewayConnection.run(GatewayConnection.java:238)
at java.lang.Thread.run(Thread.java:750)
Caused by: org.apache.flink.table.api.TableException: findAndCreateTableSource failed.
at org.apache.flink.table.factories.TableFactoryUtil.findAndCreateTableSource(TableFactoryUtil.java:45)
at org.apache.flink.table.catalog.DatabaseCalciteSchema.convertCatalogTable(DatabaseCalciteSchema.java:176)
at org.apache.flink.table.catalog.DatabaseCalciteSchema.convertTable(DatabaseCalciteSchema.java:114)
at org.apache.flink.table.catalog.DatabaseCalciteSchema.lambda$getTable$0(DatabaseCalciteSchema.java:95)
at java.util.Optional.map(Optional.java:215)
at org.apache.flink.table.catalog.DatabaseCalciteSchema.getTable(DatabaseCalciteSchema.java:83)
at org.apache.calcite.jdbc.SimpleCalciteSchema.getImplicitTable(SimpleCalciteSchema.java:83)
at org.apache.calcite.jdbc.CalciteSchema.getTable(CalciteSchema.java:289)
at org.apache.calcite.sql.validate.EmptyScope.resolve_(EmptyScope.java:143)
at org.apache.calcite.sql.validate.EmptyScope.resolveTable(EmptyScope.java:99)
at org.apache.calcite.sql.validate.DelegatingScope.resolveTable(DelegatingScope.java:203)
at org.apache.calcite.sql.validate.IdentifierNamespace.resolveImpl(IdentifierNamespace.java:112)
at org.apache.calcite.sql.validate.IdentifierNamespace.validateImpl(IdentifierNamespace.java:184)
at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:997)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:975)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3085)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3070)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3335)
at org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:60)
at org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:997)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:975)
at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:232)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:952)
at org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:704)
at org.apache.flink.table.calcite.FlinkPlannerImpl.validateInternal(FlinkPlannerImpl.scala:142)
... 15 more
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.factories.TableSourceFactory' in
the classpath.

Reason: Required context properties mismatch.

The following properties are requested:
connector=kafka
format=json
properties.bootstrap.servers=172.20.8.230:9092
properties.group.id=test_3
scan.startup.mode=latest-offset
schema.0.data-type=VARCHAR(2147483647)
schema.0.name=id
schema.1.data-type=DOUBLE
schema.1.name=price
schema.2.data-type=BIGINT
schema.2.name=ts
topic=test-topic

The following factories have been considered:
com.alibaba.flink.ml.operator.ops.table.MLTableSourceFactory
org.apache.flink.table.sources.CsvBatchTableSourceFactory
org.apache.flink.table.sources.CsvAppendTableSourceFactory
at org.apache.flink.table.factories.TableFactoryService.filterByContext(TableFactoryService.java:300)
at org.apache.flink.table.factories.TableFactoryService.filter(TableFactoryService.java:178)
at org.apache.flink.table.factories.TableFactoryService.findSingleInternal(TableFactoryService.java:139)
at org.apache.flink.table.factories.TableFactoryService.find(TableFactoryService.java:93)
at org.apache.flink.table.factories.TableFactoryUtil.findAndCreateTableSource(TableFactoryUtil.java:41)
... 41 more