[ https://issues.apache.org/jira/browse/BEAM-2573?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16080933#comment-16080933 ]
Dmitry Demeshchuk commented on BEAM-2573: ----------------------------------------- Hmm, looks like it was indeed a dependency issue which setuptools was hiding. I did `pip install .` from the beam repo, and the plugins made it in there. Interestingly, Beam seems to have automatically converted the plugins list into a list of classes. In the Dataflow job's options, {{beam_plugins}} equals to {{['apache_beam.io.filesystem.FileSystem', 'apache_beam.io.localfilesystem.LocalFileSystem', 'apache_beam.io.gcp.gcsfilesystem.GCSFileSystem', 'dataflow.aws.s3.S3FileSystem']}}. Interestingly, the crash loop described in my previous comment still happens. I'll continue looking at the sources to see where I screwed up. > Better filesystem discovery mechanism in Python SDK > --------------------------------------------------- > > Key: BEAM-2573 > URL: https://issues.apache.org/jira/browse/BEAM-2573 > Project: Beam > Issue Type: Task > Components: runner-dataflow, sdk-py > Affects Versions: 2.0.0 > Reporter: Dmitry Demeshchuk > Priority: Minor > > It looks like right now custom filesystem classes have to be imported > explicitly: > https://github.com/apache/beam/blob/master/sdks/python/apache_beam/io/filesystems.py#L30 > Seems like the current implementation doesn't allow discovering filesystems > that come from side packages, not from apache_beam itself. Even if I put a > custom FileSystem-inheriting class into a package and explicitly import it in > the root __init__.py of that package, it still doesn't make the class > discoverable. > The problems I'm experiencing happen on Dataflow runner, while Direct runner > works just fine. Here's an example of Dataflow output: > {code} > (320418708fe777d7): Traceback (most recent call last): > File > "/usr/local/lib/python2.7/dist-packages/dataflow_worker/batchworker.py", line > 581, in do_work > work_executor.execute() > File "/usr/local/lib/python2.7/dist-packages/dataflow_worker/executor.py", > line 166, in execute > op.start() > File > "/usr/local/lib/python2.7/dist-packages/dataflow_worker/native_operations.py", > line 54, in start > self.output(windowed_value) > File "dataflow_worker/operations.py", line 138, in > dataflow_worker.operations.Operation.output > (dataflow_worker/operations.c:5768) > def output(self, windowed_value, output_index=0): > File "dataflow_worker/operations.py", line 139, in > dataflow_worker.operations.Operation.output > (dataflow_worker/operations.c:5654) > cython.cast(Receiver, > self.receivers[output_index]).receive(windowed_value) > File "dataflow_worker/operations.py", line 72, in > dataflow_worker.operations.ConsumerSet.receive > (dataflow_worker/operations.c:3506) > cython.cast(Operation, consumer).process(windowed_value) > File "dataflow_worker/operations.py", line 328, in > dataflow_worker.operations.DoOperation.process > (dataflow_worker/operations.c:11162) > with self.scoped_process_state: > File "dataflow_worker/operations.py", line 329, in > dataflow_worker.operations.DoOperation.process > (dataflow_worker/operations.c:11116) > self.dofn_receiver.receive(o) > File "apache_beam/runners/common.py", line 382, in > apache_beam.runners.common.DoFnRunner.receive > (apache_beam/runners/common.c:10156) > self.process(windowed_value) > File "apache_beam/runners/common.py", line 390, in > apache_beam.runners.common.DoFnRunner.process > (apache_beam/runners/common.c:10458) > self._reraise_augmented(exn) > File "apache_beam/runners/common.py", line 431, in > apache_beam.runners.common.DoFnRunner._reraise_augmented > (apache_beam/runners/common.c:11673) > raise new_exn, None, original_traceback > File "apache_beam/runners/common.py", line 388, in > apache_beam.runners.common.DoFnRunner.process > (apache_beam/runners/common.c:10371) > self.do_fn_invoker.invoke_process(windowed_value) > File "apache_beam/runners/common.py", line 281, in > apache_beam.runners.common.PerWindowInvoker.invoke_process > (apache_beam/runners/common.c:8626) > self._invoke_per_window(windowed_value) > File "apache_beam/runners/common.py", line 307, in > apache_beam.runners.common.PerWindowInvoker._invoke_per_window > (apache_beam/runners/common.c:9302) > windowed_value, self.process_method(*args_for_process)) > File > "/Users/dmitrydemeshchuk/miniconda2/lib/python2.7/site-packages/apache_beam/transforms/core.py", > line 749, in <lambda> > File > "/Users/dmitrydemeshchuk/miniconda2/lib/python2.7/site-packages/apache_beam/io/iobase.py", > line 891, in <lambda> > File > "/usr/local/lib/python2.7/dist-packages/apache_beam/options/value_provider.py", > line 109, in _f > return fnc(self, *args, **kwargs) > File > "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filebasedsink.py", > line 146, in initialize_write > tmp_dir = self._create_temp_dir(file_path_prefix) > File > "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filebasedsink.py", > line 151, in _create_temp_dir > base_path, last_component = FileSystems.split(file_path_prefix) > File > "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filesystems.py", line > 99, in split > filesystem = FileSystems.get_filesystem(path) > File > "/usr/local/lib/python2.7/dist-packages/apache_beam/io/filesystems.py", line > 61, in get_filesystem > raise ValueError('Unable to get the Filesystem for path %s' % path) > ValueError: Unable to get the Filesystem for path > s3://my-test-bucket/test_output [while running > 'write/Write/WriteImpl/InitializeWrite'] > {code} > I apologize for not providing full context or codebase, because a lot of the > code we are running is internal, and some of it is tightly coupled to our > infrastructure. If I run out of experimenting options, I'll try to narrow my > use case down to the simplest case possible (like, override a gcs filesystem > with a different path prefix or something). > I think there are several possibilities here: > 1. I'm doing something wrong, and it should be trivial to achieve something > like that. This probably implies figuring out the right approach and writing > some guideline for the sources/sinks page in the docs. > 2. The current order of imports is not optimal, and we could possibly import > the side packages before initializing the filesystem classes. I currently > possess too little knowledge about the way things get executed in Dataflow, > so it's hard for me to tell how much it's worth diving that rabbithole. > 3. There just needs to be a better way of referring to additional filesystem > classes. One way of doing that is to just specify a class name explicitly > inside the ReadFromText and WriteToText functions (or something along these > lines). PipelineOptions seems like an overkill for this, but may still be an > option. Finally, maybe there could be just a function that gets called in the > main script that somehow tells Beam to discover a specific class? -- This message was sent by Atlassian JIRA (v6.4.14#64029)