pyflink????????

Mon, 15 Jun 2020 00:42:03 -0700 

????????Flink????????????????????????????????????????????????????Traceback 
(most recent call last):   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/2add2d51-8a74-40c4-9c42-69c7c12feb05pyflink.zip/pyflink/util/exceptions.py",
 line 147, in deco   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/2add2d51-8a74-40c4-9c42-69c7c12feb05py4j-0.10.8.1-src.zip/py4j/protocol.py",
 line 328, in get_return_value py4j.protocol.Py4JJavaError: An error occurred 
while calling o46.insertInto. : org.apache.flink.table.api.TableException: 
BatchTableSink or OutputFormatTableSink required to emit batch Table.       at 
org.apache.flink.table.api.internal.BatchTableEnvImpl.writeToSink(BatchTableEnvImpl.scala:154)
       at 
org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:664)
  at 
org.apache.flink.table.api.internal.TableEnvImpl.insertInto(TableEnvImpl.scala:607)
  at 
org.apache.flink.table.api.internal.TableImpl.insertInto(TableImpl.java:411)    
     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.api.python.shaded.py4j.reflection.MethodInvoker.invoke(MethodInvoker.java:244)
      at 
org.apache.flink.api.python.shaded.py4j.reflection.ReflectionEngine.invoke(ReflectionEngine.java:357)
        at 
org.apache.flink.api.python.shaded.py4j.Gateway.invoke(Gateway.java:282)     at 
org.apache.flink.api.python.shaded.py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:132)
      at 
org.apache.flink.api.python.shaded.py4j.commands.CallCommand.execute(CallCommand.java:79)
    at 
org.apache.flink.api.python.shaded.py4j.GatewayConnection.run(GatewayConnection.java:238)
    at java.lang.Thread.run(Thread.java:748) During handling of the above 
exception, another exception occurred: Traceback (most recent call last):   
File "/usr/local/python3/lib/python3.7/runpy.py", line 193, in 
_run_module_as_main     "__main__", mod_spec)   File 
"/usr/local/python3/lib/python3.7/runpy.py", line 85, in _run_code     
exec(code, run_globals)   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/3de6e8aa-f179-4356-b2d3-792601cdc714/flinkrescount.py",
 line 33, in <module&gt;     
t_env.from_path('mySource').group_by('word').select('word,count(1)').insert_into('flink_test')
   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/2add2d51-8a74-40c4-9c42-69c7c12feb05pyflink.zip/pyflink/table/table.py",
 line 679, in insert_into   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/2add2d51-8a74-40c4-9c42-69c7c12feb05py4j-0.10.8.1-src.zip/py4j/java_gateway.py",
 line 1286, in __call__   File 
"/tmp/pyflink/db00a36e-521f-4109-b787-712687fcb3b4/2add2d51-8a74-40c4-9c42-69c7c12feb05pyflink.zip/pyflink/util/exceptions.py",
 line 154, in deco pyflink.util.exceptions.TableException: 'BatchTableSink or 
OutputFormatTableSink required to emit batch Table.' 
org.apache.flink.client.program.OptimizerPlanEnvironment$ProgramAbortException  
       at 
org.apache.flink.client.python.PythonDriver.main(PythonDriver.java:87)       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:321)
     at 
org.apache.flink.client.program.PackagedProgram.invokeInteractiveModeForExecution(PackagedProgram.java:205)
  at org.apache.flink.client.ClientUtils.executeProgram(ClientUtils.java:138)   
  at 
org.apache.flink.client.cli.CliFrontend.executeProgram(CliFrontend.java:662)    
     at org.apache.flink.client.cli.CliFrontend.run(CliFrontend.java:210)    at 
org.apache.flink.client.cli.CliFrontend.parseParameters(CliFrontend.java:893)   
     at 
org.apache.flink.client.cli.CliFrontend.lambda$main$10(CliFrontend.java:966)    
     at 
org.apache.flink.runtime.security.NoOpSecurityContext.runSecured(NoOpSecurityContext.java:30)
        at org.apache.flink.client.cli.CliFrontend.main(CliFrontend.java:966) 

python??????????



from pyflink.datastream import StreamExecutionEnvironment
from pyflink.dataset import ExecutionEnvironment
from pyflink.table import 
BatchTableEnvironment,TableEnvironment,TableConfig,DataTypes,StreamTableEnvironment,EnvironmentSettings
from pyflink.table.descriptors import Schema, OldCsv, FileSystem



flink_test = """
CREATE TABLE flink_test (
 name VARCHAR,
 cnt BIGINT
) WITH (
 'connector.type' = 'jdbc',
 'connector.url' = 'jdbc:mysql://localhost:3306/test',
 'connector.table' = 'flink_test',
 'connector.username' = 'root',
 'connector.password' = '11122344',
 'connector.write.flush.interval' = '1s'
)
"""

#exec_env = ExecutionEnvironment.get_execution_environment()
#t_config = TableConfig()
#t_env = BatchTableEnvironment.create(exec_env,t_config)
env = ExecutionEnvironment.get_execution_environment()
#environment_settions = 
EnvironmentSettings.new_instance().use_blink_planner().build()
t_config = TableConfig()
t_env = BatchTableEnvironment.create(env,t_config)
#,environment_settings=environment_settions

t_env.connect(FileSystem().path('/home/admin/data/input')).with_format(OldCsv().field('word',DataTypes.STRING())).with_schema(Schema().field('word',DataTypes.STRING())).create_temporary_table('mySource')
t_env.sql_update(flink_test)
t_env.from_path('mySource').group_by('word').select('word,count(1)').insert_into('flink_test')

t_env.execute("flink_link_mysql")

回复