欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

flink-table-sql-demo2

程序员文章站 2022-06-07 23:14:17
...

一.背景

     这个和demo1类似,只是提供另外一种实现方式,类似kafka 利用streamTableSource 来做。

 

二.代码

      

@Data
@ToString
public class UserInfo implements Serializable {
    private Timestamp pTime;
    private String userId;
    private String itemId;

    public UserInfo() {
    }

    public UserInfo(String userId, String itemId) {
        this.userId = userId;
        this.itemId = itemId;
        this.pTime = new Timestamp(System.currentTimeMillis());
    }
}

 

 

public class UserTableSource implements StreamTableSource<UserInfo>, DefinedRowtimeAttributes {
    /**
     * 返回类型
     * @return
     */
    @Override
    public TypeInformation<UserInfo> getReturnType() {
        return TypeInformation.of(UserInfo.class);
    }

    @Override
    public TableSchema getTableSchema() {
          // 可以 这样定义
//        TableSchema schema = new TableSchema(
//                new String[]{"pTime","userId","itemId"},
//                new TypeInformation[]{Types.SQL_TIMESTAMP,Types.STRING,Types.STRING});
        return TableSchema.fromTypeInfo(getReturnType());
    }



    @Override
    public String explainSource() {
        return "userSource";
    }


    @Override
    public DataStream<UserInfo> getDataStream(StreamExecutionEnvironment execEnv) {
        UserDataSource source = new UserDataSource();
        DataStream<UserInfo> userInfoDataStream = execEnv.addSource(source);
        return userInfoDataStream;
    }

    @Override
    public List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors() {
        RowtimeAttributeDescriptor descriptor =
                new RowtimeAttributeDescriptor("pTime",
                new ExistingField("pTime"),
                new AscendingTimestamps());
        return Collections.singletonList(descriptor);
    }


}

 

 

public class UserStreamTableApp {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // create a TableEnvironment
        StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);

        tEnv.registerTableSource("test", new UserTableSource());


        Table result = tEnv.sqlQuery("SELECT userId,TUMBLE_END(pTime, INTERVAL '5' SECOND) as pTime,count(1) as cnt FROM  test" +
                " GROUP BY TUMBLE(pTime, INTERVAL '5' SECOND),userId ");
        // deal with (Tuple2<Boolean, Row> value) -> out.collect(row)
        SingleOutputStreamOperator allClick = tEnv.toRetractStream(result, Row.class)
                .flatMap((Tuple2<Boolean, Row> value, Collector<Row> out) -> {
                    out.collect(value.f1);
                }).returns(Row.class);
        // add sink or print
        allClick.print();
        env.execute("test");

    }

}