登录
当前位置： 首页 > 知识库问答 >
问题：

“键组不在键组范围内”根据 Flink 中的事件时间加入两个流时出错

庞元青
2023-03-14

我有一个简单的测试，可以在事件发生时连接两个流：

 @Test
void testJoinStream() {
    var settings = EnvironmentSettings
            .newInstance()
            .inStreamingMode()
            .build();
    var tableEnv = TableEnvironment.create(settings);
    var configuration = tableEnv.getConfig().getConfiguration();
    configuration.setString("table.exec.resource.default-parallelism", "2");

    var testTable = tableEnv.from(TableDescriptor.forConnector("datagen")
            .schema(Schema.newBuilder()
                    .column("ts", DataTypes.TIMESTAMP(3))
                    .column("v", DataTypes.INT())
                    .watermark("ts", "ts - INTERVAL '1' second")
                    .build())
            .option(DataGenConnectorOptions.ROWS_PER_SECOND, 2L)
            .option("fields.v.kind", "sequence")
            .option("fields.v.start", "0")
            .option("fields.v.end", "1000000")
            .build());
    testTable.printSchema();
    tableEnv.createTemporaryView("test", testTable );

    var joined = tableEnv.sqlQuery("SELECT ts, v, v2 from test" +
            " join (SELECT ts as ts2, v as v2 from test) on ts = ts2");

    try {
        var tableResult = joined.executeInsert(TableDescriptor.forConnector("print").build());
        tableResult.await();

    } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
    }
}

几秒钟内就失败了：

(
  `ts` TIMESTAMP(3) *ROWTIME*,
  `v` INT,
  WATERMARK FOR `ts`: TIMESTAMP(3) AS ts - INTERVAL '1' second
)
1> +I[2021-11-16T17:48:24.415, 1, 1]
1> +I[2021-11-16T17:48:24.415, 0, 1]
1> +I[2021-11-16T17:48:24.415, 1, 0]
1> +I[2021-11-16T17:48:24.415, 0, 0]

java.util.concurrent.ExecutionException: org.apache.flink.table.api.TableException: Failed to wait job finish
java.lang.RuntimeException: java.util.concurrent.ExecutionException: org.apache.flink.table.api.TableException: Failed to wait job finish
    at com.microstrategy.realtime.FlinkTest.testJoinStream(FlinkTest.java:123)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.junit.platform.commons.util.ReflectionUtils.invokeMethod(ReflectionUtils.java:688)
    at org.junit.jupiter.engine.execution.MethodInvocation.proceed(MethodInvocation.java:60)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain$ValidatingInvocation.proceed(InvocationInterceptorChain.java:131)
    at org.junit.jupiter.engine.extension.TimeoutExtension.intercept(TimeoutExtension.java:149)
    at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestableMethod(TimeoutExtension.java:140)
    at org.junit.jupiter.engine.extension.TimeoutExtension.interceptTestMethod(TimeoutExtension.java:84)
    at org.junit.jupiter.engine.execution.ExecutableInvoker$ReflectiveInterceptorCall.lambda$ofVoidMethod$0(ExecutableInvoker.java:115)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.lambda$invoke$0(ExecutableInvoker.java:105)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain$InterceptedInvocation.proceed(InvocationInterceptorChain.java:106)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.proceed(InvocationInterceptorChain.java:64)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.chainAndInvoke(InvocationInterceptorChain.java:45)
    at org.junit.jupiter.engine.execution.InvocationInterceptorChain.invoke(InvocationInterceptorChain.java:37)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:104)
    at org.junit.jupiter.engine.execution.ExecutableInvoker.invoke(ExecutableInvoker.java:98)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeTestMethod$6(TestMethodTestDescriptor.java:210)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.invokeTestMethod(TestMethodTestDescriptor.java:206)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:131)
    at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.execute(TestMethodTestDescriptor.java:65)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:139)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at java.base/java.util.ArrayList.forEach(ArrayList.java:1541)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.invokeAll(SameThreadHierarchicalTestExecutorService.java:38)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$5(NodeTestTask.java:143)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$7(NodeTestTask.java:129)
    at org.junit.platform.engine.support.hierarchical.Node.around(Node.java:137)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.lambda$executeRecursively$8(NodeTestTask.java:127)
    at org.junit.platform.engine.support.hierarchical.ThrowableCollector.execute(ThrowableCollector.java:73)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.executeRecursively(NodeTestTask.java:126)
    at org.junit.platform.engine.support.hierarchical.NodeTestTask.execute(NodeTestTask.java:84)
    at org.junit.platform.engine.support.hierarchical.SameThreadHierarchicalTestExecutorService.submit(SameThreadHierarchicalTestExecutorService.java:32)
    at org.junit.platform.engine.support.hierarchical.HierarchicalTestExecutor.execute(HierarchicalTestExecutor.java:57)
    at org.junit.platform.engine.support.hierarchical.HierarchicalTestEngine.execute(HierarchicalTestEngine.java:51)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:108)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:88)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.lambda$execute$0(EngineExecutionOrchestrator.java:54)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.withInterceptedStreams(EngineExecutionOrchestrator.java:67)
    at org.junit.platform.launcher.core.EngineExecutionOrchestrator.execute(EngineExecutionOrchestrator.java:52)
    at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:96)
    at org.junit.platform.launcher.core.DefaultLauncher.execute(DefaultLauncher.java:75)
    at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.processAllTestClasses(JUnitPlatformTestClassProcessor.java:99)
    at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor$CollectAllTestClassesExecutor.access$000(JUnitPlatformTestClassProcessor.java:79)
    at org.gradle.api.internal.tasks.testing.junitplatform.JUnitPlatformTestClassProcessor.stop(JUnitPlatformTestClassProcessor.java:75)
    at org.gradle.api.internal.tasks.testing.SuiteTestClassProcessor.stop(SuiteTestClassProcessor.java:61)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base/java.lang.reflect.Method.invoke(Method.java:566)
    at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:36)
    at org.gradle.internal.dispatch.ReflectionDispatch.dispatch(ReflectionDispatch.java:24)
    at org.gradle.internal.dispatch.ContextClassLoaderDispatch.dispatch(ContextClassLoaderDispatch.java:33)
    at org.gradle.internal.dispatch.ProxyDispatchAdapter$DispatchingInvocationHandler.invoke(ProxyDispatchAdapter.java:94)
    at com.sun.proxy.$Proxy2.stop(Unknown Source)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker$3.run(TestWorker.java:193)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.executeAndMaintainThreadName(TestWorker.java:129)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:100)
    at org.gradle.api.internal.tasks.testing.worker.TestWorker.execute(TestWorker.java:60)
    at org.gradle.process.internal.worker.child.ActionExecutionWorker.execute(ActionExecutionWorker.java:56)
    at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:133)
    at org.gradle.process.internal.worker.child.SystemApplicationClassLoaderWorker.call(SystemApplicationClassLoaderWorker.java:71)
    at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
    at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.table.api.TableException: Failed to wait job finish
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
    at org.apache.flink.table.api.internal.TableResultImpl.awaitInternal(TableResultImpl.java:129)
    at org.apache.flink.table.api.internal.TableResultImpl.await(TableResultImpl.java:92)
    at com.microstrategy.realtime.FlinkTest.testJoinStream(FlinkTest.java:120)
    ... 83 more
Caused by: org.apache.flink.table.api.TableException: Failed to wait job finish
    at app//org.apache.flink.table.api.internal.InsertResultIterator.hasNext(InsertResultIterator.java:56)
    at app//org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.hasNext(TableResultImpl.java:370)
    at app//org.apache.flink.table.api.internal.TableResultImpl$CloseableRowIteratorWrapper.isFirstRowReady(TableResultImpl.java:383)
    at app//org.apache.flink.table.api.internal.TableResultImpl.lambda$awaitInternal$1(TableResultImpl.java:116)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture$AsyncRun.run(CompletableFuture.java:1736)
    at java.base@11.0.11/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base@11.0.11/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base@11.0.11/java.lang.Thread.run(Thread.java:829)
Caused by: java.util.concurrent.ExecutionException: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
    at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
    at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
    at org.apache.flink.table.api.internal.InsertResultIterator.hasNext(InsertResultIterator.java:54)
    ... 7 more
Caused by: org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
    at app//org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:144)
    at app//org.apache.flink.runtime.minicluster.MiniClusterJobClient.lambda$getJobExecutionResult$3(MiniClusterJobClient.java:137)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture$UniApply.tryFire(CompletableFuture.java:642)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
    at org.apache.flink.runtime.rpc.akka.AkkaInvocationHandler.lambda$invokeRpc$0(AkkaInvocationHandler.java:250)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
    at app//org.apache.flink.util.concurrent.FutureUtils.doForward(FutureUtils.java:1389)
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$null$1(ClassLoadingUtils.java:93)
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:68)
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.lambda$guardCompletionWithContextClassLoader$2(ClassLoadingUtils.java:92)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.uniWhenComplete(CompletableFuture.java:859)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture$UniWhenComplete.tryFire(CompletableFuture.java:837)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.postComplete(CompletableFuture.java:506)
    at java.base@11.0.11/java.util.concurrent.CompletableFuture.complete(CompletableFuture.java:2073)
    at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$1.onComplete(AkkaFutureUtils.java:47)
    at akka.dispatch.OnComplete.internal(Future.scala:300)
    at akka.dispatch.OnComplete.internal(Future.scala:297)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:224)
    at akka.dispatch.japi$CallbackBridge.apply(Future.scala:221)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
    at org.apache.flink.runtime.concurrent.akka.AkkaFutureUtils$DirectExecutionContext.execute(AkkaFutureUtils.java:65)
    at scala.concurrent.impl.CallbackRunnable.executeWithValue(Promise.scala:68)
    at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1(Promise.scala:284)
    at scala.concurrent.impl.Promise$DefaultPromise.$anonfun$tryComplete$1$adapted(Promise.scala:284)
    at scala.concurrent.impl.Promise$DefaultPromise.tryComplete(Promise.scala:284)
    at akka.pattern.PromiseActorRef.$bang(AskSupport.scala:621)
    at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:24)
    at akka.pattern.PipeToSupport$PipeableFuture$$anonfun$pipeTo$1.applyOrElse(PipeToSupport.scala:23)
    at scala.concurrent.Future.$anonfun$andThen$1(Future.scala:532)
    at scala.concurrent.impl.Promise.liftedTree1$1(Promise.scala:29)
    at scala.concurrent.impl.Promise.$anonfun$transform$1(Promise.scala:29)
    at scala.concurrent.impl.CallbackRunnable.run(Promise.scala:60)
    at akka.dispatch.BatchingExecutor$AbstractBatch.processBatch(BatchingExecutor.scala:63)
    at akka.dispatch.BatchingExecutor$BlockableBatch.$anonfun$run$1(BatchingExecutor.scala:100)
    at scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:12)
    at scala.concurrent.BlockContext$.withBlockContext(BlockContext.scala:81)
    at akka.dispatch.BatchingExecutor$BlockableBatch.run(BatchingExecutor.scala:100)
    at akka.dispatch.TaskInvocation.run(AbstractDispatcher.scala:49)
    at akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(ForkJoinExecutorConfigurator.scala:48)
    at java.base@11.0.11/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.base@11.0.11/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.base@11.0.11/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.base@11.0.11/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.base@11.0.11/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)
Caused by: org.apache.flink.runtime.JobException: Recovery is suppressed by NoRestartBackoffTimeStrategy
    at app//org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.handleFailure(ExecutionFailureHandler.java:138)
    at app//org.apache.flink.runtime.executiongraph.failover.flip1.ExecutionFailureHandler.getFailureHandlingResult(ExecutionFailureHandler.java:82)
    at app//org.apache.flink.runtime.scheduler.DefaultScheduler.handleTaskFailure(DefaultScheduler.java:228)
    at app//org.apache.flink.runtime.scheduler.DefaultScheduler.maybeHandleTaskFailure(DefaultScheduler.java:218)
    at app//org.apache.flink.runtime.scheduler.DefaultScheduler.updateTaskExecutionStateInternal(DefaultScheduler.java:209)
    at app//org.apache.flink.runtime.scheduler.SchedulerBase.updateTaskExecutionState(SchedulerBase.java:679)
    at app//org.apache.flink.runtime.scheduler.SchedulerNG.updateTaskExecutionState(SchedulerNG.java:79)
    at app//org.apache.flink.runtime.jobmaster.JobMaster.updateTaskExecutionState(JobMaster.java:444)
    at java.base@11.0.11/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at java.base@11.0.11/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at java.base@11.0.11/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.base@11.0.11/java.lang.reflect.Method.invoke(Method.java:566)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.lambda$handleRpcInvocation$1(AkkaRpcActor.java:316)
    at org.apache.flink.runtime.concurrent.akka.ClassLoadingUtils.runWithContextClassLoader(ClassLoadingUtils.java:83)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcInvocation(AkkaRpcActor.java:314)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleRpcMessage(AkkaRpcActor.java:217)
    at org.apache.flink.runtime.rpc.akka.FencedAkkaRpcActor.handleRpcMessage(FencedAkkaRpcActor.java:78)
    at org.apache.flink.runtime.rpc.akka.AkkaRpcActor.handleMessage(AkkaRpcActor.java:163)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:24)
    at akka.japi.pf.UnitCaseStatement.apply(CaseStatements.scala:20)
    at scala.PartialFunction.applyOrElse(PartialFunction.scala:123)
    at scala.PartialFunction.applyOrElse$(PartialFunction.scala:122)
    at akka.japi.pf.UnitCaseStatement.applyOrElse(CaseStatements.scala:20)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:171)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
    at scala.PartialFunction$OrElse.applyOrElse(PartialFunction.scala:172)
    at akka.actor.Actor.aroundReceive(Actor.scala:537)
    at akka.actor.Actor.aroundReceive$(Actor.scala:535)
    at akka.actor.AbstractActor.aroundReceive(AbstractActor.scala:220)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
    at akka.actor.ActorCell.invoke(ActorCell.scala:548)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
    at akka.dispatch.Mailbox.run(Mailbox.scala:231)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
    ... 5 more
Caused by: java.lang.IllegalArgumentException: Key group 11 is not in KeyGroupRange{startKeyGroup=64, endKeyGroup=127}. Unless you're directly using low level state access APIs, this is most likely caused by non-deterministic shuffle key (hashCode and equals implementation).
    at org.apache.flink.runtime.state.KeyGroupRangeOffsets.newIllegalKeyGroupException(KeyGroupRangeOffsets.java:37)
    at org.apache.flink.runtime.state.heap.StateTable.getMapForKeyGroup(StateTable.java:305)
    at org.apache.flink.runtime.state.heap.StateTable.get(StateTable.java:261)
    at org.apache.flink.runtime.state.heap.StateTable.get(StateTable.java:143)
    at org.apache.flink.runtime.state.heap.HeapMapState.iterator(HeapMapState.java:161)
    at org.apache.flink.runtime.state.UserFacingMapState.iterator(UserFacingMapState.java:95)
    at org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoin.processElement1(TimeIntervalJoin.java:171)
    at org.apache.flink.table.runtime.operators.join.interval.RowTimeIntervalJoin.processElement1(RowTimeIntervalJoin.java:26)
    at org.apache.flink.table.runtime.operators.join.interval.TimeIntervalJoin.processElement1(TimeIntervalJoin.java:51)
    at org.apache.flink.streaming.api.operators.co.KeyedCoProcessOperator.processElement1(KeyedCoProcessOperator.java:78)
    at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.processRecord1(StreamTwoInputProcessorFactory.java:217)
    at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory.lambda$create$0(StreamTwoInputProcessorFactory.java:183)
    at org.apache.flink.streaming.runtime.io.StreamTwoInputProcessorFactory$StreamTaskNetworkOutput.emitRecord(StreamTwoInputProcessorFactory.java:266)
    at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.processElement(AbstractStreamTaskNetworkInput.java:134)
    at org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:105)
    at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
    at org.apache.flink.streaming.runtime.io.StreamMultipleInputProcessor.processInput(StreamMultipleInputProcessor.java:86)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:496)
    at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:809)
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:761)
    at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:958)
    at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:937)
    at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:766)
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:575)
    at java.base/java.lang.Thread.run(Thread.java:829)

问题出在哪里？错误消息不直观，我找不到导致问题的原因。

我看到的例外是由https://issues.apache.org/jira/browse/FLINK-18637等人报告的。但那些都是用数据流API。我把这个登记为https://issues.apache.org/jira/browse/FLINK-24926

已更新：查询计划

== Abstract Syntax Tree ==
LogicalProject(ts=[$0], v=[$1], v2=[$3])
+- LogicalJoin(condition=[=($0, $2)], joinType=[inner])
   :- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($0, 1000:INTERVAL SECOND)])
   :  +- LogicalTableScan(table=[[default_catalog, default_database, Unnamed_TableDescriptor$1]])
   +- LogicalProject(ts2=[$0], v2=[$1])
      +- LogicalWatermarkAssigner(rowtime=[ts], watermark=[-($0, 1000:INTERVAL SECOND)])
         +- LogicalTableScan(table=[[default_catalog, default_database, Unnamed_TableDescriptor$1]])

== Optimized Physical Plan ==
Calc(select=[ts, v, v0 AS v2])
+- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=true, leftLowerBound=0, leftUpperBound=0, leftTimeIndex=0, rightTimeIndex=0], where=[=(ts, ts0)], select=[ts, v, ts0, v0])
   :- Exchange(distribution=[hash[ts]])
   :  +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
   :     +- TableSourceScan(table=[[default_catalog, default_database, Unnamed_TableDescriptor$1]], fields=[ts, v])
   +- Exchange(distribution=[hash[ts]])
      +- WatermarkAssigner(rowtime=[ts], watermark=[-(ts, 1000:INTERVAL SECOND)])
         +- TableSourceScan(table=[[default_catalog, default_database, Unnamed_TableDescriptor$1]], fields=[ts, v])

== Optimized Execution Plan ==
Calc(select=[ts, v, v0 AS v2])
+- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=true, leftLowerBound=0, leftUpperBound=0, leftTimeIndex=0, rightTimeIndex=0], where=[(ts = ts0)], select=[ts, v, ts0, v0])
   :- Exchange(distribution=[hash[ts]])(reuse_id=[1])
   :  +- WatermarkAssigner(rowtime=[ts], watermark=[(ts - 1000:INTERVAL SECOND)])
   :     +- TableSourceScan(table=[[default_catalog, default_database, Unnamed_TableDescriptor$1]], fields=[ts, v])
   +- Reused(reference_id=[1])

共有2个答案

洪和平
2023-03-14

这是一个将在Flink 1.15中修复的错误。请参阅https://issues.apache.org/jira/browse/FLINK-24926和https://issues.apache.org/jira/browse/FLINK-19792.

符渊
2023-03-14

我在Flink 1.14.0上试过这个，我让它运行了几分钟；它工作得很好。在1.14中，解释：

EXPLAIN select ts, v, v2 from test join (select ts as ts2, v as v2 from test) on ts = ts2

== Optimized Execution Plan ==
Calc(select=[ts, v, v0 AS v2])
+- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=true, leftLowerBound=0, leftUpperBound=0, leftTimeIndex=0, rightTimeIndex=0], where=[(ts = ts0)], select=[ts, v, ts0, v0])
   :- Exchange(distribution=[hash[ts]])(reuse_id=[1])
   :  +- WatermarkAssigner(rowtime=[ts], watermark=[(ts - 1000:INTERVAL SECOND)])
   :     +- TableSourceScan(table=[[vvp, default, test]], fields=[ts, v])
   +- Reused(reference_id=[1])

表明规划人员将其理解为间隔联接。这一点很重要：流连接很容易变得病态，除非运行时能够利用某种时间约束对其进行优化。

也许您使用的是早期版本的Flink SQL，它为这个查询生成了一个不同的计划？

 类似资料：
  • 根据键合并两个对象数组

    问题内容： 我有两个数组： 阵列1： 和数组2： 我需要基于合并这两个数组，并获得此： 我如何在不迭代低谷的情况下做到这一点？ 问题答案： 你可以这样- 如果和的顺序不同，请使用以下代码： 如果和以相同的顺序使用

  • 根据php中的键分组数组值？

    问题内容： 我有一个像这样的数组： 我想基于“否”作为主键对数组中的元素进行分组。输出应如下所示： 请注意，作者的值是与主键``否’‘合并的，请问有人可以帮我吗？ 我尝试这样做： 问题答案： 您可以使用通用函数： 我添加了一些示例代码进行测试

  • 基于事件时间的Flink时间窗口不输出任何内容

    在流处理问题中，我们有3个传感器，每个传感器每8毫秒生成一个时间戳样本（传感器的时间是同步的）。所以我想合并每个时间戳的数据（对于3个传感器，我们应该为每个时间戳输出3个合并的样本数据）。此外，我们有一个160毫秒的时间限制，这样每个数据在生成时间戳后最多应该在160毫秒后输出。所以我决定使用Flink事件时间概念和时间窗口。因为时间戳在每个传感器的样本中都是唯一的，所以我们认为它是数据流的关键。

  • 每小时获取一个时间范围内的时间

    问题内容： 因此，我试图生成的是特定时间范围内的所有小时。 因此，鉴于范围从上午11点到下午2:00，我将得到： 我试图避免必须在商店每隔特定的小时存储一次，而只存储范围（我需要将小时数与其他时间进行比较） 谢谢 问题答案： 如果您有数字表（如果没有，请单击链接以创建一个表）… 如果这是专门的，则可以创建仅包含24个值的小时表。

  • 按键分组时Spark内存不足

    我试图使用Spark主机在EC2上使用本指南对常见爬网数据执行简单转换，我的代码如下所示：

  • 合并具有重叠时间范围的时间范围元组的列表

    问题内容： 我有一个元组列表，每个元组都是一个。我正在尝试合并所有重叠的时间范围，并返回不同时间范围的列表。例如 这是我的实现方法。 我想弄清楚是否 是某些python模块中的内置函数可以更有效地做到这一点吗？要么 有没有达到相同目标的更Python方式？ 感谢您的帮助。谢谢！ 问题答案： 使用Pythonic可以提高效率的几种方法： 消除了构造，因为该算法应在主循环中删除重复项。 如果只需要遍历

相关问答

根据多个键的比较合并两个哈希数组React组件中具有相同键的两个子组件根据键型事件的类型禁用键型事件thinkphp5 怎么根据时间范围查询？Apache Flink-事件时间

相关文章

Java系统范围内的键盘快捷键Flink键控流键为空获取Java范围内的键的值MySQL在给定的日期时间范围内插入随机的日期时间实施一个地图，其中键是一组非重叠范围

相关阅读

使用scanf读取从键盘输入的数据Cassandra创建键空间（Keyspace）键盘快捷键Ruby 范围（Range）JavaScript之事件流与事件处理程序（事件）

相关工具

开源数据访问组件DAC数字键盘添加“完成”按钮tapestry组件库eZ 组件库右键助手

相关文档

Element UI 组件中文文档Porter 数据同步中间件elasticsearch-rest-plus 组件MOYE 知心组件库中文文档Element React 组件帮助中文文档
快捷导航： 新手教程 算法原理 架构设计 Java进阶 数据库进阶 大厂专栏 面试经验 编程笔记 编程问答 所有专题 文档资料 工具软件 电子书籍 小牛导航
在线工具： 房贷计算器 个税计算器 Linux命令查询 Json格式化 正则表达式 颜色转换 AES加解密 SHA1加密 MD5加密 毒鸡汤 字数统计 随机密码生成 进制转换 Base64编解码 励志句子
Copyright © 2019-2024 小牛知识库@xnip.cn. All Rights Reserved.