?????? flink sql ????mysql ??????????

?????? Wed, 25 Nov 2020 04:34:37 -0800

Hi,
&nbsp; ??????????????????execute??????????????????????????????
StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = 
EnvironmentSettings.newInstance().useAnyPlanner().inStreamingMode().build();
StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, 
settings);



bsTableEnv.getConfig().getConfiguration().set(
&nbsp; &nbsp; &nbsp; ExecutionCheckpointingOptions.CHECKPOINTING_MODE, 
CheckpointingMode.EXACTLY_ONCE);
bsTableEnv.getConfig().getConfiguration().set(
&nbsp; &nbsp; &nbsp; ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL, 
Duration.ofSeconds(10));


String ddl = "CREATE TABLE meson_budget_data (\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; ID BIGINT,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Gangtise_Code 
STRING,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; InnerCode 
INT,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Tradingday 
DATE, \n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; ClosePrice 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_3day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_5day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_10day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_20day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_60day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_120day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; Close_250day 
DOUBLE,\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; PRIMARY KEY 
(ID) NOT ENFORCED\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; ) WITH(\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; &nbsp; 
'connector' = 'jdbc',\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; &nbsp; 'url' = 
'jdbc:mysql://gangtisedb.mysql.database.chinacloudapi.cn_3306',\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; &nbsp; 
'table-name' = 'gangtise_budget.meson_budget_data',\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // "&nbsp; &nbsp; 
'username' = 'root',\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // "&nbsp; &nbsp; 
'password' = '123456',\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; &nbsp; 'driver' 
= 'com.mysql.jdbc.Driver'\n" +
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; "&nbsp; )";


bsTableEnv.executeSql(ddl);
TableResult mtr = bsTableEnv.executeSql("select * from meson_budget_data");
mtr.print();


env.execute("sql test");
// bsTableEnv.execute("sql test");



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


------------------&nbsp;????????&nbsp;------------------
??????:                                                                         
                                               "user-zh"                        
                                                            
<[email protected]&gt;;
????????:&nbsp;2020??11??25??(??????) ????6:22
??????:&nbsp;"user-zh"<[email protected]&gt;;

????:&nbsp;Re: flink sql ????mysql ??????????






????????????????????????tableEnv.execute("test");



AlfredFeng wrote
&gt; Hi All,
&gt; &amp;nbsp; &amp;nbsp; ????Idea????flink-jdbc-connector????mysql,
&gt; ????????????env.executeSql("select * from 
my_table").print()????????????????????????????????????????
&gt; flink????1.11.2





--
Sent from: http://apache-flink.147419.n8.nabble.com/

?????? flink sql ????mysql ??????????

回复