modify

injection/data_sync/tasks.py

@@ -7,17 +7,22 @@ from trans2es.type_info import get_type_info_map
 #from rpc.all import get_rpc_remote_invoker
 from libs.es import ESPerform
 import logging
+import traceback
 
 @shared_task
 def write_to_es(es_type, pk_list, configuration, use_batch_query_set=False):
-    pk_list = list(frozenset(pk_list))
-    type_info_map = get_type_info_map()
-    type_info = type_info_map[es_type]
+    try:
+        pk_list = list(frozenset(pk_list))
+        type_info_map = get_type_info_map()
+        type_info = type_info_map[es_type]
+
+        logging.info("duan add,es_type:%s" % str(es_type))
+        type_info.insert_table_by_pk_list(
+            sub_index_name=es_type,
+            pk_list=pk_list,
+            use_batch_query_set=use_batch_query_set,
+            es=ESPerform.get_cli()
+        )
+    except:
+        logging.error("catch exception,err_msg:%s" % traceback.format_exc())
 
-    logging.info("duan add,es_type:%s" % str(es_type))
-    type_info.insert_table_by_pk_list(
-        sub_index_name=es_type,
-        pk_list=pk_list,
-        use_batch_query_set=use_batch_query_set,
-        es=ESPerform.get_cli()
-    )