Thank you for the reply. " save_main_session" did not work, however, situation had changed.
1. get_all_options() output. "save_main_session" set to True. ================================================================================= 2019-09-26 09:04:11,586 DEBUG Pipeline Options: {'wait_until_finish_duration': None, 'update': False, 'min_cpu_platform': None, 'dataflow_endpoint': 'https://dataflow.googleapis.com', 'environment_config': 'asia.gcr.io/creationline001/beam/python3:latest', 'machine_type': None, 'enable_streaming_engine': False, 'sdk_location': 'default', 'profile_memory': False, 'max_num_workers': None, 'type_check_strictness': 'DEFAULT_TO_ANY', 'streaming': False, 'setup_file': None, 'network': None, 'on_success_matcher': None, 'requirements_cache': None, 'service_account_email': None, 'environment_type': 'DOCKER', 'disk_type': None, 'labels': None, 'profile_location': None, 'direct_runner_use_stacked_bundle': True, 'use_public_ips': None, ***** 'save_main_session': True, ******* 'direct_num_workers': 1, 'num_workers': None, 'worker_harness_container_image': None, 'template_location': None, 'hdfs_port': None, 'flexrs_goal': None, 'profile_cpu': False, 'transform_name_mapping': None, 'profile_sample_rate': 1.0, 'runner': 'PortableRunner', 'project': None, 'dataflow_kms_key': None, 'job_endpoint': 'localhost:8099', 'extra_packages': None, 'environment_cache_millis': 0, 'dry_run': False, 'autoscaling_algorithm': None, 'staging_location': None, 'job_name': None, 'no_auth': False, 'runtime_type_check': False, 'direct_runner_bundle_repeat': 0, 'subnetwork': None, 'pipeline_type_check': True, 'hdfs_user': None, 'dataflow_job_file': None, 'temp_location': None, 'sdk_worker_parallelism': 0, 'zone': None, 'experiments': ['beam_fn_api'], 'hdfs_host': None, 'disk_size_gb': None, 'dataflow_worker_jar': None, 'requirements_file': None, 'beam_plugins': None, 'pubsubRootUrl': None, 'region': None} ================================================================================= 2. Error in Task Manager log did not change. ================================================================================== File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 474, in find_class return StockUnpickler.find_class(self, module, name) AttributeError: Can't get attribute 'FlattenTagFilesFn' on <module 'apache_beam.runners.worker.sdk_worker_main' from '/usr/local/lib/python3.5/site-packages/apache_beam/runners/worker/sdk_worker_main.py'> ================================================================================== 3. However, if I comment out "super().__init__()" in my code , error changes. ================================================================================== File "/usr/local/lib/python3.5/site-packages/apache_beam/runners/worker/bundle_processor.py", line 1078, in _create_pardo_operation dofn_data = pickler.loads(serialized_fn) File "/usr/local/lib/python3.5/site-packages/apache_beam/internal/pickler.py", line 265, in loads return dill.loads(s) File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 317, in loads return load(file, ignore) File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 305, in load obj = pik.load() File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 474, in find_class return StockUnpickler.find_class(self, module, name) ImportError: No module named 's3_credentials' ================================================================================== 4. My whole class is below. ================================================================================== class FlattenTagFilesFn(beam.DoFn): def __init__(self, s3Bucket, s3Creds, maxKeys=1000): self.s3Bucket = s3Bucket self.s3Creds = s3Creds self.maxKeys = maxKeys super().__init__() def process(self, elem): if not hasattr(self, 's3Client'): import boto3 self.s3Client = boto3.client('s3', aws_access_key_id=self.s3Creds.awsAccessKeyId, aws_secret_access_key=self.s3Creds.awsSecretAccessKey) (key, info) = elem preFrm = {} resp1 = self.s3Client.get_object(Bucket=self.s3Bucket, Key=info['pre'][0][0]) yaml1 = yaml.load(resp1['Body']) for elem in yaml1['body']: preFrm[ elem['frame_tag']['frame_no'] ] = elem postFrm = {} resp2 = self.s3Client.get_object(Bucket=self.s3Bucket, Key=info['post'][0][0]) yaml2 = yaml.load(resp2['Body']) for elem in yaml2['body']: postFrm[ elem['frame_tag']['frame_no'] ] = elem commonFrmNums = set(list(preFrm.keys())).intersection(list(postFrm.keys())) for f in commonFrmNums: frames = Frames( self.s3Bucket, info['pre'][0][0], # Pre S3Key info['post'][0][0], # Post S3Key yaml1['head']['operator_id'], # Pre OperatorId yaml2['head']['operator_id'], # Post OperatorId preFrm[f], # Pre Frame Line postFrm[f], # Post Frame Line info['pre'][0][1], # Pre Last Modified Time info['post'][0][1]) # Post Last Modified Time yield (frames) tagCounts = TagCounts( self.s3Bucket, yaml1, # Pre Yaml yaml2, # Post Yaml info['pre'][0][0], # Pre S3Key info['post'][0][0], # Post S3Key info['pre'][0][1], # Pre Last Modified Time info['post'][0][1] ) # Post Last Modified Time yield beam.pvalue.TaggedOutput('counts', tagCounts) ================================================================================== I was using super() to define single instance of boto instance in ParDo. May I ask, is there a way to call super() in the constructor of ParDo ? Thanks, Yu On Thu, Sep 26, 2019 at 7:49 AM Kyle Weaver <kcwea...@google.com> wrote: > You will need to set the save_main_session pipeline option to True. > > Kyle Weaver | Software Engineer | github.com/ibzib | kcwea...@google.com > > > On Wed, Sep 25, 2019 at 3:44 PM Yu Watanabe <yu.w.ten...@gmail.com> wrote: > >> Hello. >> >> I would like to ask question for ParDo . >> >> I am getting below error inside TaskManager when running code on Apache >> Flink using Portable Runner. >> ===================================================== >> .... >> File >> "/usr/local/lib/python3.5/site-packages/apache_beam/runners/worker/bundle_processor.py", >> line 1078, in _create_pardo_operation >> dofn_data = pickler.loads(serialized_fn) >> File >> "/usr/local/lib/python3.5/site-packages/apache_beam/internal/pickler.py", >> line 265, in loads >> return dill.loads(s) >> File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 317, >> in loads >> return load(file, ignore) >> File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 305, >> in load >> obj = pik.load() >> File "/usr/local/lib/python3.5/site-packages/dill/_dill.py", line 474, >> in find_class >> return StockUnpickler.find_class(self, module, name) >> AttributeError: Can't get attribute 'FlattenTagFilesFn' on <module >> 'apache_beam.runners.worker.sdk_worker_main' from >> '/usr/local/lib/python3.5/site-packages/apache_beam/runners/worker/sdk_worker_main.py'> >> ===================================================== >> >> " FlattenTagFilesFn" is defined as ParDo and called from Pipeline as >> below. >> ===================================================== >> frames, counts = ({'pre': pcollPre, 'post': pcollPost} >> | 'combined:cogroup' >> beam.CoGroupByKey() >> | 'combined:exclude' >> beam.Filter(lambda x: >> (len(x[1]['pre']) > 0) and (len(x[1]['post']) > 0)) >> | 'combined:flat' >> >> beam.ParDo(FlattenTagFilesFn(s3Bucket, s3Creds)) >> .with_outputs('counts', >> main='frames')) >> ===================================================== >> >> In the same file I have defined the class as below. >> ===================================================== >> class FlattenTagFilesFn(beam.DoFn): >> def __init__(self, s3Bucket, s3Creds, maxKeys=1000): >> self.s3Bucket = s3Bucket >> self.s3Creds = s3Creds >> self.maxKeys = maxKeys >> ===================================================== >> >> This is not a problem when running pipeline using DirectRunner. >> May I ask , how should I import class for ParDo when running on Flink ? >> >> Thanks, >> Yu Watanabe >> >> -- >> Yu Watanabe >> Weekend Freelancer who loves to challenge building data platform >> yu.w.ten...@gmail.com >> [image: LinkedIn icon] <https://www.linkedin.com/in/yuwatanabe1> [image: >> Twitter icon] <https://twitter.com/yuwtennis> >> > -- Yu Watanabe Weekend Freelancer who loves to challenge building data platform yu.w.ten...@gmail.com [image: LinkedIn icon] <https://www.linkedin.com/in/yuwatanabe1> [image: Twitter icon] <https://twitter.com/yuwtennis>