Google Dataflow是否支持使用Python SDK开发的有状态管道?


根据

能力矩阵和Quora上的早期帖子,Google Dataflow似乎在某种程度上支持有状态管道

但是,虽然我有一个在 DirectRunner 上运行良好的有状态管道,但我在使用 DataflowRunner 时收到此错误:

...
 File "pipelines/StatefulPipeline.py", line 197, in process
    states = list(location_state.read())
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/bundle_processor.py", line 318, in __iter__
    for elem in self.first:
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/bundle_processor.py", line 187, in __iter__
    data, continuation_token = self._state_handler.blocking_get(self._state_key)
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 489, in blocking_get
    continuation_token=continuation_token)))
  File "/usr/local/lib/python2.7/dist-packages/apache_beam/runners/worker/sdk_worker.py", line 518, in _blocking_request
    raise RuntimeError(response.error)
RuntimeError: java.lang.IllegalStateException: Tried to access state for stateless step: NameContext{stageName=s02, originalName=s12, systemName=s12, userName=AlertEngine}
    at org.apache.beam.vendor.guava.v20_0.com.google.common.base.Preconditions.checkState(Preconditions.java:444)
    at org.apache.beam.runners.dataflow.worker.StreamingModeExecutionContext$StepContext.stateInternals(StreamingModeExecutionContext.java:685)
    at org.apache.beam.runners.dataflow.worker.StreamingModeExecutionContext$UserStepContext.stateInternals(StreamingModeExecutionContext.java:737)
    at org.apache.beam.runners.dataflow.worker.fn.control.RegisterAndProcessBundleOperation.lambda$handleBagUserState$5(RegisterAndProcessBundleOperation.java:509)
    at java.util.concurrent.ConcurrentHashMap.computeIfAbsent(ConcurrentHashMap.java:1660)
    at org.apache.beam.runners.dataflow.worker.fn.control.RegisterAndProcessBundleOperation.handleBagUserState(RegisterAndProcessBundleOperation.java:505)
    at org.apache.beam.runners.dataflow.worker.fn.control.RegisterAndProcessBundleOperation.delegateByStateKeyType(RegisterAndProcessBundleOperation.java:384)
    at org.apache.beam.runners.fnexecution.state.GrpcStateService$Inbound.onNext(GrpcStateService.java:130)
    at org.apache.beam.runners.fnexecution.state.GrpcStateService$Inbound.onNext(GrpcStateService.java:118)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.stub.ServerCalls$StreamingServerCallHandler$StreamingServerCallListener.onMessage(ServerCalls.java:248)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.ForwardingServerCallListener.onMessage(ForwardingServerCallListener.java:33)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.Contexts$ContextualizedServerCallListener.onMessage(Contexts.java:76)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.messagesAvailable(ServerCallImpl.java:263)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1MessagesAvailable.runInContext(ServerImpl.java:683)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
    at org.apache.beam.vendor.grpc.v1p13p1.io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
 [while running 'generatedPtransform-327']
...

我的代码看起来有点像:

 class AlertEngine(DoFn):
        LOCATION_STATE = BagStateSpec('location', StrUtf8Coder())
        def process(self, msg, location_state=DoFn.StateParam(LOCATION_STATE)):
            location = msg.get('location')
            states = list(location_state.read())
            previous_location = states[0] if states else None
            if previous_location != location:
                yield location
                location_state.clear()
                location_state.add(location)

因此,我开始不确定数据流是否不支持带有Python SDK的有状态管道。我在Python SDK发行说明中发现的信息很少。

https://issues.apache.org/jira/browse/BEAM-10786

数据流运行程序对状态的支持有限(主要是在流式处理模式下(。

最新更新