?????? Pyflink jdbc????

???? Tue, 01 Jun 2021 04:23:30 -0700

??????????????????




------------------&nbsp;????????&nbsp;------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<dian0511...@gmail.com&gt;;
????????:&nbsp;2021??6??1??(??????) ????7:21
??????:&nbsp;"user-zh"<user-zh@flink.apache.org&gt;;"????"<1129656...@qq.com&gt;;

????:&nbsp;Re: Pyflink jdbc????



??????????????\????????/????

file:///D:/Pyproject/flink-connector-jdbc_2.11-1.13.1.jar



&gt; 2021??6??1?? ????5:40?????? <1129656...@qq.com&gt; ??????
&gt; 
&gt; 
????????????????????pycharm????????????windows????????????????????file:///D:\Pyproject\flink-connector-jdbc_2.11-1.13.1.jar;&nbsp;
 ??????????????????????????????
&gt; 
&gt; 
&gt; ------------------&amp;nbsp;????????&amp;nbsp;------------------
&gt; 
??????:&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 
"????"&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 <1129656...@qq.com&amp;gt;;
&gt; ????????:&amp;nbsp;2021??6??1??(??????) ????5:30
&gt; ??????:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;;
&gt; 
&gt; ????:&amp;nbsp;?????? Pyflink jdbc????
&gt; 
&gt; 
&gt; 
&gt; 
&gt; 
&gt; ????????????2.11??????????????????
&gt; 
&gt; 
&gt; ------------------ ???????? ------------------
&gt; 
??????:&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 
"user-zh"&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
 <dian0511...@gmail.com&amp;gt;;
&gt; ????????:&amp;nbsp;2021??6??1??(??????) ????5:04
&gt; 
??????:&amp;nbsp;"user-zh"<user-zh@flink.apache.org&amp;gt;;"????"<1129656...@qq.com&amp;gt;;
&gt; 
&gt; ????:&amp;nbsp;Re: Pyflink jdbc????
&gt; 
&gt; 
&gt; 
&gt; Hi??
&gt; 
&gt; ??????????
&gt; 
1????flink-connector-jdbc_2.11-1.13.1.jar??????????PyFlink??????????JAR????scala
 2.11??
&gt; 
&gt; 
&gt; flink run:
&gt; 1) 
????????sink????????????print??????????table_sink??????????SQL??????????table_sink??
&gt; 
&gt; 
&gt; &amp;gt; 2021??6??1?? ????4:33?????? <1129656...@qq.com&amp;gt; ??????
&gt; &amp;gt; 
&gt; &amp;gt; Hi??
&gt; &amp;gt; &amp;amp;nbsp; 
&amp;amp;nbsp;??????????????https://mp.weixin.qq.com/s/GyFTjQl6ch8jc733mpCP7Q 
????????pyflink??????????????????????jdbc??jar??????????????????jdbc??????????flink??????1.13.1
&gt; &amp;gt; ????????????????
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; from pyflink.datastream import StreamExecutionEnvironment
&gt; &amp;gt; from pyflink.table import StreamTableEnvironment, 
EnvironmentSettings
&gt; &amp;gt; env = StreamExecutionEnvironment.get_execution_environment()
&gt; &amp;gt; table_env = StreamTableEnvironment.create( 
env,environment_settings=EnvironmentSettings.new_instance().use_blink_planner().build())
&gt; &amp;gt; 
table_env.get_config().get_configuration().set_string("pipeline.classpaths", 
"file:///home/flink/lib/flink-connector-jdbc_2.12-1.13.1.jar;file:///home/flink/lib/mysql-connector-java-8.0.25.jar")
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; # 2. create source Table
&gt; &amp;gt; table_env.execute_sql("""
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; CREATE TABLE table_source (
&gt; &amp;gt; &amp;amp;nbsp; e string
&gt; &amp;gt; ) WITH (
&gt; &amp;gt; &amp;amp;nbsp;'connector' = 'jdbc',
&gt; &amp;gt; &amp;amp;nbsp; 'url' = 'jdbc:mysql://********:3306/test',
&gt; &amp;gt; &amp;amp;nbsp; 'driver' = 'com.mysql.cj.jdbc.Driver',
&gt; &amp;gt; &amp;amp;nbsp; 'table-name' = 'enum_test',
&gt; &amp;gt; &amp;amp;nbsp; 'username' = 'pms_etl',
&gt; &amp;gt; &amp;amp;nbsp; 'password' = 'pms_etl_q'
&gt; &amp;gt; )
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; """)
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; # 3. create sink Table
&gt; &amp;gt; table_env.execute_sql("""
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; CREATE TABLE print (
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; e 
string
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; ) WITH (
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; 
'connector' = 'print'
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; )
&gt; &amp;gt; """)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp;
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; table_env.execute_sql("INSERT INTO table_sink SELECT * FROM 
table_source").wait()
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; ????????python????????????????????
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; Traceback (most recent call last):
&gt; &amp;gt; &amp;amp;nbsp; File "demo.py", line 41, in <module&amp;amp;gt;
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; table_env.execute_sql("INSERT INTO 
table_sink SELECT * FROM table_source").wait()
&gt; &amp;gt; &amp;amp;nbsp; File 
"/usr/local/lib/python3.7/site-packages/pyflink/table/table_environment.py", 
line 804, in execute_sql
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; return 
TableResult(self._j_tenv.executeSql(stmt))
&gt; &amp;gt; &amp;amp;nbsp; File 
"/usr/local/lib/python3.7/site-packages/py4j/java_gateway.py", line 1286, in 
__call__
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; answer, self.gateway_client, 
self.target_id, self.name)
&gt; &amp;gt; &amp;amp;nbsp; File 
"/usr/local/lib/python3.7/site-packages/pyflink/util/exceptions.py", line 146, 
in deco
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; return f(*a, **kw)
&gt; &amp;gt; &amp;amp;nbsp; File 
"/usr/local/lib/python3.7/site-packages/py4j/protocol.py", line 328, in 
get_return_value
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; format(target_id, ".", name), value)
&gt; &amp;gt; py4j.protocol.Py4JJavaError: An error occurred while calling 
o4.executeSql.
&gt; &amp;gt; : org.apache.flink.table.api.ValidationException: Unable to 
create a source for reading table 
'default_catalog.default_database.table_source'.
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; Table options are:
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; 'connector'='jdbc'
&gt; &amp;gt; 'driver'='com.mysql.cj.jdbc.Driver'
&gt; &amp;gt; 'password'='pms_etl_q'
&gt; &amp;gt; 'table-name'='enum_test'
&gt; &amp;gt; 'url'='jdbc:mysql://*******:3306/test'
&gt; &amp;gt; 'username'='pms_etl'
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:137)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.plan.schema.CatalogSourceTable.createDynamicTableSource(CatalogSourceTable.java:116)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.plan.schema.CatalogSourceTable.toRel(CatalogSourceTable.java:82)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3585)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2507)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2144)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2093)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2050)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:663)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:644)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3438)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:570)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:170)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:162)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:967)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:936)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:275)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlInsert(SqlToOperationConverter.java:595)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:268)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:101)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:724)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
java.lang.reflect.Method.invoke(Method.java:498)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
java.lang.Thread.run(Thread.java:748)
&gt; &amp;gt; Caused by: org.apache.flink.table.api.ValidationException: Cannot 
discover a connector using option: 'connector'='jdbc'
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.enrichNoMatchingConnectorError(FactoryUtil.java:467)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.getDynamicTableFactory(FactoryUtil.java:441)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.createTableSource(FactoryUtil.java:133)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; ... 
31 more
&gt; &amp;gt; Caused by: org.apache.flink.table.api.ValidationException: Could 
not find any factory for identifier 'jdbc' that implements 
'org.apache.flink.table.factories.DynamicTableFactory' in the classpath.
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; Available factory identifiers are:
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; blackhole
&gt; &amp;gt; datagen
&gt; &amp;gt; filesystem
&gt; &amp;gt; print
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.discoverFactory(FactoryUtil.java:319)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.factories.FactoryUtil.enrichNoMatchingConnectorError(FactoryUtil.java:463)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; ... 
33 more
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; ????flink run -py demo.py ??????????????
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; &amp;amp;nbsp; File "./demo.py", line 41, in <module&amp;amp;gt;
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; table_env.execute_sql("INSERT INTO 
table_sink SELECT * FROM table_source").wait()
&gt; &amp;gt; &amp;amp;nbsp; File 
"/home/flink/opt/python/pyflink.zip/pyflink/table/table_environment.py", line 
804, in execute_sql
&gt; &amp;gt; &amp;amp;nbsp; File 
"/home/flink/opt/python/py4j-0.10.8.1-src.zip/py4j/java_gateway.py", line 1286, 
in __call__
&gt; &amp;gt; &amp;amp;nbsp; File 
"/home/flink/opt/python/pyflink.zip/pyflink/util/exceptions.py", line 158, in 
deco
&gt; &amp;gt; pyflink.util.exceptions.TableException: 
org.apache.flink.table.api.TableException: Sink 
`default_catalog`.`default_database`.`table_sink` does not exists
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:233)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.delegation.PlannerBase.$anonfun$translate$1(PlannerBase.scala:162)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:233)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.Iterator.foreach(Iterator.scala:937)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.Iterator.foreach$(Iterator.scala:937)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.AbstractIterator.foreach(Iterator.scala:1425)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.IterableLike.foreach(IterableLike.scala:70)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.IterableLike.foreach$(IterableLike.scala:69)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.AbstractIterable.foreach(Iterable.scala:54)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.TraversableLike.map(TraversableLike.scala:233)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.TraversableLike.map$(TraversableLike.scala:226)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
scala.collection.AbstractTraversable.map(Traversable.scala:104)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:162)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:1518)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:740)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeInternal(TableEnvironmentImpl.java:856)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:730)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
java.lang.reflect.Method.invoke(Method.java:498)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
java.lang.Thread.run(Thread.java:748)
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; org.apache.flink.client.program.ProgramAbortException: 
java.lang.RuntimeException: Python process exits with code: 1
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:134)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
java.lang.reflect.Method.invoke(Method.java:498)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.program.PackagedProgram.callMainMethod(PackagedProgram.java:355)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:222)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:114)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:812)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:246)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.cli.CliFrontend.parseAndRun(CliFrontend.java:1054)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:1132)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.runtime.security.contexts.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:28)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:1132)
&gt; &amp;gt; Caused by: java.lang.RuntimeException: Python process exits with 
code: 1
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; at 
org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:124)
&gt; &amp;gt; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; &amp;amp;nbsp; ... 
13 more
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; 
&gt; &amp;gt; ??????????????????

?????? Pyflink jdbc????

回复