#coding=utf-8 # evaluate为该方法的入口函数，必须用这个名字 from odps.udf import annotate from odps.distcache import get_cache_archive from odps.distcache import get_cache_file from odps.udf import BaseUDTF # 配置pandas依赖包 def include_package_path(res_name): import os, sys archive_files = get_cache_archive(res_name) dir_names = sorted([os.path.dirname(os.path.normpath(f.name)) for f in archive_files if '.dist_info' not in f.name], key=lambda v: len(v)) sys.path.append(dir_names[0]) return os.path.dirname(dir_names[0]) # 可能出现类似RuntimeError: xxx has been blocked by sandbox # 这是因为包含C的库，会被沙盘block，可设置set odps.isolation.session.enable = true def include_file(file_name): import os, sys so_file = get_cache_file(file_name) sys.path.append(os.path.dirname(os.path.abspath(so_file.name))) def include_so(file_name): import os, sys so_file = get_cache_file(file_name) with open(so_file.name, 'rb') as fp: content=fp.read() so = open(file_name, "wb") so.write(content) so.flush() so.close() #初始化业务数据包，由于上传限制,python版本以及archive解压包不统一等各种问题，需要手动导入 def init_env(list_files,package_name): import os,sys if len(list_files)==1: so_file = get_cache_file(list_files[0]) cmd_line = os.path.abspath(so_file.name) os.system("unzip -o %s -d %s"%(cmd_line,package_name)) elif len(list_files)>1: cmd_line = "cat" for _file in list_files: so_file = get_cache_file(_file) cmd_line += " "+os.path.abspath(so_file.name) cmd_line += " > temp.zip" os.system(cmd_line) os.system("unzip -o temp.zip -d %s"%(package_name)) # os.system("rm -rf %s/*.dist-info"%(package_name)) # return os.listdir(os.path.abspath("local_package")) # os.system("echo export LD_LIBRARY_PATH=%s >> ~/.bashrc"%(os.path.abspath("local_package"))) # os.system("source ~/.bashrc") sys.path.insert(0,os.path.abspath(package_name)) # sys.path.append(os.path.join(os.path.abspath("local_package"),"interface_real")) # UDF主程序 # 由于Series可能在多处调用，所以先在__init__中将其定义为全局类。 @annotate("string->string") class JiebaCut(object): def __init__(self): # zip_01 = include_package_path('testB01.zip') # zip_02 = include_package_path('testB02.zip') # self.cat_cmd = "cat %s %s > %s"%(zip_01+"/files/*",zip_02+"/files/*","testH.zip") # import os import sys # os.system(self.cat_cmd) # self.out = str(os.path.getsize("testH.zip")) # # self.out = str(os.listdir(zip_01+"/files/testB01/")) # os.system("mkdir jieba_t") # os.system("unzip testH.zip -d jieba_t") # self.out = str(os.listdir("jieba_t")) # sys.path.append(".") # # sys.path.append(os.path.dirname(os.path.normpath("jieba_test"))) # # import jieba_test # from jieba_t import cut include_package_path("jiebaA.zip") import jieba reload(sys) sys.setdefaultencoding('utf-8') global jieba def evaluate(self, x): import os # return self.out # return str(os.listdir("jieba_test")) # return self.cat_cmd return '--'.join(jieba.cut(x)) @annotate("string->string") class Preprocess(BaseUDTF): def __init__(self): # init_env(["gensim_package.zip.env"],"local_package1") import sys import uuid self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = include_package_path("envs_py37.env.zip") # self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Preprocessing as Preprocessing import BiddingKG.dl.entityLink.entityLink as entityLink import BiddingKG.dl.interface.getAttributes as getAttributes global Preprocessing,entityLink,predictor,uuid,getAttributes # import gensim # include_package_path("numpy.zip") # init_env(["tensorflow-1.14.0-cp37-cp37m-manylinux1_x86_64.whl"]) # so_file = get_cache_file("tensorflow-1.14.0-cp37-cp37m-manylinux1_x86_64.whl") # import os # self.out = os.path.abspath(so_file.name) # import tensorflow def process(self, x): k = str(uuid.uuid4()) list_articles = Preprocessing.get_preprocessed_articles([[k,x,"","_doc_id",""]],useselffool=True) self.forward(list_articles[0].toJson()) # list_articles,list_sentences,list_entitys,_cost_time = Preprocessing.get_preprocessed([[k,x,"","_doc_id",""]],useselffool=True) # # codeName = predictor.getPredictor("codeName").predict(list_sentences,MAX_AREA=2000,list_entitys=list_entitys) # # predictor.getPredictor("prem").predict(list_sentences,list_entitys) # # predictor.getPredictor("roleRule").predict(list_articles,list_sentences, list_entitys,codeName) # # predictor.getPredictor("epc").predict(list_sentences,list_entitys) # # entityLink.link_entitys(list_entitys) # # prem = getAttributes.getPREMs(list_sentences,list_entitys,list_articles) # # return str(self.out) # return "1" # list_articles,list_sentences,list_entitys,_ = Preprocessing.get_articles_processed([["doc_id",x,"","",""]],useselffool=True) # if len(list_articles)==1: # json_article = list_articles[0] # self.forward(list_sentences[0][0].sentence_text) @annotate("string -> string,string") class Preprocess_article(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import uuid import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Preprocessing as Preprocessing import BiddingKG.dl.entityLink.entityLink as entityLink import BiddingKG.dl.interface.getAttributes as getAttributes global Preprocessing,entityLink,predictor,uuid,getAttributes def process(self, x): if x is not None: k = str(uuid.uuid4()) list_articles = Preprocessing.get_preprocessed_article([[k,x,"","_doc_id",""]]) self.forward(list_articles[0].id,list_articles[0].toJson()) @annotate("string->string,string") class Preprocess_sentences(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.Preprocessing as Preprocessing import BiddingKG.dl.interface.Entitys as Entitys import json global Preprocessing,Entitys,json def process(self,x): _article = Entitys.Article.fromJson(x) list_sentences = Preprocessing.get_preprocessed_sentences([_article],True) list_out = [] for _sentence in list_sentences[0]: list_out.append(_sentence.toJson()) self.forward(_article.id,json.dumps(list_out)) @annotate("string->string,string") class Preprocess_entitys(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.Preprocessing as Preprocessing import BiddingKG.dl.interface.Entitys as Entitys import json global Preprocessing,Entitys,json def process(self,x): list_sentence = [] for _x in json.loads(x): list_sentence.append(Entitys.Sentences.fromJson(_x)) list_out = [] list_entitys = Preprocessing.get_preprocessed_entitys([list_sentence],True) for _entity in list_entitys[0]: list_out.append(_entity.toJson()) self.forward(list_sentence[0].doc_id,json.dumps(list_out)) @annotate("string->string,string") class Predict_codename(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Entitys as Entitys import json global predictor,Entitys,json def process(self,x): list_sentence = [] for _x in json.loads(x): list_sentence.append(Entitys.Sentences.fromJson(_x)) codename = predictor.getPredictor("codeName").predict([list_sentence],MAX_AREA=2000) self.forward(codename[0][0],json.dumps(codename[0])) @annotate("string,string->string,string") class Predict_role(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Entitys as Entitys import json global predictor,Entitys,json def process(self,x,y): list_sentence = [] list_entity = [] for _x in json.loads(x): list_sentence.append(Entitys.Sentences.fromJson(_x)) for _y in json.loads(y): list_entity.append(Entitys.Entity.fromJson(_y)) predictor.getPredictor("prem").predict_role([list_sentence],[list_entity]) list_out = [] for _entity in list_entity: if _entity.label is not None: list_out.append(_entity.toJson()) self.forward(list_sentence[0].doc_id,json.dumps(list_out)) @annotate("string,string->string,string") class Predict_money(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Entitys as Entitys import json global predictor,Entitys,json def process(self,x,y): list_sentence = [] list_entity = [] for _x in json.loads(x): list_sentence.append(Entitys.Sentences.fromJson(_x)) for _y in json.loads(y): list_entity.append(Entitys.Entity.fromJson(_y)) predictor.getPredictor("prem").predict_money([list_sentence],[list_entity]) list_out = [] for _entity in list_entity: if _entity.label is not None: list_out.append(_entity.toJson()) self.forward(list_sentence[0].doc_id,json.dumps(list_out)) @annotate("string,string->string,string") class Predict_person(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Entitys as Entitys import json global predictor,Entitys,json def process(self,x,y): list_sentence = [] list_entity = [] for _x in json.loads(x): list_sentence.append(Entitys.Sentences.fromJson(_x)) for _y in json.loads(y): list_entity.append(Entitys.Entity.fromJson(_y)) predictor.getPredictor("epc").predict_person([list_sentence],[list_entity]) list_out = [] for _entity in list_entity: if _entity.label is not None: list_out.append(_entity.toJson()) self.forward(list_sentence[0].doc_id,json.dumps(list_out)) @annotate("string,string,string,string,string,string,string->string,string,string") class ContentUnion(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") self.out = init_env(["BiddingKG.zip.env"],"local_package") self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") self.out = init_env(["envs_py37.zip.env"],"local_package") self.out = init_env(["so.env"],".") import BiddingKG.dl.interface.predictor as predictor import BiddingKG.dl.interface.Entitys as Entitys import BiddingKG.dl.interface.getAttributes as getAttributes import BiddingKG.dl.entityLink.entityLink as entityLink import BiddingKG.dl.interface.Preprocessing as Preprocessing import json global predictor,Entitys,getAttributes,entityLink,json,MyEncoder,Preprocessing #自定义jsonEncoder class MyEncoder(json.JSONEncoder): def __init__(self): import numpy as np global np def default(self, obj): if isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, bytes): return str(obj, encoding='utf-8') elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64)): return float(obj) return json.JSONEncoder.default(self, obj) def process(self,json_article,list_json_sentence,list_json_entity,list_json_entity_role,list_json_entity_money,list_json_entity_person,json_codename): dict_entity = {} list_sentence = [] list_entity = [] _article = Entitys.Article.fromJson(json_article) for list_json in [list_json_entity_role,list_json_entity_money,list_json_entity_person]: for _json_entity in json.loads(list_json): _entity = Entitys.Entity.fromJson(_json_entity) _key = "%s-%s-%s"%(str(_entity.doc_id),str(_entity.entity_id),str(_entity.entity_type)) dict_entity[_key] = _entity for _json_sentence in json.loads(list_json_sentence): list_sentence.append(Entitys.Sentences.fromJson(_json_sentence)) for _json_entity in json.loads(list_json_entity): _entity = Entitys.Entity.fromJson(_json_entity) _key = "%s-%s-%s"%(str(_entity.doc_id),str(_entity.entity_id),str(_entity.entity_type)) if _key in dict_entity: list_entity.append(dict_entity[_key]) else: list_entity.append(_entity) codeName = json.loads(json_codename) predictor.getPredictor("roleRule").predict([_article],[list_sentence], [list_entity],[codeName]) entityLink.link_entitys([list_entity]) prem = getAttributes.getPREMs([list_sentence],[list_entity],[_article]) # result = json.dumps(Preprocessing.union_result([codeName], prem)[0][1],cls=MyEncoder,sort_keys=True,indent=4,ensure_ascii=False) result = json.dumps(Preprocessing.union_result([codeName], prem)[0][1],ensure_ascii=False) self.forward(_article.id,_article.doc_id,result) @annotate("string,bigint,string,string->string,bigint,string") class Extract(BaseUDTF): def __init__(self): # self.out = init_env(["BiddingKG.z01","BiddingKG.z02"],"local_package") import uuid global uuid import logging import datetime logging.basicConfig(level = logging.INFO,format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s') logging.info("time1"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) self.out = init_env(["BiddingKG.zip.env"],str(uuid.uuid4())) logging.info("time2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) self.out = init_env(["wiki_128_word_embedding_new.vector.env"],".") logging.info("time3"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) # self.out = init_env(["envs_py37.zip.env"],str(uuid.uuid4())) self.out = include_package_path("envs_py37.env.zip") logging.info("time4"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) self.out = init_env(["so.env"],".") logging.info("time5"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import BiddingKG.dl.interface.predictor as predictor logging.info("time6"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import BiddingKG.dl.interface.Entitys as Entitys logging.info("time6.1"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import BiddingKG.dl.interface.getAttributes as getAttributes logging.info("time6.2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import BiddingKG.dl.entityLink.entityLink as entityLink logging.info("time6.2"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import BiddingKG.dl.interface.Preprocessing as Preprocessing logging.info("time6.3"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import json import time from BiddingKG.dl.common.Utils import log logging.info("time7"+str(datetime.datetime.now().strftime('%y-%m-%d %H:%M:%S'))) import numpy as np global predictor,Entitys,getAttributes,entityLink,json,MyEncoder,Preprocessing,time,log,MyEncoder,np class MyEncoder(json.JSONEncoder): def default(self, obj): if isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, bytes): return str(obj, encoding='utf-8') elif isinstance(obj, (np.float_, np.float16, np.float32, np.float64)): return float(obj) elif isinstance(obj,(np.int64)): return int(obj) return json.JSONEncoder.default(self, obj) def process(self,content,_doc_id,_title,page_time): if content is not None and _doc_id not in [105677700,126694044,126795572,126951461]: k = str(uuid.uuid4()) cost_time = dict() start_time = time.time() log("start process doc %s"%(str(_doc_id))) try: list_articles,list_sentences,list_entitys,_cost_time = Preprocessing.get_preprocessed([[k,content,"",str(_doc_id),str(_title)]],useselffool=True) log("get preprocessed done of doc_id%s"%(_doc_id)) cost_time["preprocess"] = time.time()-start_time cost_time.update(_cost_time) ''' for articles in list_articles: print(articles.content) ''' start_time = time.time() codeName = predictor.getPredictor("codeName").predict(list_sentences,MAX_AREA=2000,list_entitys=list_entitys) log("get codename done of doc_id%s"%(_doc_id)) cost_time["codename"] = time.time()-start_time start_time = time.time() predictor.getPredictor("prem").predict(list_sentences,list_entitys) log("get prem done of doc_id%s"%(_doc_id)) cost_time["prem"] = time.time()-start_time start_time = time.time() predictor.getPredictor("roleRule").predict(list_articles,list_sentences, list_entitys,codeName) cost_time["rule"] = time.time()-start_time start_time = time.time() predictor.getPredictor("epc").predict(list_sentences,list_entitys) log("get epc done of doc_id%s"%(_doc_id)) cost_time["person"] = time.time()-start_time start_time = time.time() entityLink.link_entitys(list_entitys) ''' for list_entity in list_entitys: for _entity in list_entity: for _ent in _entity.linked_entitys: print(_entity.entity_text,_ent.entity_text) ''' prem = getAttributes.getPREMs(list_sentences,list_entitys,list_articles) log("get attributes done of doc_id%s"%(_doc_id)) cost_time["attrs"] = time.time()-start_time #print(prem) data_res = Preprocessing.union_result(codeName, prem)[0][1] data_res["cost_time"] = cost_time data_res["success"] = True _article = list_articles[0] self.forward(page_time,int(_article.doc_id),json.dumps(data_res,cls=MyEncoder,ensure_ascii=False)) except Exception as e: log("%s===error docid:%s"%(str(e),str(_doc_id)))