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

Support automatic adjustment of window parameters

    XMLWordPrintableJSON

    Details

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

      Description

      
      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");
          }
      
      

      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.

        Attachments

          Activity

            People

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

              Dates

              • Created:
                Updated: