I agree with Roman that it seems that the Python process has crashed.
Besides the suggestions from Roman, I guess you could also try to configure the
bundle size to smaller value via “python.fn-execution.bundle.size”.
Regards,
Dian
> 2021年9月24日 上午3:48，Roman Khachatryan <ro...@apache.org> 写道：
>
> Hi,
>
> Is it possible that the python process crashed or hung up? (probably
> performing a snapshot)
> Could you validate this by checking the OS logs for OOM killer
> messages or process status?
>
> Regards,
> Roman
>
> On Wed, Sep 22, 2021 at 6:30 PM Curt Buechter <tricksho...@gmail.com> wrote:
>>
>> Hi,
>> I'm getting an error after enabling checkpointing in my pyflink application
>> that uses a keyed stream and rocksdb state.
>>
>> Here is the error message:
>>
>> 2021-09-22 16:18:14,408 INFO
>> org.apache.flink.contrib.streaming.state.RocksDBKeyedStateBackend [] -
>> Closed RocksDB State Backend. Cleaning up RocksDB working directory
>> /mnt/yarn/usercache/hadoop/appcache/application_1631886817823_0058/flink-io-db923140-fce6-433e-b26b-b7d53afbc38f/job_383b888cc0d49bbd7464906e45b88187_op_PythonKeyedProcessOperator_90bea66de1c231edf33913ecd54406c1__1_1__uuid_16eefb37-8bae-43ef-b680-d6a4f9660c39.
>> 2021-09-22 16:18:14,409 WARN org.apache.flink.runtime.taskmanager.Task [] -
>> KEYED PROCESS -> Sink: Unnamed (1/1)#34 (8f4fd40e863dd058822060dc3cf98831)
>> switched from RUNNING to FAILED with failure cause: java.io.IOException:
>> Could not perform checkpoint 2 for operator KEYED PROCESS -> Sink: Unnamed
>> (1/1)#34.
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1045)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.CheckpointBarrierHandler.notifyCheckpoint(CheckpointBarrierHandler.java:135)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.triggerCheckpoint(SingleCheckpointBarrierHandler.java:250)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.access$100(SingleCheckpointBarrierHandler.java:61)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler$ControllerImpl.triggerGlobalCheckpoint(SingleCheckpointBarrierHandler.java:431)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.AbstractAlignedBarrierHandlerState.barrierReceived(AbstractAlignedBarrierHandlerState.java:61)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.SingleCheckpointBarrierHandler.processBarrier(SingleCheckpointBarrierHandler.java:227)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.handleEvent(CheckpointedInputGate.java:180)
>> at
>> org.apache.flink.streaming.runtime.io.checkpointing.CheckpointedInputGate.pollNext(CheckpointedInputGate.java:158)
>> at
>> org.apache.flink.streaming.runtime.io.AbstractStreamTaskNetworkInput.emitNext(AbstractStreamTaskNetworkInput.java:110)
>> at
>> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:66)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:423)
>> at
>> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:204)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:681)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.executeInvoke(StreamTask.java:636)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.runWithCleanUpOnFail(StreamTask.java:647)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:620)
>> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:779)
>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:566)
>> at java.lang.Thread.run(Thread.java:748)
>> Caused by: java.lang.RuntimeException: Failed to close remote bundle
>> at
>> org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.finishBundle(BeamPythonFunctionRunner.java:383)
>> at
>> org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.flush(BeamPythonFunctionRunner.java:331)
>> at
>> org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.invokeFinishBundle(AbstractPythonFunctionOperator.java:320)
>> at
>> org.apache.flink.streaming.api.operators.python.AbstractPythonFunctionOperator.prepareSnapshotPreBarrier(AbstractPythonFunctionOperator.java:175)
>> at
>> org.apache.flink.streaming.runtime.tasks.OperatorChain.prepareSnapshotPreBarrier(OperatorChain.java:415)
>> at
>> org.apache.flink.streaming.runtime.tasks.SubtaskCheckpointCoordinatorImpl.checkpointState(SubtaskCheckpointCoordinatorImpl.java:292)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.lambda$performCheckpoint$8(StreamTask.java:1089)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.runThrowing(StreamTaskActionExecutor.java:50)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:1073)
>> at
>> org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:1029)
>> ... 19 more
>> Caused by: java.util.concurrent.ExecutionException:
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.StatusRuntimeException:
>> CANCELLED: cancelled before receiving half close
>> at
>> java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
>> at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
>> at org.apache.beam.sdk.util.MoreFutures.get(MoreFutures.java:60)
>> at
>> org.apache.beam.runners.fnexecution.control.SdkHarnessClient$BundleProcessor$ActiveBundle.close(SdkHarnessClient.java:504)
>> at
>> org.apache.beam.runners.fnexecution.control.DefaultJobBundleFactory$SimpleStageBundleFactory$1.close(DefaultJobBundleFactory.java:555)
>> at
>> org.apache.flink.streaming.api.runners.python.beam.BeamPythonFunctionRunner.finishBundle(BeamPythonFunctionRunner.java:381)
>> ... 28 more
>> Caused by:
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.StatusRuntimeException:
>> CANCELLED: cancelled before receiving half close
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.Status.asRuntimeException(Status.java:524)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onCancel(ServerCalls.java:275)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.PartialForwardingServerCallListener.onCancel(PartialForwardingServerCallListener.java:40)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.ForwardingServerCallListener.onCancel(ForwardingServerCallListener.java:23)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.ForwardingServerCallListener$SimpleForwardingServerCallListener.onCancel(ForwardingServerCallListener.java:40)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.Contexts$ContextualizedServerCallListener.onCancel(Contexts.java:96)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.closedInternal(ServerCallImpl.java:353)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.closed(ServerCallImpl.java:341)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1Closed.runInContext(ServerImpl.java:867)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
>> at
>> org.apache.beam.vendor.grpc.v1p26p0.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
>> at
>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
>> at
>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
>> ... 1 more
>>
>> 2021-09-22 16:18:14,410 INFO org.apache.flink.runtime.taskmanager.Task [] -
>> Freeing task resources for KEYED PROCESS -> Sink: Unnamed (1/1)#34
>> (8f4fd40e863dd058822060dc3cf98831).
>> 2021-09-22 16:18:14,411 INFO
>> org.apache.flink.runtime.taskexecutor.TaskExecutor [] - Un-registering task
>> and sending final execution state FAILED to JobManager for task KEYED
>> PROCESS -> Sink: Unnamed (1/1)#34 8f4fd40e863dd058822060dc3cf98831.
>> 2021-09-22 16:18:14,426 INFO org.apache.flink.runtime.taskmanager.Task [] -
>> Attempting to cancel task Source: Custom Source ->
>> _stream_key_by_map_operator (1/1)#34 (290905523bfbf344381b7a04a8d36831).
>> 2021-09-22 16:18:14,426 INFO org.apache.flink.runtime.taskmanager.Task [] -
>> Source: Custom Source -> _stream_key_by_map_operator (1/1)#34
>> (290905523bfbf344381b7a04a8d36831) switched from RUNNING to CANCELING.
>> 2021-09-22 16:18:14,426 INFO org.apache.flink.runtime.taskmanager.Task [] -
>> Triggering cancellation of task code Source: Custom Source ->
>> _stream_key_by_map_operator (1/1)#34 (290905523bfbf344381b7a04a8d36831).
>> 2021-09-22 16:18:14,428 INFO org.apache.flink.runtime.taskmanager.Task [] -
>> Ignoring checkpoint aborted notification for non-running task Source: Custom
>> Source -> _stream_key_by_map_operator (1/1)#34.
>> 2021-09-22 16:18:14,430 INFO org.apache.kafka.common.metrics.Metrics [] -
>> Metrics scheduler closed
>> 2021-09-22 16:18:14,430 INFO org.apache.kafka.common.metrics.Metrics [] -
>> Closing reporter org.apache.kafka.common.metrics.JmxReporter
>> 2021-09-22 16:18:14,430 INFO org.apache.kafka.common.metrics.Metrics [] -
>> Metrics reporters closed
>>
>>
>> And, here is some source code:
>>
>> class DedupeFunction(FlatMapFunction):
>> def __init__(self, schema):
>> super().__init__()
>> self.logger = None
>> self.state = None
>> self.my_state = None
>> self.schema = schema
>> self.metric_columns = [c.column_name for c in schema.columns if
>> c.is_metric]
>>
>> def open(self, runtime_context: RuntimeContext):
>> self.logger = logging
>> self.logger.info('Opening the FlatMapFunction')
>> descriptor = MapStateDescriptor("my_map_state_descriptor",
>> Types.PICKLED_BYTE_ARRAY(), Types.PICKLED_BYTE_ARRAY())
>> self.state = runtime_context.get_map_state(descriptor)
>>
>> def flat_map(self, value):
>> try:
>> if not self.state.is_empty():
>> # self.logger.info('key in state')
>> previous_dict = {}
>> for k, v in self.state.items():
>> # reverse the metric columns
>> if k in self.metric_columns:
>> if v:
>> v = -v
>> previous_dict[k] = v
>> yield Row(**previous_dict)
>> new_dict = value.as_dict()
>> self.state.put_all(new_dict.items())
>> yield value
>> except Exception as ex:
>> self.logger.error(f'ERROR in dedupe_datastream: {str(ex)}')
>>
>> class PrimaryKeySelector(KeySelector):
>>
>> def __init__(self, primary_key):
>> self.__primary_key__ = primary_key
>>
>> def get_key(self, kv_obj):
>> return kv_obj.as_dict().get(self.__primary_key__)
>>
>>
>>
>> backend = RocksDBStateBackend(self.__conf__.get('CHECKPOINT_DIR'))
>>
>> self.__env__.set_state_backend(backend)
>>
>>
>> input_ds = input_ds.key_by(PrimaryKeySelector(self.__primary_key__),
>> key_type_info=primary_key_type_info)
>>
>> deduped_ds = input_ds.flat_map(DedupeFunction(self.__schema__),
>> output_type=type_info)
>>
>>
>> This program works fine if checkpointing is not enabled. Any advice here?
>>
>>
>> Thanks
