使用window 例如五分钟一个固定窗口
如果save任务后 过几个小时再去追数据，重启的时候source会大量涌进来数据（暂停的几个小时数据有很多），
在这种情况会不会使窗口提前关闭（后面的数据先进来了），导致很多本该计算在这个个窗口的数据 没有计算进来，最后结果的错误，这种情况如何解决？

2.1 Flink Checkpoint-轻量级分布式快照 找不到视频

3.12 30分钟教你如何入门 – Apache Flink 最全知识图谱详解的第一个链接格式错了
会导致直接打开无法看到内容

如题

flink aliyun安装报错 缺少 kafka-schema-registry-client-3.3.1 需要本地手动安装才可编译成功，希望文档中可以提示体现下

https://github.com/flink-china/flink-training-course/blob/master/%E8%AF%BE%E7%A8%8B%E8%A1%A8%EF%BC%88%E6%8C%81%E7%BB%AD%E6%9B%B4%E6%96%B0%EF%BC%89.md

我按照官网上的防欺诈案例写了个项目，gradle的，通过WebUI提交job之后报以下错误：

org.apache.flink.streaming.runtime.tasks.StreamTaskException: Could not instantiate outputs in order. at org.apache.flink.streaming.api.graph.StreamConfig.getOutEdgesInOrder(StreamConfig.java:429) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1150) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriterDelegate(StreamTask.java:1134) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:284) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:271) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:73) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.runtime.tasks.SourceStreamTask.<init>(SourceStreamTask.java:69) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method) ~[?:?] at jdk.internal.reflect.NativeConstructorAccessorImpl.newInstance(Unknown Source) ~[?:?] at jdk.internal.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source) ~[?:?] at java.lang.reflect.Constructor.newInstance(Unknown Source) ~[?:?] at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1372) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:699) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.runtime.taskmanager.Task.run(Task.java:546) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at java.lang.Thread.run(Unknown Source) ~[?:?] Caused by: java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner.keySelector of type org.apache.flink.api.java.functions.KeySelector in instance of org.apache.flink.streaming.runtime.partitioner.KeyGroupStreamPartitioner at java.io.ObjectStreamClass$FieldReflector.setObjFieldValues(Unknown Source) ~[?:?] at java.io.ObjectStreamClass$FieldReflector.checkObjectFieldValueTypes(Unknown Source) ~[?:?] at java.io.ObjectStreamClass.checkObjFieldValueTypes(Unknown Source) ~[?:?] at java.io.ObjectInputStream.defaultCheckFieldValues(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?] at java.io.ObjectInputStream.defaultReadFields(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?] at java.util.ArrayList.readObject(Unknown Source) ~[?:?] at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?] at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(Unknown Source) ~[?:?] at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source) ~[?:?] at java.lang.reflect.Method.invoke(Unknown Source) ~[?:?] at java.io.ObjectStreamClass.invokeReadObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readSerialData(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readOrdinaryObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject0(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?] at java.io.ObjectInputStream.readObject(Unknown Source) ~[?:?] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:576) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:562) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.util.InstantiationUtil.deserializeObject(InstantiationUtil.java:550) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.util.InstantiationUtil.readObjectFromConfig(InstantiationUtil.java:511) ~[flink-dist_2.12-1.11.3.jar:1.11.3] at org.apache.flink.streaming.api.graph.StreamConfig.getOutEdgesInOrder(StreamConfig.java:426) ~[flink-dist_2.12-1.11.3.jar:1.11.3] ... 14 more

我应该怎么解决呢？
以下是我的java代码：

FraudDetector.java:

`package com.hcg.flink.frauddetection.detector;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;

public class FraudDetector extends KeyedProcessFunction<Long, Transaction, Alert> {
private static final double SMALL_AMOUNT = 1.00;
private static final double LARGE_AMOUNT = 500.00;
private static final long ONE_MINUTE = 60 * 1000;

private transient ValueState<Boolean> flagState;
private transient ValueState<Long> timerState;

@Override
public void open(Configuration parameters) throws Exception {
    ValueStateDescriptor<Boolean> flagDescriptor = new ValueStateDescriptor<>("flag", Types.BOOLEAN);
    flagState = getRuntimeContext().getState(flagDescriptor);

    ValueStateDescriptor<Long> timerDescriptor = new ValueStateDescriptor<>("timer", Types.LONG);
    timerState = getRuntimeContext().getState(timerDescriptor);
}

@Override
public void processElement(Transaction transaction,
                           Context context,
                           Collector<Alert> collector) throws Exception {
    // Get the current state for the current key
    Boolean lastTransactionWasSmall = flagState.value();

    if (lastTransactionWasSmall != null && lastTransactionWasSmall) {
        if (transaction.getAmount() > LARGE_AMOUNT) {
            Alert alert = new Alert();
            alert.setId(transaction.getAccountId());
            collector.collect(alert);
        }

        // Clean up our state
        cleanUp(context);
    }

    if (transaction.getAmount() < SMALL_AMOUNT) {
        flagState.update(true);

        // set the timer and timer state
        long timer = context.timerService().currentProcessingTime() + ONE_MINUTE;
        context.timerService().registerProcessingTimeTimer(timer);
        timerState.update(timer);
    }
}

@Override
public void onTimer(long timestamp, OnTimerContext ctx, Collector<Alert> out) {
    // remove flag after 1 minute
    timerState.clear();
    flagState.clear();
}

private void cleanUp(Context ctx) throws Exception {
    // delete timer
    Long timer = timerState.value();
    ctx.timerService().deleteProcessingTimeTimer(timer);

    // clean up all state
    timerState.clear();
    flagState.clear();
}

}
`
FraudDetectionJob.java:

`package com.hcg.flink.frauddetection;

import com.hcg.flink.frauddetection.detector.FraudDetector;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.walkthrough.common.entity.Alert;
import org.apache.flink.walkthrough.common.entity.Transaction;
import org.apache.flink.walkthrough.common.sink.AlertSink;
import org.apache.flink.walkthrough.common.source.TransactionSource;

@slf4j
public class FraudDetectionJob {

public static void main(String[] args) throws Exception {
    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

    DataStream<Transaction> transactions = env
            .addSource(new TransactionSource())
            .name("transactions");

    DataStream<Alert> alerts = transactions
            .keyBy(Transaction::getAccountId)
            .process(new FraudDetector())
            .name("fraud-detector");

    alerts
            .addSink(new AlertSink())
            .name("send-alerts");

    env.execute("Fraud Detection");
}

}
`
build.gradle:

`plugins {
id 'io.spring.dependency-management' version '1.0.10.RELEASE'
id 'java'
id 'application'
}

group = 'com.hcg'
version = '1.0-SNAPSHOT'
sourceCompatibility = '11'

configurations {
compileOnly {
extendsFrom annotationProcessor
}
}

repositories {
mavenCentral()
}

configurations {
// configuration that holds jars to include in the jar
extraLibs
}

dependencies {
compileOnly 'org.projectlombok:lombok:1.18.12'
annotationProcessor 'org.projectlombok:lombok:1.18.12'

implementation 'org.apache.flink:flink-clients_2.12:1.12.0'
implementation 'org.apache.flink:flink-streaming-java_2.12:1.12.0'
implementation 'org.apache.flink:flink-walkthrough-common_2.12:1.12.0'

}

application {
mainClass.set("com.hcg.flink.frauddetection.FraudDetectionJob")
}

jar {
manifest {
attributes('Main-Class': 'com.hcg.flink.frauddetection.FraudDetectionJob')
}
from {
configurations.extraLibs.collect { it.isDirectory() ? it : zipTree(it) }
}
}`

希望大神能够出一些列的源码解析啊，帮助我们更好地理解flink。

在曝光和点击进行interval join 时生成的数据是正样本，但是需要拿到曝光没有点击的数据。interval join 是否可以输出该数据，比如在数据淘汰的时候？
使用cogroup 的话，由于

public static long getWindowStartWithOffset(long timestamp, long offset, long windowSize) {
return timestamp - (timestamp - offset + windowSize) % windowSize;
}

曝光和点击可能位于不了一个window，这个情况怎么处理呢，非要借助外部存储进行记录吗

当前这个仓库是停止更新维护了吗？ 有新的仓库吗

如题，网上搜索了很久都没有相关回答，望大家能给我这个小萌新讲讲

我用WINDOWS从官网下载解压了最新的FLINK，但是在bin中没有.bat文件，只有.sh

我的下载的源码虚拟机老是编译不成功，希望能发一份编译后的源码[email protected]