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)
