Re: Python SDF for unbound source
I did a few experiments and am still a little confused about what’s
happening using either of the DirectRunner multiprocessing modes.
I’m running the following code:
import argparseimport sys
import apache_beam as beamfrom apache_beam.options.pipeline_options
import PipelineOptionsfrom apache_beam.transforms.core import
RestrictionProvider
class InfRestrictionProvider(RestrictionProvider):
"""
An infinite restriction provider
"""
def initial_restriction(self, element):
from apache_beam.io.restriction_trackers import OffsetRange
return OffsetRange(0, float('inf'))
def create_tracker(self, restriction):
from apache_beam.io.restriction_trackers import OffsetRestrictionTracker
return OffsetRestrictionTracker(restriction)
def restriction_size(self, element, restriction):
return 1
class _EventReadSDF(beam.DoFn):
"""
An SDF for subscribing to custom events.
"""
restriction_tracker = beam.DoFn.RestrictionParam(InfRestrictionProvider())
def __init__(self):
super(_FtrackEventReadSDF, self).__init__()
# number of events received
self._counter = 0
@beam.DoFn.unbounded_per_element()
def process(self, _, restriction_tracker=restriction_tracker):
import random
import time
print('Attempting to claim {!r}'.format(self._counter))
if not restriction_tracker.try_claim(self._counter):
print('Failed to claim {!r}'.format(self._counter))
return
try:
n = random.randrange(0, 5)
time.sleep(n)
yield n
finally:
self._counter += 1
restriction_tracker.defer_remainder()
def main(argv=None):
parser = argparse.ArgumentParser()
known_args, pipeline_args = parser.parser_known_args(argv)
pipe = beam.Pipeline(options=PipelineOptions(pipeline_args))
(
pipe
| beam.Impulse()
| beam.ParDo(_EventReadSDF())
| beam.Map(print)
)
result = pipe.run()
result.wait_until_finish()
if __name__ == '__main__':
main(sys.argv)
If I run this pipeline with more than one worker (e.g.
--direct_num_workers=2), then the SDF only produces one element. This seems
to be the case using either the direct_running_mode of multi_processing or
multi_threading. I tested this in python2 running beam-2.24.0 and python3
running beam-2.29.0.
Interestingly, if I run this with --direct_num_workers=1 it will generate
elements forever as intended. It would be great to understand what’s going
on a bit further. Is this the right technique for the RestrictionProvider?
Is there any other sample code that anyone has for doing something similar?
Thanks,
-Sam
On Wed, Mar 9, 2022 at 8:22 PM Ankur Goenka wrote:
> This might be an issue with multi_processing implementation of
> DirectRunner.
>
> I experimented with the following code and got following error. You can
> try using --direct_running_mode='multi_threading'
>
> --
> File
> "/usr/local/google/home/goenka/d/work/tmp/t1/venv/lib/python3.9/site-packages/apache_beam/runners/worker/bundle_processor.py",
> line 1425, in process
> initial_restriction = self.restriction_provider.initial_restriction(
> File "/usr/local/google/home/goenka/d/work/tmp/t1/test.py", line 15, in
> initial_restriction
> return OffsetRange(0, float('inf'))
> NameError: name 'OffsetRange' is not defined
>
>
> --
>
>
> import apache_beam as beam
> import sys
> import argparse
> from apache_beam.io.restriction_trackers import OffsetRange,
> OffsetRestrictionTracker
> from apache_beam.options.pipeline_options import PipelineOptions
> from apache_beam.transforms.core import RestrictionProvider
>
>
> class InfRestrictionProvider(RestrictionProvider):
> """
> An infinite restriction provider
> """
>
> def initial_restriction(self, element):
> return OffsetRange(0, float('inf'))
>
> def create_tracker(self, restriction):
> return OffsetRestrictionTracker(restriction)
>
> def restriction_size(self, element, restriction):
> return 1
>
> @beam.typehints.with_output_types(int)
> class _EventReadSDF(beam.DoFn):
> """
> An SDF for subscribing to custom events.
> """
>
> restriction_tracker =
> beam.DoFn.RestrictionParam(InfRestrictionProvider())
>
> def __init__(self):
> # type: () -> None
> self._counter = 0
>
> def process(self, _, restriction_tracker=restriction_tracker):
> # type: (Any, beam.DoFn.RestrictionParam) -> Iterator[Event]
>
> if not restriction_tracker.try_claim(self._counter):
> return
>
> i = 0
> for _ in range(100):
> try:
> # Blocks until the next event is received.
> i+=1
> yield i
> finally:
> self._counter += 1
> restriction_tracker.defer_remainder()
>
>
> def run(a
Re: Re: Why is portable python runner trying to stage a flink job server jar?
Hi Cham, Thanks. I tried adding my jar to the classpath in the Java SDK container, but it made no difference. I then saw that the loader program for the Java SDK harness hard-codes a few jars to be in the classpath [1]. It also tacks on any jars specified to be staged, but in any case, it does not seem to include every jar in /opt/apache/beam/jars directory. Regards, Deepak [1] https://github.com/apache/beam/blob/c1db5fe0afabc6639ea09ca36ef13ebb63b283de/sdks/java/container/boot.go#L135 On Tue, Mar 8, 2022 at 11:07 AM Chamikara Jayalath wrote: > > > > On Fri, Mar 4, 2022 at 3:16 PM Deepak Nagaraj wrote: >> >> Hi Cham, >> >> Thanks, we set filesToStage to be empty within expansion service >> configuration. >> >> Now when we run a Python pipeline that reads from Kafka, we see an exception >> about a missing class. [1] >> >> This class is present in the jar that was getting staged earlier, so in a >> way the failure is expected. >> >> However, the same jar is also present within the classpath of the >> taskmanager [2], yet it does not seem to be considered by the function >> harness. > > > Hi Deepak, > > In general, when "filesToStage" is excluded, everything in the CLASSPATH > should be staged for the SDK Harness. I'm not too familiar with the Flink > dependency setup but is it possible that the relevant jar does not get staged > for the SDK harness somehow ? > > Thanks, > Cham > >> >> >> Is there a way we can let the function harness consider the available jar? >> >> Also, is our understanding correct that the harness only looks at jars >> previously uploaded to the staging service? >> >> Thanks, >> Deepak >> >> [1] Exception about missing class: >> >> 2022-03-03 17:37:29,531 ERROR >> org.apache.beam.fn.harness.control.BeamFnControlClient [] - >> Error thrown when handling InstructionRequest 1 >> java.lang.NoClassDefFoundError: >> org/springframework/expression/EvaluationContext >>at >> org.apache.beam.sdk.io.kafka.KafkaRecordCoder.toHeaders(KafkaRecordCoder.java:86) >>at >> org.apache.beam.sdk.io.kafka.KafkaRecordCoder.decode(KafkaRecordCoder.java:81) >>at >> org.apache.beam.sdk.io.kafka.KafkaRecordCoder.decode(KafkaRecordCoder.java:40) >>at org.apache.beam.sdk.coders.Coder.decode(Coder.java:159) >>at >> org.apache.beam.sdk.coders.LengthPrefixCoder.decode(LengthPrefixCoder.java:64) >>at org.apache.beam.sdk.coders.Coder.decode(Coder.java:159) >>at >> org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.decode(WindowedValue.java:621) >>at >> org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.decode(WindowedValue.java:612) >>at >> org.apache.beam.sdk.util.WindowedValue$FullWindowedValueCoder.decode(WindowedValue.java:558) >>at >> org.apache.beam.sdk.fn.data.BeamFnDataInboundObserver2.awaitCompletion(BeamFnDataInboundObserver2.java:126) >>at >> org.apache.beam.fn.harness.control.ProcessBundleHandler.processBundle(ProcessBundleHandler.java:467) >>at >> org.apache.beam.fn.harness.control.BeamFnControlClient.delegateOnInstructionRequestType(BeamFnControlClient.java:151) >>at >> org.apache.beam.fn.harness.control.BeamFnControlClient$InboundObserver.lambda$onNext$0(BeamFnControlClient.java:116) >>at >> java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) >>at >> java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) >>at java.base/java.lang.Thread.run(Thread.java:829) >> Caused by: java.lang.ClassNotFoundException: >> org.springframework.expression.EvaluationContext >>at >> java.base/jdk.internal.loader.BuiltinClassLoader.loadClass(BuiltinClassLoader.java:581) >>at >> java.base/jdk.internal.loader.ClassLoaders$AppClassLoader.loadClass(ClassLoaders.java:178) >>at java.base/java.lang.ClassLoader.loadClass(ClassLoader.java:522) >>... 16 more >> >> [2] Classpath of task manager (the missing class is present in >> beam-runners-flink-job-server-no-slf4j.jar, which apparently is not >> considered by the harness): >> >> 2022-03-03 07:18:42,146 INFO >> org.apache.flink.runtime.taskexecutor.TaskManagerRunner [] - >> Classpath: /opt/flink/lib/beam-runners-flink-job-server-no-slf4j.jar:/op >> t/flink/lib/flink-csv-1.13.5.jar:/opt/flink/lib/flink-json-1.13.5.jar:/opt/flink/lib/flink-shaded-zookeeper-3.4.14.jar:/opt/flink/lib/flink-table-blink_2.12-1.13.5.jar:/ >> opt/flink/lib/flink-table_2.12-1.13.5.jar:/opt/flink/lib/log4j-1.2-api-2.16.0.jar:/opt/flink/lib/log4j-api-2.16.0.jar:/opt/flink/lib/log4j-core-2.16.0.jar:/opt/flink/lib >> /log4j-slf4j-impl-2.16.0.jar:/opt/flink/lib/flink-dist_2.12-1.13.5.jar::: >> >> On 2022/02/11 22:22:16 Chamikara Jayalath wrote: >> > Hi Jeremy, >> > >> > By default we stage all jars in the CLASSPATH of expansion service. You can >> > override this by setting the filesToStage option when starting up the >>
