Uploaded image for project: 'Flink'
  1. Flink
  2. FLINK-20201

Support automatic adjustment of window parameters



    • Type: Wish
    • Status: Open
    • Priority: Major
    • Resolution: Unresolved
    • Affects Version/s: 1.11.2
    • Fix Version/s: None
    • Component/s: API / DataStream
    • Labels:


      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();
              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();
                  .fromElements(new Entity("jack", 10), new Entity("tom", 20), new Entity("jack", 30))
                  .keyBy((KeySelector<Entity, String>) value -> value.name, BasicTypeInfo.STRING_TYPE_INFO)
                  .process(new ProcessWindowFunction<Entity, Integer, String, TimeWindow>() {
                      public void process(String s, Context context, Iterable<Entity> elements, Collector<Integer> out) throws Exception {
                               StreamSupport.stream(elements.spliterator(), false)
              env.execute("window aggregate");

      I provide the sample code to show how to use the window parameter in the Flink. we want to provide the function to adjust the size of the window parameter.

      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.




            • Assignee:
              Jack-Lee lqjacklee
            • Votes:
              0 Vote for this issue
              2 Start watching this issue


              • Created: