Pyflink writes MySQL examples with JDBC

pyflink use jdbc Connect mysql Example
Get ready ：

Software ：python3.7 pycharm

oriented flink1.13.5
Install the module ：python -m pip install apache-flink==1.13.5
add to mysql Of jdbc jar plug-in unit 、pyflink Of flink-connector-jdbc_2.11 Two plug-ins
To pyflink Modular lib Folder （ Be careful mysql Of jdbc edition 、pyflink The path of 、flink Version of ）

${PYTHON_HOME} \Python37\site-packages\pyflink\lib

establish mysql surface

CREATE TABLE `print_table` (

  `f0` int(11) DEFAULT NULL,
  `f1` int(11) DEFAULT NULL,
  `f2` varchar(500) DEFAULT NULL
)`

Specifically python Code

（ Be careful jdbc url Of hostname、database and user、password Make changes according to your ）

from pyflink.table import EnvironmentSettings, TableEnvironment

# 1.  establish  TableEnvironment
env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
table_env = TableEnvironment.create(env_settings)


# 2.  establish  source  surface 
table_env.execute_sql("""CREATE TABLE source_table ( f0 INT, f1 INT, f2 STRING ) WITH ( 'connector' = 'datagen', 'rows-per-second'='5' ) """)

# 3.  establish  sink  surface 
table_env.execute_sql("""CREATE TABLE print_table ( f0 INT, f1 INT, f2 STRING ) WITH ( 'connector' = 'jdbc', 'url'='jdbc:mysql://hostname:3306/test?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC', 'username'='root', 'password'='password', 'table-name' = 'print_table' ) """)


#  Or by  SQL  Query statements to write  sink  surface ：
table_env.execute_sql("insert into print_table select f0,f1,f2 from source_table").wait(60000)
# wait(60000) Of 60000 It's a timeout ,60000 millisecond , namely 60 second , This can be changed according to your own needs 

Source of reprint