🚕SQL操作

总结：

在基本使用中的例子用的是流来连接
在kafka的例子中使用的是connector直接连接数据库

另外group by + 聚合操作自动是撤回流

基本使用

查询未注册的表

直接使用动态表

  1. public class Flink05_SQL_BaseUse {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.         DataStreamSource<WaterSensor> waterSensorStream =
  6.             env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
  7.                              new WaterSensor("sensor_1", 2000L, 20),
  8.                              new WaterSensor("sensor_2", 3000L, 30),
  9.                              new WaterSensor("sensor_1", 4000L, 40),
  10.                              new WaterSensor("sensor_1", 5000L, 50),
  11.                              new WaterSensor("sensor_2", 6000L, 60));
  13.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  14.         // 使用sql查询未注册的表
  15.         Table inputTable = tableEnv.fromDataStream(waterSensorStream);
  16.         Table resultTable = tableEnv.sqlQuery("select * from " + inputTable + " where id='sensor_1'");
  17.         tableEnv.toAppendStream(resultTable, Row.class).print();
  19.         env.execute();
  20.     }
  21. }

查询已注册的表

注册后，多了一个表名

  1. public class Flink05_SQL_BaseUse_2 {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.         DataStreamSource<WaterSensor> waterSensorStream =
  6.             env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
  7.                              new WaterSensor("sensor_1", 2000L, 20),
  8.                              new WaterSensor("sensor_2", 3000L, 30),
  9.                              new WaterSensor("sensor_1", 4000L, 40),
  10.                              new WaterSensor("sensor_1", 5000L, 50),
  11.                              new WaterSensor("sensor_2", 6000L, 60));
  13.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  14.         // 使用sql查询一个已注册的表
  15.         // 1. 从流得到一个表
  16.         Table inputTable = tableEnv.fromDataStream(waterSensorStream);
  17.         // 2. 把注册为一个临时视图
  18.         tableEnv.createTemporaryView("sensor", inputTable);
  19.         // 3. 在临时视图查询数据, 并得到一个新表
  20.         Table resultTable = tableEnv.sqlQuery("select * from sensor where id='sensor_1'");
  21.         // 4. 显示resultTable的数据
  22.         tableEnv.toAppendStream(resultTable, Row.class).print();
  23.         env.execute();
  24.     }
  25. }

_

Kafka到Kafka

使用sql从Kafka读数据, 并写入到Kafka中
如果注册了临时表，就可以使用临时表来查询。

  1. public class Flink05_SQL_Kafka2Kafka {
  2.     public static void main(String[] args) throws Exception {
  3.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  4.         env.setParallelism(1);
  5.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
  7.         // 1. 注册SourceTable: source_sensor
  8.         tableEnv.executeSql("create table source_sensor (id string, ts bigint, vc int) with("
  9.                                 + "'connector' = 'kafka',"
  10.                                 + "'topic' = 'topic_source_sensor',"
  11.                                 + "'properties.bootstrap.servers' = 'hadoop162:9029,hadoop163:9092,hadoop164:9092',"
  12.                                 + "'properties.group.id' = 'atguigu',"
  13.                                 + "'scan.startup.mode' = 'latest-offset',"
  14.                                 + "'format' = 'json'"
  15.                                 + ")");
  17.         //中间表表好像找不到注册表的方法
  18.         // Table table = ten.sqlQuery("select * from sensor")
  19.         // 2. 注册SinkTable: sink_sensor
  20.         tableEnv.executeSql("create table sink_sensor(id string, ts bigint, vc int) with("
  21.                                 + "'connector' = 'kafka',"
  22.                                 + "'topic' = 'topic_sink_sensor',"
  23.                                 + "'properties.bootstrap.servers' = 'hadoop162:9029,hadoop163:9092,hadoop164:9092',"
  24.                                 + "'format' = 'json'"
  25.                                 + ")");
  27.         // 3. 从SourceTable 查询数据, 并写入到 SinkTable
  28. // 不注册：
  29.         //tenv.executeSql("insert into s2 select * from + table）
  30.         tableEnv.executeSql("insert into sink_sensor select * from source_sensor where id='sensor_1'");
  31.     }
  32. }