pyflink.sql

-- Mysql 建表语句，注意这是在Mysql执行的
 CREATE TABLE `dim_behavior` (
 `id` int(10) NOT NULL AUTO_INCREMENT COMMENT '自增主键',
 `en_behavior` varchar(255) COLLATE utf8_bin DEFAULT NULL COMMENT '英文 行为',
 `zh_behavior` varchar(255) COLLATE utf8_bin DEFAULT NULL COMMENT '中文 行为',
 PRIMARY KEY (`id`)
 ) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;



-- Mysql插入两条数据
INSERT INTO `dijie_test`.`dim_behavior`(`id`, `en_behavior`, `zh_behavior`) VALUES (1, 'buy', '购买');
INSERT INTO `dijie_test`.`dim_behavior`(`id`, `en_behavior`, `zh_behavior`) VALUES (2, 'pv', '浏览');



---下面是在终端运行
$FLINK_HOME/bin/sql-client.sh embedded -e $FLINK_HOME/conf/env.yaml -pyfs $FLINK_HOME/conf/test.py



---下面在Flink SQL Client中运行
show functions;

---下面是flink SQL Client中运行的source语句
 CREATE TABLE `dim_behavior` (
 `id` int ,
 `en_behavior` varchar,
 `zh_behavior` varchar 
 )WITH (
 'connector.type' = 'jdbc',
 'connector.url' = 'jdbc:mysql://Desktop:3306/dijie_test',
 'connector.table' = 'dim_behavior',
 'connector.driver' = 'com.mysql.cj.jdbc.Driver', 
 'connector.username' = 'appleyuchi',
 'connector.password' = 'appleyuchi',
 'connector.lookup.cache.max-rows' = '5000', 
 'connector.lookup.cache.ttl' = '10s' 
 );
 
select func1(en_behavior) from dim_behavior;