FlinkSQL编程

2022-07-09 187

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

实时计算 Flink 版，5000CU*H 3个月

云数据库 RDS MySQL Serverless，0.5-2RCU 50GB

简介： 查询已注册表查询未注册表Flink sql 标准代码结果Flink sql读写kafkaFlink sql去kafka写mysql

查询已注册表

packagecom.blink.sb.sql;
importcom.blink.sb.beans.ClickLogs;
importorg.apache.flink.streaming.api.datastream.DataStream;
importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
importorg.apache.flink.table.api.Table;
importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;
importorg.apache.flink.types.Row;
/*** Flink SQL查询已注册的表*/publicclassFlinkSQLQueryRegedTable {
publicstaticvoidmain(String[] args) throwsException {
//1、获取Stream执行环境StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();
//2、创建表执行环境StreamTableEnvironmenttEnv=StreamTableEnvironment.create(env);
env.setParallelism(1);
//3、读取数据DataStream<ClickLogs>clickLogs=env.fromElements(
"Mary,./home,2022-02-02 12:00:00",
"Bob,./cart,2022-02-02 12:00:00",
"Mary,./prod?id=1,2022-02-02 12:00:05",
"Liz,./home,2022-02-02 12:01:00",
"Bob,./prod?id=3,2022-02-02 12:01:30",
"Mary,./prod?id=7,2022-02-02 12:01:45"        ).map(event-> {
String[] props=event.split(",");
returnClickLogs                    .builder()
                    .user(props[0])
                    .url(props[1])
                    .cTime(props[2])
                    .build();
        });
//4、流转换为动态表Tabletable=tEnv.fromDataStream(clickLogs);
//5、注册表tEnv.createTemporaryView("sourceTable",table);
//6、SQL查询已注册表TableresultTable=tEnv.sqlQuery("select user,count(url) as cnt from sourceTable group by user");
//7、转换成流直接打印或者输出tEnv.toChangelogStream(resultTable).print();
//8、执行任务env.execute("FlinkSQLQueryUnregTable");
    }
}

查询未注册表

packagecom.blink.sb.sql;
importcom.blink.sb.beans.ClickLogs;
importorg.apache.flink.streaming.api.datastream.DataStream;
importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
importorg.apache.flink.table.api.Table;
importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;
/*** Flink SQL查询未注册的表*/publicclassFlinkSQLQueryUnregTable {
publicstaticvoidmain(String[] args) throwsException {
//1、获取Stream执行环境StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();
//2、创建表执行环境StreamTableEnvironmenttEnv=StreamTableEnvironment.create(env);
env.setParallelism(1);
//3、读取数据DataStream<ClickLogs>clickLogs=env.fromElements(
"Mary,./home,2022-02-02 12:00:00",
"Bob,./cart,2022-02-02 12:00:00",
"Mary,./prod?id=1,2022-02-02 12:00:05",
"Liz,./home,2022-02-02 12:01:00",
"Bob,./prod?id=3,2022-02-02 12:01:30",
"Mary,./prod?id=7,2022-02-02 12:01:45"        ).map(event-> {
String[] props=event.split(",");
returnClickLogs                    .builder()
                    .user(props[0])
                    .url(props[1])
                    .cTime(props[2])
                    .build();
        });
//4、流转换为动态表Tabletable=tEnv.fromDataStream(clickLogs);
//5、SQL查询未注册表(注意table两边得有空格)TableresultTable=tEnv.sqlQuery("select user,count(url) as cnt from "+table+" group by user");
//6、执行并打印resultTable.execute().print();
    }
}

Flink SQL标准代码结构

Flink SQL的代码结构跟Flink Table API⼀模⼀样，核⼼就是Table。

这⾥以⽂件系统作为source和sink table举例如下

packagecom.blink.sb.sql;
importorg.apache.flink.table.api.*;
/*** Flink SQL标准结构*/publicclassFlinkSQLStandardStructure {
publicstaticvoidmain(String[] args) {
//1、创建TableEnvironmentEnvironmentSettingssettings=EnvironmentSettings                .newInstance()
                .build();
TableEnvironmenttEnv=TableEnvironment.create(settings);
//2、创建source table(DDL语句)-会自动注册表的tEnv.executeSql("CREATE TABLE emp_info ("+"    emp_id INT,"+"    name VARCHAR,"+"    dept_id INT"+") WITH ("+"    'connector' = 'filesystem',"+"    'path' = 'data/emp/input/',"+"    'format' = 'csv'"+")");//最后不要有分号,注意空格//3、创建sink table(DDL)//executeSql执行tEnv.executeSql("CREATE TABLE emp_info_copy ("+"    emp_id INT,"+"    name VARCHAR,"+"    dept_id INT"+") WITH ("+"    'connector' = 'filesystem',"+"    'path' = 'data/emp/output/',"+"    'format' = 'csv'"+")");
//4、执行SQL查询并输出结果TableresultTable=tEnv.sqlQuery("select * from emp_info where dept_id=10");
tEnv.createTemporaryView("result",resultTable);
tEnv.executeSql("INSERT INTO emp_info_copy "+"SELECT"+"   emp_id,"+"   name,"+"   dept_id "+"FROM result");
    }
}

Flink SQL 语法支持

Flink SQL截⽌到1.14⽀持的SQL语法详⻅如下链接：

https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/sql/overview/

Flink SQL输入和输出

读写kafka

需求：从kafka消费点击⽇志（JSON），转化为CSV格式之后输出到kafka

packagecom.blink.sb.sql;
importorg.apache.flink.streaming.api.datastream.DataStream;
importorg.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
importorg.apache.flink.table.api.EnvironmentSettings;
importorg.apache.flink.table.api.Table;
importorg.apache.flink.table.api.TableEnvironment;
importorg.apache.flink.table.api.bridge.java.StreamTableEnvironment;
importorg.apache.flink.types.Row;
importjava.awt.event.TextEvent;
/*** 以Flink SQL方式从kafka消费点击日志（JSON），转化为CSV格式之后输出到Kafka*/publicclassFlinkSQLKafka2Kafka {
publicstaticfinalStringinput_topic="clicklog_input";
publicstaticfinalStringoutput_topic="clicklog_output";
publicstaticvoidmain(String[] args) throwsException {
//1、创建TableEnvironmentEnvironmentSettingssettings=EnvironmentSettings                .newInstance()
                .build();
StreamTableEnvironmenttEnv=StreamTableEnvironment.create(StreamExecutionEnvironment.getExecutionEnvironment(), settings);
//2、创建source table(DDL语句)-会自动注册表的tEnv.executeSql("CREATE TABLE sourceTable ("+"  `user` STRING,"+"  `url` STRING,"+"  `cTime` STRING"+") WITH ("+"  'connector' = 'kafka',"+"  'topic' = '"+input_topic+"',"+"  'properties.bootstrap.servers' = 'node02:6667',"+"  'properties.group.id' = 'test1',"+"  'scan.startup.mode' = 'latest-offset',"+"  'format' = 'json'"+")");
//3、创建sink table(DDL)tEnv.executeSql("CREATE TABLE sinkTable ("+"  `user` STRING,"+"  `url` STRING,"+"  `cTime` STRING"+") WITH ("+"  'connector' = 'kafka',"+"  'topic' = '"+output_topic+"',"+"  'properties.bootstrap.servers' = 'node02:6667',"+"  'format' = 'csv'"+")");
//4、执行SQL查询并输出结果tEnv.executeSql("INSERT INTO sinkTable "+"SELECT"+"   user,"+"   url,"+"   cTime "+"FROM sourceTable");
    }
}

读kafka写mysql

引入jdbc connector需要的依赖库：

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.11</artifactId><version>${flink.version}</version></dependency>

添加对应的数据库驱动：

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.28</version></dependency>

提前在mysql中建表

CREATETABLEtest.sinkTable (
`user`varchar(100) ,
`cnt`BIGINT)
ENGINE=InnoDBDEFAULTCHARSET=utf8COLLATE=utf8_general_ci;

相关代码：

packagecom.blink.sb.sql;
importorg.apache.flink.table.api.EnvironmentSettings;
importorg.apache.flink.table.api.TableEnvironment;
importjava.math.BigInteger;
/*** 以Flink SQL方式从kafka消费点击日志（JSON），统计之后输出到MySQL*/publicclassFlinkSQLKafka2MySQL {
publicstaticfinalStringinput_topic="clicklog_input";
publicstaticvoidmain(String[] args) throwsException {
//1、创建TableEnvironmentEnvironmentSettingssettings=EnvironmentSettings                .newInstance()
                .build();
TableEnvironmenttEnv=TableEnvironment.create(settings);
//2、创建source table(DDL语句)-会自动注册表的tEnv.executeSql("CREATE TABLE sourceTable ("+"  `user` STRING,"+"  `url` STRING,"+"  `cTime` STRING"+") WITH ("+"  'connector' = 'kafka',"+"  'topic' = '"+input_topic+"',"+"  'properties.bootstrap.servers' = 'node02:6667',"+"  'properties.group.id' = 'test1',"+"  'scan.startup.mode' = 'latest-offset',"+"  'format' = 'json'"+")");
//3、创建sink table(DDL)//        tEnv.executeSql("CREATE TABLE sinkTable (" +//                "  `user` STRING," +//                "  `url` STRING," +//                "  `cTime` STRING" +//                ") WITH (" +//                "  'connector' = 'jdbc'," +//                "  'url' = 'jdbc:mysql://node01:3306/test'," +//                "  'username' = 'root'," +//                "  'password' = 'root%123'," +//                "  'table-name' = 'sinkTable'" +//                ")");tEnv.executeSql("CREATE TABLE sinkTable ("+"  `user` STRING,"+"  `cnt` BIGINT,"+"  PRIMARY KEY (`user`) NOT ENFORCED"+") WITH ("+"  'connector' = 'jdbc',"+"  'url' = 'jdbc:mysql://node01:3306/test',"+"  'username' = 'root',"+"  'password' = 'root%123',"+"  'table-name' = 'sinkTable'"+")");
//4、执行SQL查询并输出结果//        tEnv.executeSql("INSERT INTO sinkTable " +//                "SELECT" +//                "   user," +//                "   url," +//                "   cTime " +//                "FROM sourceTable");tEnv.executeSql("INSERT INTO sinkTable "+"SELECT"+"   user,"+"   count(url) as cnt "+"FROM sourceTable "+"group by user");
    }
}