[jira] [Created] (FLINK-20201) Support automatic adjustment of window parameters

[Shang Yuanchun (Jira)]

lqjacklee created FLINK-20201:
---------------------------------

             Summary: Support automatic adjustment of window parameters
                 Key: FLINK-20201
                 URL: https://issues.apache.org/jira/browse/FLINK-20201
             Project: Flink
          Issue Type: Wish
          Components: Runtime / Coordination
    Affects Versions: 1.11.2
            Reporter: lqjacklee



{code:java}

static Map<String, Field> fieldMap;

    static  {
        fieldMap = Stream.of(Entity.class.getDeclaredFields())
                .collect(Collectors.toMap(Field::getName, field -> field));
    }

    public static class Entity extends PojoTypeInfo<Entity> implements Comparable<Entity> {
        public String name;
        public long currentDate;
        public int purchaseVolume;

        public Entity(String name, int purchaseVolume) {
            super(Entity.class, Arrays.asList(new PojoField(fieldMap.get("name"), BasicTypeInfo.STRING_TYPE_INFO),
                    new PojoField(fieldMap.get("purchaseVolume"), BasicTypeInfo.INT_TYPE_INFO),
                    new PojoField(fieldMap.get("currentDate"), BasicTypeInfo.DATE_TYPE_INFO)
                    ));
            this.name = name;
            this.purchaseVolume = purchaseVolume;
            this.currentDate = System.currentTimeMillis();
        }

        @Override
        public int compareTo(Entity o) {
            return Long.compare(currentDate, o.currentDate);
        }

        public String getName() {
            return name;
        }

        public void setName(String name) {
            this.name = name;
        }

        public long getCurrentDate() {
            return currentDate;
        }

        public void setCurrentDate(long currentDate) {
            this.currentDate = currentDate;
        }

        public int getPurchaseVolume() {
            return purchaseVolume;
        }

        public void setPurchaseVolume(int purchaseVolume) {
            this.purchaseVolume = purchaseVolume;
        }
    }

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env
            .fromElements(new Entity("jack", 10), new Entity("tom", 20), new Entity("jack", 30))
            .keyBy((KeySelector<Entity, String>) value -> value.name, BasicTypeInfo.STRING_TYPE_INFO)
            .timeWindow(Time.milliseconds(1))
            .process(new ProcessWindowFunction<Entity, Integer, String, TimeWindow>() {
                @Override
                public void process(String s, Context context, Iterable<Entity> elements, Collector<Integer> out) throws Exception {
                         StreamSupport.stream(elements.spliterator(), false)
                            .map(Entity::getPurchaseVolume)
                            .iterator()
                            .forEachRemaining(out::collect);
                }
            })
            .print();

        env.execute("window aggregate");
    }

{code}



Because the speed of data generation changes greatly, we hope to dynamically adjust the size of the window according to different data volume.

The window parameter should be generated by machine learning. in that case, we can just provide the window time range to user to configure.





--
This message was sent by Atlassian Jira
(v8.3.4#803005)