ApacheBeam在医疗领域中的应用：实时数据处理和AI辅助诊断

    import apache_beam as beam
    from apache_beam.io import ReadFromText
    from apache_beam.io import WriteToText
    from apache_beam.options.pipeline_options import PipelineOptions
    from apache_beam.metrics import Metrics
    from apache_beam.metrics.metric import MetricsFilter
    import logging
    import json
    import re
    import os

    INPUT_DIR = 'gs://bucket/data'
    OUTPUT_DIR = 'gs://bucket/output'
    SCHEMA_FILE = '/path/to/schema.json' # 可以在这里定义数据字典的JSON文件位置
    LOOKUP_TABLE_FILE = '/path/to/lookup_table.csv' # 可以在这里定义检验数据库的CSV文件位置

    class DataCleaner(beam.DoFn):
    def __init__(self):
        self.metrics = Metrics.counter(self.__class__, "clean_records")
    
    def process(self, element):
        data = json.loads(element)
        
        try:
            patient_id = int(data['patient_id']) if isinstance(data['patient_id'], str) else None
            
            name = ''
            if 'first_name' in data and 'last_name' in data:
                name += (data['first_name'] +'') if len(data['first_name'].strip()) > 0 else ''
                name += (data['last_name'] +'') if len(data['last_name'].strip()) > 0 else ''
                
            age = int(data['age']) if ('age' in data and isinstance(data['age'], (int, float))) else None
            
            blood_type = ''
            if 'blood_group' in data and isinstance(data['blood_group'], str):
                match = re.match('^([A-Za-z]+)([-+])(\d+)$', data['blood_group'])
                
                if match is not None:
                    blood_type = ''.join(match.groups()).upper()
                    
            date = ''
            if 'date' in data and isinstance(data['date'], str):
                match = re.search('\d{4}-\d{2}-\d{2}', data['date'])
                
                if match is not None:
                    date = match.group()
                
            result = {
                'patient_id': patient_id, 
                'name': name.strip(), 
                'age': age, 
                'blood_type': blood_type, 
               'sample_time': date
            }
            
        except KeyError:
            logging.error("KeyError when processing record with keys %s", ','.join(data.keys()))
            return []
                
        for key, value in list(result.items()):
            if value == '':
                del result[key]
                
        yield json.dumps(result).encode('utf-8')
        
    def finish_bundle(self):
        self.metrics.commit()

    class TransformData(beam.DoFn):
    
    def __init__(self):
        self.schema = {}
        
        with open(SCHEMA_FILE) as f:
            schema_dict = json.load(f)
            
            for col in schema_dict:
                self.schema[col['name']] = col
                
    def process(self, element):
        record = json.loads(element)
        
        transformed_record = {'patient_id': '', 'name': '', 'age': -1, 'blood_type': '','sample_time': '',
                               **{''.join((k[:1].lower()+re.sub('[^A-Za-z0-9]+','',v)).split()) : v for k,v in record.items()}}
        
        missing_cols = set(self.schema.keys()) - set(transformed_record.keys())
        
        if len(missing_cols)!= 0:
            logging.warning("Record with id '%s' has missing columns %s.", transformed_record['patient_id'], ', '.join(list(missing_cols)))
            return []
        
        clean_values = [value for field_name, value in transformed_record.items()
                        if ((field_name in ['name','blood_type']) or
                            (isinstance(value,(str,float)) and len(value)>0))]
        
        if len(clean_values)<len(self.schema):
            logging.warning("Record with id '%s' does not have any valid values after cleaning up.", transformed_record['patient_id'])
            return []
        
        sample_time = transformed_record.pop('sample_time')
        
        output = {'PatientID': transformed_record.pop('patient_id'),
                  'SampleTime': sample_time}
                  
        for field_name, value in transformed_record.items():
            info = self.schema.get(field_name)
            
            if info is not None:
                col_type = info.get('type')
                
                if col_type=='categorical':
                    output[info['name']] = int(value==info['value'])
                    
                elif col_type=='continuous':
                    lower_bound = float(info.get('range').get('lower'))
                    upper_bound = float(info.get('range').get('upper'))
                    
                    scale = max(upper_bound-lower_bound,1e-5)
                    
                    scaled_value = min(max((value-lower_bound)/scale,0),1)
                    
                    output[info['name']] = round(scaled_value,info.get('precision',4))
                    
        yield json.dumps(output).encode('utf-8')

    def write_to_file(output_dir):
    class FileWriter(beam.DoFn):
        def process(self, element):
            file_path = os.path.join(output_dir, "%s_%s.json" % (element["PatientID"],
                                                               element["SampleTime"].replace('-', '_')))
    
            with open(file_path, mode='ab+') as f:
                f.seek(0, 2)
                pos = f.tell()
    
                if pos == 0:
                    header = ",".join(['"%s"'%x for x in sorted(element.keys())]).encode()+'
    '.encode()
                    f.write(header)
    
                row = ",".join(['"%s"'%str(x) for x in element.values()]).encode()+'
    '.encode()
                f.write(row)
    
        def setup(self):
            print("Setting up...")
    
    return FileWriter()

    def run(argv=None):
    pipeline_args = ["--runner=DirectRunner"] # 使用本地运行器
    
    options = PipelineOptions(pipeline_args)
    p = beam.Pipeline(options=options)
    
    records = p | 'ReadFromGCS' >> ReadFromText('%s/*.json' % INPUT_DIR)
    cleaned_records = records | 'DataCleaner' >> beam.ParDo(DataCleaner())
    
    processed_records = cleaned_records | 'TransformData' >> beam.ParDo(TransformData())
    
    written_files = processed_records | 'WriteToFiles' >> beam.ParDo(write_to_file(OUTPUT_DIR))
    
    result = p.run()
    result.wait_until_finish()

    gcloud dataproc jobs submit pyspark --cluster=[your cluster name] --region=[your region] \
       --jars=pyarrow==3.0.0 \
       --project=[your project ID] \
       /path/to/script.py --setup_file=/path/to/setup.py