Uploaded image for project: 'Beam'
  1. Beam
  2. BEAM-10854

PeriodicImpulse default arguments are not valid values

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Triage Needed
    • Priority: P2
    • Resolution: Unresolved
    • Affects Version/s: 2.23.0
    • Fix Version/s: None
    • Component/s: sdk-py-core
    • Labels:

      Description

      Based on the examples from Slowly updating side input using windowing I attempted to test the PeriodicImpulse using no variables, such that it triggered immediately and ran forever.

      The below code shows how:

              def pair_account_ids(
                  api_key: str, account_ids: Dict[str, str]
              ) -> Optional[Tuple[str, str, int]]:
                  if api_key not in account_ids:
                      return None
      
                  return (api_key, account_ids[api_key], int(time.time()))
      
              def echo(elm) -> Dict[str, str]:
                  print(elm)
                  return elm
      
              def api_keys(elm) -> Dict[str, str]:
                  return {"<api_key_1>": "<account_id_1>", "<api_key_2>": "<account_id_2>"}
      
              pipeline_options = PipelineOptions(streaming=True)
              with TestPipeline(
                  options=pipeline_options, runner=beam.runners.DirectRunner()
              ) as p:
                  side_input = (
                      p
                      | "PeriodicImpulse"
                      >> PeriodicImpulse(
                          # start_timestamp=start,
                          # stop_timestamp=stop,
                          fire_interval=5,
                          apply_windowing=True,
                      )
                      | "api_keys" >> beam.Map(api_keys)
                  )
      
                  main_input = (
                      p
                      | "MpImpulse"
                      >> beam.Create(["<api_key_1>", "<api_key_2>", "<unknown_api_key>"])
                      | "MapMpToTimestamped"
                      >> beam.Map(lambda src: TimestampedValue(src, time.time()))
                      | "WindowMpInto" >> beam.WindowInto(beam.window.FixedWindows(5))
                  )
      
                  result = (
                      main_input
                      | "Pair with AccountIDs"
                      >> beam.Map(
                          pair_account_ids, account_ids=beam.pvalue.AsSingleton(side_input)
                      )
                      | "filter" >> beam.Filter(lambda x: x is not None)
                      | "echo 2" >> beam.Map(lambda x: print(f"{int(time.time())}: {x}"))
                  )
              print(f"done:  {int(time.time())}")
      

      The above pipeline throws the following exception however:

      Traceback (most recent call last):
        File "/test/not_test.py", line 141, in test_side_input
          | "echo 2" >> beam.Map(lambda x: print(f"{int(time.time())}: {x}"))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/pipeline.py", line 555, in __exit__
          self.run().wait_until_finish()
        File "/usr/local/lib/python3.7/site-packages/apache_beam/testing/test_pipeline.py", line 112, in run
          False if self.not_use_test_runner_api else test_runner_api))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/pipeline.py", line 521, in run
          allow_proto_holders=True).run(False)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/pipeline.py", line 534, in run
          return self.runner.run_pipeline(self, self._options)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py", line 119, in run_pipeline
          return runner.run_pipeline(pipeline, options)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 173, in run_pipeline
          pipeline.to_runner_api(default_environment=self._default_environment))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 183, in run_via_runner_api
          return self.run_stages(stage_context, stages)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 340, in run_stages
          bundle_context_manager,
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 519, in _run_stage
          bundle_manager)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 557, in _run_bundle
          data_input, data_output, input_timers, expected_timer_output)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 941, in process_bundle
          timer_inputs)):
        File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 598, in result_iterator
          yield fs.pop().result()
        File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 435, in result
          return self.__get_result()
        File "/usr/local/lib/python3.7/concurrent/futures/_base.py", line 384, in __get_result
          raise self._exception
        File "/usr/local/lib/python3.7/site-packages/apache_beam/utils/thread_pool_executor.py", line 44, in run
          self._future.set_result(self._fn(*self._fn_args, **self._fn_kwargs))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 937, in execute
          dry_run)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 837, in process_bundle
          result_future = self._worker_handler.control_conn.push(process_bundle_req)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/worker_handlers.py", line 352, in push
          response = self.worker.do_instruction(request)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 480, in do_instruction
          getattr(request, request_type), request.instruction_id)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/sdk_worker.py", line 515, in process_bundle
          bundle_processor.process_bundle(instruction_id))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 978, in process_bundle
          element.data)
        File "/usr/local/lib/python3.7/site-packages/apache_beam/runners/worker/bundle_processor.py", line 218, in process_encoded
          self.output(decoded_value)
        File "apache_beam/runners/worker/operations.py", line 330, in apache_beam.runners.worker.operations.Operation.output
        File "apache_beam/runners/worker/operations.py", line 332, in apache_beam.runners.worker.operations.Operation.output
        File "apache_beam/runners/worker/operations.py", line 195, in apache_beam.runners.worker.operations.SingletonConsumerSet.receive
        File "apache_beam/runners/worker/operations.py", line 755, in apache_beam.runners.worker.operations.SdfProcessSizedElements.process
        File "apache_beam/runners/worker/operations.py", line 764, in apache_beam.runners.worker.operations.SdfProcessSizedElements.process
        File "apache_beam/runners/common.py", line 971, in apache_beam.runners.common.DoFnRunner.process_with_sized_restriction
        File "apache_beam/runners/common.py", line 711, in apache_beam.runners.common.PerWindowInvoker.invoke_process
        File "apache_beam/runners/common.py", line 807, in apache_beam.runners.common.PerWindowInvoker._invoke_process_per_window
        File "apache_beam/runners/common.py", line 1095, in apache_beam.runners.common._OutputProcessor.process_outputs
        File "/usr/local/lib/python3.7/site-packages/apache_beam/transforms/periodicsequence.py", line 124, in process
          timestamp.Timestamp(current_output_timestamp))
        File "/usr/local/lib/python3.7/site-packages/apache_beam/utils/timestamp.py", line 64, in __init__
          'Cannot interpret %s %s as seconds.' % (seconds, type(seconds)))
      TypeError: Cannot interpret Timestamp(1599216802.136201) <class 'apache_beam.utils.timestamp.Timestamp'> as seconds.
      

        Attachments

          Activity

            People

            • Assignee:
              Unassigned
              Reporter:
              prm Patrick Madsen
            • Votes:
              0 Vote for this issue
              Watchers:
              1 Start watching this issue

              Dates

              • Created:
                Updated: