使用kafka源和使用Flink sql查询的Flink流表

Question

我试图将数据从kafka主题读入DataStream并注册DataStream，然后使用tableEnvironment.sqlQuery（“sql”）查询数据，当tableEnvironment.execute（）没有错误也没有输出时。

public static void main(String[] args){
   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   env.setStreamTimeCharacteristic(TimeCharateristic.EventTime);
   env.enableCheckpointing(5000);
   StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
   FlinkKafkaConsumer<Person> consumer = new FlinkKafkaConsumer(
                                                                "topic",
                                                                 new JSONDeserializer(),
                                                                 Job.getKafkaProperties
                                                               );

  consumer.setStartFromEarliest();
  DataStream<Person> stream = env.addSource(consumer).fliter(x -> x.status != -1).assignTimestampAndWatermarks(new AssignerWithPeriodicWatermarks<Person>(){
      long current = 0L;
      final long expire = 1000L;

      @Override
      public Watermakr getCurrentWatermark(){
         return new Watermark(current - expire);
      }

      @Override
      public long extractTimestamp(Person person){
         long timestamp = person.createTime;
         current = Math.max(timestamp,current);
         return timestamp;
      }
  });
  //set createTime as rowtime
  tableEnvironment.registerDataStream("Table_Person",stream,"name,age,sex,createTime.rowtime");
  Table res = tableEnvironment.sqlQuery("select TUMBLE_END(createTime,INTERVAL '1' minute) as registTime,sex,count(1) as total from Table_Person group by sex,TUMBLE(createTime,INTERVAL '1' minute)");
  tableEnvironment.toAppendStream(t,Types.Row(new TypeInformation[]{Types.SQL_TIMESTAMP,Types.STRING,Types.LONG})).print();
tableEnvironment.execute("person-query");
}

依赖关系：

1. flink-streaming-java2.11版本:1.9.0-csa1.0.0.0；
2. Flink-Streaming-Scala2.11版本:1.9.0-CSA1.0.0.0；
3. flink-connector-kafka2.11版本:1.9.0-csa1.0.0.0；
4. flink-table-api-java-bridge2.11版本:1.9.0-csa1.0.0.0；
5. flink-table-planner2.11版本:1.9.0-csa1.0.0.0；

Answer 1

在将SQL查询结果转换回DataStream的代码中，需要将res而不是t传递给ToAppendStream。（我不知道您发布的代码将如何编译--t在哪里声明？）我想你应该能做到

Table res = tableEnvironment.sqlQuery("select TUMBLE_END(createTime,INTERVAL '1' minute) as registTime,sex,count(1) as total from Table_Person group by sex,TUMBLE(createTime,INTERVAL '1' minute)");
tableEnvironment.toAppendStream(res,Row.class).print();

而不是麻烦于typeinformation。