Serverless 实战：利用函数计算与对象存储实现 WordCount

对象存储1  ap-guangzhou  srcmr对象存储2  ap-guangzhou  middlestagebucket对象存储3  ap-guangzhou  destcmr

# -*- coding: utf8 -*-import datetimefrom qcloud_cos_v5 import CosConfigfrom qcloud_cos_v5 import CosS3Clientfrom qcloud_cos_v5 import CosServiceErrorimport reimport osimport sysimport logginglogging.basicConfig(level=logging.INFO, stream=sys.stdout)logger = logging.getLogger()logger.setLevel(level=logging.INFO)region = u'ap-guangzhou'  # 根据实际情况，修改地域middle_stage_bucket = 'middlestagebucket'  # 根据实际情况，修改bucket名def delete_file_folder(src):    if os.path.isfile(src):        try:            os.remove(src)        except:            pass    elif os.path.isdir(src):        for item in os.listdir(src):            itemsrc = os.path.join(src, item)            delete_file_folder(itemsrc)        try:            os.rmdir(src)        except:            passdef download_file(cos_client, bucket, key, download_path):    logger.info("Get from [%s] to download file [%s]" % (bucket, key))    try:        response = cos_client.get_object(Bucket=bucket, Key=key, )        response['Body'].get_stream_to_file(download_path)    except CosServiceError as e:        print(e.get_error_code())        print(e.get_error_msg())        return -1    return 0def upload_file(cos_client, bucket, key, local_file_path):    logger.info("Start to upload file to cos")    try:        response = cos_client.put_object_from_local_file(            Bucket=bucket,            LocalFilePath=local_file_path,            Key='{}'.format(key))    except CosServiceError as e:        print(e.get_error_code())        print(e.get_error_msg())        return -1    logger.info("Upload data map file [%s] Success" % key)    return 0def do_mapping(cos_client, bucket, key, middle_stage_bucket, middle_file_key):    src_file_path = u'/tmp/' + key.split('/')[-1]    middle_file_path = u'/tmp/' + u'mapped_' + key.split('/')[-1]    download_ret = download_file(cos_client, bucket, key, src_file_path)  # download src file    if download_ret == 0:        inputfile = open(src_file_path, 'r')  # open local /tmp file        mapfile = open(middle_file_path, 'w')  # open a new file write stream        for line in inputfile:            line = re.sub('[^a-zA-Z0-9]', ' ', line)  # replace non-alphabetic/number characters            words = line.split()            for word in words:                mapfile.write('%s\t%s' % (word, 1))  # count for 1                mapfile.write('\n')        inputfile.close()        mapfile.close()        upload_ret = upload_file(cos_client, middle_stage_bucket, middle_file_key,                                 middle_file_path)  # upload the file's each word        delete_file_folder(src_file_path)        delete_file_folder(middle_file_path)        return upload_ret    else:        return -1def map_caller(event, context, cos_client):    appid = event['Records'][0]['cos']['cosBucket']['appid']    bucket = event['Records'][0]['cos']['cosBucket']['name'] + '-' + appid    key = event['Records'][0]['cos']['cosObject']['key']    key = key.replace('/' + str(appid) + '/' + event['Records'][0]['cos']['cosBucket']['name'] + '/', '', 1)    logger.info("Key is " + key)    middle_bucket = middle_stage_bucket + '-' + appid    middle_file_key = '/' + 'middle_' + key.split('/')[-1]    return do_mapping(cos_client, bucket, key, middle_bucket, middle_file_key)def main_handler(event, context):    logger.info("start main handler")    if "Records" not in event.keys():        return {"errorMsg": "event is not come from cos"}    secret_id = ""     secret_key = ""      config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, )    cos_client = CosS3Client(config)    start_time = datetime.datetime.now()    res = map_caller(event, context, cos_client)    end_time = datetime.datetime.now()    print("data mapping duration: " + str((end_time - start_time).microseconds / 1000) + "ms")    if res == 0:        return "Data mapping SUCCESS"    else:        return "Data mapping FAILED"

# -*- coding: utf8 -*-from qcloud_cos_v5 import CosConfigfrom qcloud_cos_v5 import CosS3Clientfrom qcloud_cos_v5 import CosServiceErrorfrom operator import itemgetterimport osimport sysimport datetimeimport loggingregion = u'ap-guangzhou'  # 根据实际情况，修改地域result_bucket = u'destmr'  # 根据实际情况，修改bucket名logging.basicConfig(level=logging.INFO, stream=sys.stdout)logger = logging.getLogger()logger.setLevel(level=logging.INFO)def delete_file_folder(src):    if os.path.isfile(src):        try:            os.remove(src)        except:            pass    elif os.path.isdir(src):        for item in os.listdir(src):            itemsrc = os.path.join(src, item)            delete_file_folder(itemsrc)        try:            os.rmdir(src)        except:            passdef download_file(cos_client, bucket, key, download_path):    logger.info("Get from [%s] to download file [%s]" % (bucket, key))    try:        response = cos_client.get_object(Bucket=bucket, Key=key, )        response['Body'].get_stream_to_file(download_path)    except CosServiceError as e:        print(e.get_error_code())        print(e.get_error_msg())        return -1    return 0def upload_file(cos_client, bucket, key, local_file_path):    logger.info("Start to upload file to cos")    try:        response = cos_client.put_object_from_local_file(            Bucket=bucket,            LocalFilePath=local_file_path,            Key='{}'.format(key))    except CosServiceError as e:        print(e.get_error_code())        print(e.get_error_msg())        return -1    logger.info("Upload data map file [%s] Success" % key)    return 0def qcloud_reducer(cos_client, bucket, key, result_bucket, result_key):    word2count = {}    src_file_path = u'/tmp/' + key.split('/')[-1]    result_file_path = u'/tmp/' + u'result_' + key.split('/')[-1]    download_ret = download_file(cos_client, bucket, key, src_file_path)    if download_ret == 0:        map_file = open(src_file_path, 'r')        result_file = open(result_file_path, 'w')        for line in map_file:            line = line.strip()            word, count = line.split('\t', 1)            try:                count = int(count)                word2count[word] = word2count.get(word, 0) + count            except ValueError:                logger.error("error value: %s, current line: %s" % (ValueError, line))                continue        map_file.close()        delete_file_folder(src_file_path)    sorted_word2count = sorted(word2count.items(), key=itemgetter(1))[::-1]    for wordcount in sorted_word2count:        res = '%s\t%s' % (wordcount[0], wordcount[1])        result_file.write(res)        result_file.write('\n')    result_file.close()    upload_ret = upload_file(cos_client, result_bucket, result_key, result_file_path)    delete_file_folder(result_file_path)    return upload_retdef reduce_caller(event, context, cos_client):    appid = event['Records'][0]['cos']['cosBucket']['appid']    bucket = event['Records'][0]['cos']['cosBucket']['name'] + '-' + appid    key = event['Records'][0]['cos']['cosObject']['key']    key = key.replace('/' + str(appid) + '/' + event['Records'][0]['cos']['cosBucket']['name'] + '/', '', 1)    logger.info("Key is " + key)    res_bucket = result_bucket + '-' + appid    result_key = '/' + 'result_' + key.split('/')[-1]    return qcloud_reducer(cos_client, bucket, key, res_bucket, result_key)def main_handler(event, context):    logger.info("start main handler")    if "Records" not in event.keys():        return {"errorMsg": "event is not come from cos"}    secret_id = "SecretId"     secret_key = "SecretKey"      config = CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key, )    cos_client = CosS3Client(config)    start_time = datetime.datetime.now()    res = reduce_caller(event, context, cos_client)    end_time = datetime.datetime.now()    print("data reducing duration: " + str((end_time - start_time).microseconds / 1000) + "ms")    if res == 0:        return "Data reducing SUCCESS"    else:        return "Data reducing FAILED"

WordCountMapper:  component: "@serverless/tencent-scf"  inputs:    name: mapper    codeUri: ./code    handler: index.main_handler    runtime: Python3.6    region: ap-guangzhou    description: 网站监控    memorySize: 64    timeout: 20    events:      - cos:          name: srcmr-1256773370.cos.ap-guangzhou.myqcloud.com          parameters:            bucket: srcmr-1256773370.cos.ap-guangzhou.myqcloud.com            filter:              prefix: ''              suffix: ''            events: cos:ObjectCreated:*            enable: true
WordCountReducer:  component: "@serverless/tencent-scf"  inputs:    name: reducer    codeUri: ./code    handler: index.main_handler    runtime: Python3.6    region: ap-guangzhou    description: 网站监控    memorySize: 64    timeout: 20    events:      - cos:          name: middlestagebucket-1256773370.cos.ap-guangzhou.myqcloud.com          parameters:            bucket: middlestagebucket-1256773370.cos.ap-guangzhou.myqcloud.com            filter:              prefix: ''              suffix: ''            events: cos:ObjectCreated:*            enable: true