[GitHub] [beam] ConverJens commented on pull request #13723: Propagate pipeline options to direct runner

[GitBox <gi...@apache.org>]

ConverJens commented on pull request #13723:
URL: https://github.com/apache/beam/pull/13723#issuecomment-770755541


   @dandy10 No not much, unfortunately. I added logging to both classes that you suggested and from SubprocessSdkWorker I get empty pipeline options, but even more strange is that I get no output at all from my loggs in SwitchingDirectRunner. 
   
   If this is indeed the case, it would explain why there are no options present downstream since they are exported in SwitchingDirectRunner. 
   
   Is there any other path that to initialize the pipeline that doesn't go through SwitchingDirectRunner?
   
   These are the first part of the beam logs where the subprocesses are initalized and each subprocess calls sdk_worker_main.
   
   ```
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function annotate_downstream_side_inputs at 0x7f8370065710> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function fix_side_input_pcoll_coders at 0x7f8370065830> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function eliminate_common_key_with_none at 0x7f83700659e0> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function pack_combiners at 0x7f8370065a70> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function lift_combiners at 0x7f8370065b00> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function expand_sdf at 0x7f8370065cb0> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function expand_gbk at 0x7f8370065d40> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function sink_flattens at 0x7f8370065e60> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function greedily_fuse at 0x7f8370065ef0> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function read_to_impulse at 0x7f8370065f80> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function impulse_to_input at 0x7f8370066050> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function sort_stages at 0x7f8370066290> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function setup_timer_mapping at 0x7f8370066200> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.translations:==================== <function populate_data_channel_coders at 0x7f8370066320> ====================
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting control server on port 39917
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting data server on port 45603
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting state server on port 38319
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:starting logging server on port 35791
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Created Worker handler <apache_beam.runners.portability.fn_api_runner.worker_handlers.SubprocessSdkWorkerHandler object at 0x7f84433a6910> for environment ref_Environment_default_environment_1 (beam:env:harness_subprocess_python:v1, b'/usr/bin/python -m apache_beam.runners.worker.sdk_worker_main')
   WARNING:root:SubprocessSdkWorker
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Created Worker handler <apache_beam.runners.portability.fn_api_runner.worker_handlers.SubprocessSdkWorkerHandler object at 0x7f8441b39950> for environment ref_Environment_default_environment_1 (beam:env:harness_subprocess_python:v1, b'/usr/bin/python -m apache_beam.runners.worker.sdk_worker_main')
   WARNING:root:SubprocessSdkWorker
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Created Worker handler <apache_beam.runners.portability.fn_api_runner.worker_handlers.SubprocessSdkWorkerHandler object at 0x7f8441b591d0> for environment ref_Environment_default_environment_1 (beam:env:harness_subprocess_python:v1, b'/usr/bin/python -m apache_beam.runners.worker.sdk_worker_main')
   WARNING:root:SubprocessSdkWorker
   INFO:apache_beam.runners.portability.fn_api_runner.worker_handlers:Created Worker handler <apache_beam.runners.portability.fn_api_runner.worker_handlers.SubprocessSdkWorkerHandler object at 0x7f8441b59810> for environment ref_Environment_default_environment_1 (beam:env:harness_subprocess_python:v1, b'/usr/bin/python -m apache_beam.runners.worker.sdk_worker_main')
   WARNING:root:SubprocessSdkWorker pipeline_options: {}
   WARNING:root:SubprocessSdkWorker
   WARNING:root:SubprocessSdkWorker pipeline_options: {}
   INFO:apache_beam.runners.portability.fn_api_runner.fn_runner:Running (((((ref_AppliedPTransform_WriteSplit[train]/Write/Write/WriteImpl/DoOnce/Impulse_27)+(ref_AppliedPTransform_WriteSplit[train]/Write/Write/WriteImpl/DoOnce/FlatMap(<lambda at core.py:2957>)_28))+(ref_AppliedPTransform_WriteSplit[train]/Write/Write/WriteImpl/DoOnce/Map(decode)_30))+(ref_AppliedPTransform_WriteSplit[train]/Write/Write/WriteImpl/InitializeWrite_31))+(ref_PCollection_PCollection_17/Write))+(ref_PCollection_PCollection_18/Write)
   WARNING:root:SubprocessSdkWorker pipeline_options: {}
   WARNING:root:SubprocessSdkWorker pipeline_options: {}
   ```


----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org