大数据flink实时计算方案大数据实时计算引擎flink _生活百科

一、背景
Flink 由于阿里在国内的助推，火爆程度可以想象，且目前Flink 有非常明显的趋势是往SQL 方向进行的。很多大厂已经实现了Flink SQL化，那我们怎么去实现一个流式计算平台呢？

文章插图
二、Flink SQL 初探以及代码实现
连接kafka 对数据进行处理写入mysql
package org.example;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
public class SqlDemo {
public static void main(String[] args) throws Exception {
//创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
TableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
//把kafka 中的topic映射成一个输入临时表
tableEnv.executeSql(
"create table sensor_source (id string,name string) with (" +
" 'connector' = 'kafka'," +
"'topic' = 'test_info_test'," +
"'properties.bootstrap.servers' = 'localhost:9092'," +
"'properties.group.id' = 'testGroup'," +
"'scan.startup.mode' = 'earliest-offset'," +
"'format' = 'json')"
);
//把mysql 中的表映射成一个输出临时表
String sql = "CREATE TABLE print_table (\n" +
" id STRING,\n" +
" name STRING\n" +
") WITH (\n" +
" 'connector' = 'print'\n" +
")";
String mysql_sql = "CREATE TABLE mysql_sink (\n" +
"id string,\n" +
"name string\n" +
" ) WITH (\n" +
"'connector' = 'jdbc',\n" +
"'url' = 'jdbc:mysql://ip:8081/kafka?serverTimezone=UTC',\n" +
"'table-name' = 'test_info',\n" +
"'username' = 'kafka',\n" +
"'password' = 'Bonc123'\n" +
" )";
String kafka_sink_sql=
"create table kafka_sink (id string,name string) with (" +
" 'connector' = 'kafka'," +
"'topic' = 'test_info_2'," +
"'properties.bootstrap.servers' = 'localhost:9092'," +
"'format' = 'json')";
tableEnv.executeSql(mysql_sql);
//tableEnv.executeSql(kafka_sink_sql);
//tableEnv.executeSql(sql);
//插入数据的sql语句
//tableEnv.executeSql("insert into print_table select * from sensor_source");
tableEnv.executeSql("insert into mysql_sink select * from sensor_source");
//tableEnv.executeSql("insert into kafka_sink select * from sensor_source");
}
}
运行之后mysql里面数据就有了