algorithm_platform/data_processing/data_operation/hive_to_mongo.py

from flask import Flask,request,jsonify
import time
import logging
import traceback
from common.database_dml import insert_data_into_mongo, get_xmo_data_from_hive, delete_data_from_mongo

app = Flask('hive_to_mongo——service')


@app.route('/hello', methods=['POST'])
def hello():
    return jsonify(message='Hello, World!')


@app.route('/hive_to_mongo', methods=['POST'])
def data_join():
    # 获取程序开始时间  
    start_time = time.time()  
    result = {}
    success = 0
    print("Program starts execution!")
    try:
        args = request.values.to_dict()
        print('args', args)
        logger.info(args)
        delete_data_from_mongo(args)
        df_hive = get_xmo_data_from_hive(args)
        if isinstance(df_hive, str):
            success = 0
            result['msg'] = df_hive
        else:
            insert_data_into_mongo(df_hive, args)
            success = 1
    except Exception as e:
        my_exception = traceback.format_exc()
        my_exception.replace("\n", "\t")
        result['msg'] = my_exception
    end_time = time.time()
    result['success'] = success
    result['args'] = args
    result['start_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(start_time))
    result['end_time'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(end_time))
    print("Program execution ends!")
    return result


if __name__ == "__main__":
    print("Program starts execution!")
    logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
    logger = logging.getLogger("hive_to_mongo")
    from waitress import serve
    serve(app, host="0.0.0.0", port=10127)
    print("server start!")