Traceback (most recent call last):
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/worker/data_plane.py", line 421, in input_elements
    element = received.get(timeout=1)
  File "/usr/lib64/python3.7/queue.py", line 178, in get
    raise Empty
_queue.Empty

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/lib64/python3.7/runpy.py", line 193, in _run_module_as_main
    "__main__", mod_spec)
  File "/usr/lib64/python3.7/runpy.py", line 85, in _run_code
    exec(code, run_globals)
  File "/home/ec2-user/workspace/src/python/v2/koopa/koopa.py", line 98, in <module>
    generate_statistics(args.workspace, args.metastore_dump_url, args.hive_db, args.hive_table, args.start, args.end)
  File "/home/ec2-user/workspace/src/python/v2/koopa/koopa.py", line 72, in generate_statistics
    pipeline_options=PipelineOptions(['--direct_running_mode', 'multi_processing', '--direct_num_workers', '0'])
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/tensorflow_data_validation/utils/stats_gen_lib.py", line 111, in generate_statistics_from_tfrecord
    output_path))
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/pipeline.py", line 555, in __exit__
    self.run().wait_until_finish()
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/pipeline.py", line 534, in run
    return self.runner.run_pipeline(self, self._options)
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/direct/direct_runner.py", line 119, in run_pipeline
    return runner.run_pipeline(pipeline, options)
  File "/home/ec2-user/workspace/lib64/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 "/home/ec2-user/workspace/lib64/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 "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 340, in run_stages
    bundle_context_manager,
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 519, in _run_stage
    bundle_manager)
  File "/home/ec2-user/workspace/lib64/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 "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 941, in process_bundle
    timer_inputs)):
  File "/usr/lib64/python3.7/concurrent/futures/_base.py", line 598, in result_iterator
    yield fs.pop().result()
  File "/usr/lib64/python3.7/concurrent/futures/_base.py", line 428, in result
    return self.__get_result()
  File "/usr/lib64/python3.7/concurrent/futures/_base.py", line 384, in __get_result
    raise self._exception
  File "/home/ec2-user/workspace/lib64/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 "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 937, in execute
    dry_run)
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/portability/fn_api_runner/fn_runner.py", line 855, in process_bundle
    abort_callback=lambda:
  File "/home/ec2-user/workspace/lib64/python3.7/site-packages/apache_beam/runners/worker/data_plane.py", line 424, in input_elements
    raise RuntimeError('Channel closed prematurely.')
RuntimeError: Channel closed prematurely.