标签跑完 第一版。

src/main/java/com/platomix/userprofile/Application.java

@@ -33,8 +33,7 @@ public class Application {
 	static final String[] classHandlers=new String[]
 			{ "xxhLongTagHandler", "xxhTagHandler", "xxhAppTagHandler", "xxhOrderTagHandler",
 					"userOrderStateTagHandler", "orderSumDateTagHandler", "appPreferenceTagHandler",
-					"complainTagHandler","forecastCancelTagHandler", "forecastOrderTagHandler",
-					"collectAllTagHandler"};
+					"complainTagHandler","forecastCancelTagHandler", "forecastOrderTagHandler"};
 	//运行时需要取前2个月的handler
 	static final String[] classPerHandlers=new String[]
 			{ "xxhTagHandler", "xxhAppTagHandler","appPreferenceTagHandler"};
@@ -48,21 +47,20 @@ public class Application {
 	 */
 	public static void main(String[] args) {
 		ConfigurableApplicationContext applicationContext = SpringApplication.run(Application.class, args);
-//		XxhTagManager xxhTagManager = applicationContext.getBean(XxhTagManager.class);
-//		xxhTagManager.xxhFromEtlSp("202204");
-//		AbstractTagHandler tagHandler = applicationContext.getBean("forecastCancelTagHandler",AbstractTagHandler.class);
+
+//		AbstractTagHandler tagHandler = applicationContext.getBean("xxhLongTagHandler",AbstractTagHandler.class);
 //		tagHandler.handle("202204");
 		//manager.tagCancelDataC4V2("202205");
-//		AbstractTagHandler tagHandler = applicationContext.getBean("orderSumDateTagHandler", AbstractTagHandler.class);
-//		tagHandler.handle("202111");
+//		AbstractTagHandler tagHandler = applicationContext.getBean("appPreferenceTagHandler", AbstractTagHandler.class);
+//		tagHandler.handle("202204");
 
 		//批量
 //		tagHandler.handleMultiMonth("202111", false, 3);
 //		//Flat表操作
 //		FlatTagHandler tagHandler = applicationContext.getBean(FlatTagHandler.class);
-//		tagHandler.tempDataToFlatView("202111");
+//		tagHandler.tempDataToFlatView("202204");
 		//运行程序
-		StartProg(applicationContext,args);
+		startProg(applicationContext,args);
 		applicationContext.close();
 	}
 
@@ -70,7 +68,7 @@ public class Application {
 	 * 根据args参数
 	 * @param args
 	 */
-	static void StartProg(ConfigurableApplicationContext applicationContext ,String[] args){
+	static void startProg(ConfigurableApplicationContext applicationContext ,String[] args){
 		//arg0 调用类别
 		String sendTp = "";
 		//arg1 月份
@@ -82,63 +80,73 @@ public class Application {
 			sendTp = args[0];
 			if (args.length > 1) {
 				permonth = args[1];
-				p_permonth = DateUtil.format(DateUtil.offsetMonth(DateUtil.parse(permonth.substring(0, 4) + "-" + permonth.substring(4) + "-01", "yyyy-MM-dd"), -1), "yyyyMM");
+				//传值不需要 向前取一个月
+				p_permonth=permonth;
+				//p_permonth = DateUtil.format(DateUtil.offsetMonth(DateUtil.parse(permonth.substring(0, 4) + "-" + permonth.substring(4) + "-01", "yyyy-MM-dd"), -1), "yyyyMM");
 			}
 			if (args.length > 2) {
 				handlerName = args[2];
 			}
 		}
 
-		/**
-		 * 生成指定月份全部标签（不传月份参数会默认跑前一个月的标签）
-		 */
-		if (sendTp.equals("zt_tag")) {
-			for (Integer i = 0; i < classHandlers.length; i++) {
-				AbstractTagHandler tagHandler = applicationContext.getBean(classHandlers[i], AbstractTagHandler.class);
-				if (Arrays.asList(classPerHandlers).contains(classHandlers[i])) {
-					tagHandler.handle(p_permonth);
-				} else {
-					tagHandler.handle(permonth);
+		try{
+			/**
+			 * 生成指定月份全部标签（不传月份参数会默认跑前一个月的标签）
+			 */
+			if (sendTp.equals("zt_tag")) {
+				for (Integer i = 0; i < classHandlers.length; i++) {
+					AbstractTagHandler tagHandler = applicationContext.getBean(classHandlers[i], AbstractTagHandler.class);
+					if (Arrays.asList(classPerHandlers).contains(classHandlers[i])) {
+						tagHandler.handle(p_permonth);
+					} else {
+						tagHandler.handle(permonth);
+					}
 				}
+				//生成rfm表 todo
 			}
-			//生成rfm表 todo
-		}
-		/**
-		 * 单独调用指定handler
-		 */
-		if (sendTp.equals("single_handler")) {
-			AbstractTagHandler tagHandler = applicationContext.getBean(handlerName, AbstractTagHandler.class);
-			tagHandler.handle(permonth);
-		}
-		/**
-		 * 更新mysql中的字典表
-		 */
-		if (sendTp.equals("dic_tag")) {
-			UserEventManager userEventHandler = applicationContext.getBean(UserEventManager.class);
-			userEventHandler.genEventPropToMysql();
-			UserTagManager manager = applicationContext.getBean(UserTagManager.class);
-			manager.genSomeTagToMysql();
-		}
-		/**
-		 * 更新etl2->xxh
-		 */
-		if (sendTp.equals("xxh_etl")) {
-			XxhTagManager xxhTagManager = applicationContext.getBean(XxhTagManager.class);
-			xxhTagManager.xxhFromEtlSp(permonth);
-		}
-		/**
-		 * 退订订ding  todo
-		 */
-		//退订数据准备  先调用这个准备数据，然后取TensorFlow训练，并预测 ？月结果
-		if (sendTp.equals("z_cancel_data")) {
-			ForecastCancelTagManager manager = applicationContext.getBean(ForecastCancelTagManager.class);
-			manager.tagCancelDataC4(permonth,-1,true);
-			manager.tagCancelDataC4V2(permonth);
+			/**
+			 * 单独调用指定handler
+			 */
+			if (sendTp.equals("single_handler")) {
+				AbstractTagHandler tagHandler = applicationContext.getBean(handlerName, AbstractTagHandler.class);
+				tagHandler.handle(permonth);
+			}
+			/**
+			 * 更新mysql中的字典表
+			 */
+			if (sendTp.equals("dic_tag")) {
+				UserEventManager userEventHandler = applicationContext.getBean(UserEventManager.class);
+				userEventHandler.genEventPropToMysql();
+				UserTagManager manager = applicationContext.getBean(UserTagManager.class);
+				manager.genSomeTagToMysql();
+			}
+			/**
+			 * 更新etl2->xxh
+			 */
+			if (sendTp.equals("xxh_etl")) {
+				XxhTagManager xxhTagManager = applicationContext.getBean(XxhTagManager.class);
+				xxhTagManager.xxhFromEtlSp(permonth);
+			}
+			/**
+			 * 退订订ding  todo
+			 */
+			//退订数据准备  先调用这个准备数据，然后取TensorFlow训练，并预测 ？月结果
+			if (sendTp.equals("z_cancel_data")) {
+				ForecastCancelTagManager manager = applicationContext.getBean(ForecastCancelTagManager.class);
+				manager.tagCancelDataC4(permonth,-1,true);
+				manager.tagCancelDataC4V2(permonth);
+			}
+			if(sendTp.equals("collect_all_tag")){
+				applicationContext.getBean("collectAllTagHandler", AbstractTagHandler.class).handle(permonth);
+			}
+			if(sendTp.equals("join_all")){
+				FlatTagHandler flatTagHandler = applicationContext.getBean(FlatTagHandler.class);
+				flatTagHandler.tempDataToFlatView(permonth);
+			}
+		}catch (Exception e){
+			System.out.println(e.getMessage());
+			System.out.println(e.getStackTrace());
 		}
 
-		if(sendTp.equals("join_all")){
-			FlatTagHandler flatTagHandler = applicationContext.getBean(FlatTagHandler.class);
-			flatTagHandler.tempDataToFlatView(permonth);
-		}
 	}
 }

src/main/java/com/platomix/userprofile/config/MybatisPlusConfig.java

@@ -34,15 +34,15 @@ public class MybatisPlusConfig {
 		tableNameHandlerMap.put("ads.DIC_USERTAG", (sql, tableName) -> {
 			return useDebug ? "ads.DIC_USERTAG_DEV" : tableName;
 		});
-		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_temp", (sql, tableName) -> {
-			return useDebug ? "ads.Z_USER_TAG_FLAT_temp_dev" : tableName;
-		});
-		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_clu1", (sql, tableName) -> {
-			return useDebug ? "ads.Z_USER_TAG_FLAT_clu1_dev" : tableName;
-		});
-		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_ALL", (sql, tableName) -> {
-			return useDebug ? "ads.Z_USER_TAG_FLAT_ALL_dev" : tableName;
-		});
+//		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_temp", (sql, tableName) -> {
+//			return useDebug ? "ads.Z_USER_TAG_FLAT_temp_dev" : tableName;
+//		});
+//		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_clu1", (sql, tableName) -> {
+//			return useDebug ? "ads.Z_USER_TAG_FLAT_clu1_dev" : tableName;
+//		});
+//		tableNameHandlerMap.put("ads.Z_USER_TAG_FLAT_ALL", (sql, tableName) -> {
+//			return useDebug ? "ads.Z_USER_TAG_FLAT_ALL_dev" : tableName;
+//		});
 		dynamicTableNameInnerInterceptor.setTableNameHandlerMap(tableNameHandlerMap);
 		interceptor.addInnerInterceptor(dynamicTableNameInnerInterceptor);
 		return interceptor;

src/main/java/com/platomix/userprofile/handler/AbstractTagHandler.java

@@ -110,4 +110,18 @@ public abstract class AbstractTagHandler implements ITagHandler{
 			return String.format("create table %s ENGINE = MergeTree %s ORDER BY %s as", tableName, partition, orderBy);
 		}
 	}
+	/**
+	 * 构建创建表或者插入表SQL
+	 * @param tableName
+	 * @param orderBy
+	 * @return
+	 */
+	protected String buildTableSql(String tableName, String orderBy) {
+		boolean isExists = dynamicExecuteService.checkTableExistsOfClickhouse(tableName);
+		if(isExists) {
+			return String.format("insert into %s", tableName);
+		}else {
+			return String.format("create table %s ENGINE = MergeTree ORDER BY %s as", tableName, orderBy);
+		}
+	}
 }

src/main/java/com/platomix/userprofile/handler/FlatTagHandler.java

@@ -17,6 +17,7 @@ import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
+import com.platomix.userprofile.utils.TagKit;
 import org.apache.commons.io.FileUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -54,8 +55,6 @@ public class FlatTagHandler {
 
 	final static String[] INVALID_FIELD_NAMES = new String[] { "USERID", "MONTH", "DEVICE_NUMBER", "cTime" };
 
-	@Value("${dflow.use-debug}")
-	private Boolean useDebug;
 
 	@Value("${dbcluster.clickhouse}")
 	private String[] clickhouseDbCluster;
@@ -77,7 +76,7 @@ public class FlatTagHandler {
 	 */
 	public void tempDataToFlatView(String month) {
 		//校验并处理列变动
-		_checkAndDealColumn();
+		_checkAndDealColumn(month);
 		
 		//删除月份分区数据
 		String dropViewDataSql = String.format("alter table ads.%s on cluster %s drop partition '%s'", cluTableName, CLUSTER_NAME, month);
@@ -85,7 +84,7 @@ public class FlatTagHandler {
 		dynamicExecuteService.execute(DsEnum.clickhouse, dropViewDataSql, "删除分区[month="+month+"]数据完成");
 		
 		//再次查询列变动后的字段名
-		List<JSONObject> checkTableDiffList = _getDiffColumnList();
+		List<JSONObject> checkTableDiffList = _getDiffColumnList(month);
 		List<String> columnList = new ArrayList<>();
 		for(JSONObject columnObj : checkTableDiffList) {
 			String tempName = columnObj.getString("temp_name"),
@@ -116,49 +115,49 @@ public class FlatTagHandler {
 	 */
 	@Deprecated
 	public void tmpDataToFlat(String month) {
-		List<JSONObject> sourceColumnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name from system.columns where `table`='" + tempTableName + "'");
-		List<String> sourceColumnNameList = sourceColumnList.stream().map(item -> item.getString("name")).collect(Collectors.toList());
-
-		List<JSONObject> targetColumnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name from system.columns where `table`='" + cluTableName + "'");
-		Map<String, String> targetColumnMap = new HashMap<String, String>();
-		for (JSONObject jsonObject : targetColumnList) {
-			targetColumnMap.put(jsonObject.getString("name"), jsonObject.getString("type"));
-		}
-
-		StrBuilder sqlBuilder = new StrBuilder();
-		sqlBuilder.appendLn("insert into " + flatAllViewName + "(");
-		targetColumnMap.keySet().stream().filter(f -> sourceColumnNameList.contains(f)).map(item -> "`" + item + "`")
-				.forEach(sqlBuilder::appendLn);
-		sqlBuilder.appendLn(")").appendLn("with");
-		// 省份
-		String provinceSql = userprofileTagService.genSelectSql("省份", "b.PROVINCE", null, "TMP_PROVINCE");
-		sqlBuilder.append(provinceSql).appendLn(",");
-		// 城市
-		String citySql = userprofileTagService.genSelectSql("城市", "b.AREA", null, "TMP_AREA");
-		sqlBuilder.appendLn(citySql);
-
-		sqlBuilder.appendLn("select THEMONTH_V,USERID_V,");
-		for (Entry<String, String> entry : targetColumnMap.entrySet()) {
-			String name = entry.getKey(), type = entry.getValue();
-			// 包含一些特殊字段/或者源表中不存在该字段(大宽表比临时表多的字段)，不处理
-			if(Stream.of(INVALID_FIELD_NAMES).anyMatch(f -> name.contains(f))
-					|| !sourceColumnNameList.contains(name)) {
-				continue;
-			}
-			if(	// 投诉标签处理
-				_handleComplain(sqlBuilder, name) ||
-				// 省份标签处理
-				_handleProvince(sqlBuilder, name) ||
-				// 城市标签处理
-				_handleCity(sqlBuilder, name) ||
-				//其它标签处理
-				_handleOthers(sqlBuilder, name, type)) {
-			}
-		}
-		sqlBuilder.deleteLastChar().deleteLastChar();
-		sqlBuilder.appendLn("\nfrom ads." + tempTableName + " a left join ETL_VW_TB_CALLER_AREACODE b on substring(a.USERID_V , 1, 7) = b.CALLERHEAD");
-		log.info("\n" + sqlBuilder.toString());
-		//dynamicExecuteService.execute(DsEnum.clickhouse, sqlBuilder.toString(), "临时表导入视图生成");
+//		List<JSONObject> sourceColumnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name from system.columns where `table`='" + tempTableName + "'");
+//		List<String> sourceColumnNameList = sourceColumnList.stream().map(item -> item.getString("name")).collect(Collectors.toList());
+//
+//		List<JSONObject> targetColumnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name from system.columns where `table`='" + cluTableName + "'");
+//		Map<String, String> targetColumnMap = new HashMap<String, String>();
+//		for (JSONObject jsonObject : targetColumnList) {
+//			targetColumnMap.put(jsonObject.getString("name"), jsonObject.getString("type"));
+//		}
+//
+//		StrBuilder sqlBuilder = new StrBuilder();
+//		sqlBuilder.appendLn("insert into " + flatAllViewName + "(");
+//		targetColumnMap.keySet().stream().filter(f -> sourceColumnNameList.contains(f)).map(item -> "`" + item + "`")
+//				.forEach(sqlBuilder::appendLn);
+//		sqlBuilder.appendLn(")").appendLn("with");
+//		// 省份
+//		String provinceSql = userprofileTagService.genSelectSql("省份", "b.PROVINCE", null, "TMP_PROVINCE");
+//		sqlBuilder.append(provinceSql).appendLn(",");
+//		// 城市
+//		String citySql = userprofileTagService.genSelectSql("城市", "b.AREA", null, "TMP_AREA");
+//		sqlBuilder.appendLn(citySql);
+//
+//		sqlBuilder.appendLn("select THEMONTH_V,USERID_V,");
+//		for (Entry<String, String> entry : targetColumnMap.entrySet()) {
+//			String name = entry.getKey(), type = entry.getValue();
+//			// 包含一些特殊字段/或者源表中不存在该字段(大宽表比临时表多的字段)，不处理
+//			if(Stream.of(INVALID_FIELD_NAMES).anyMatch(f -> name.contains(f))
+//					|| !sourceColumnNameList.contains(name)) {
+//				continue;
+//			}
+//			if(	// 投诉标签处理
+//				_handleComplain(sqlBuilder, name) ||
+//				// 省份标签处理
+//				_handleProvince(sqlBuilder, name) ||
+//				// 城市标签处理
+//				_handleCity(sqlBuilder, name) ||
+//				//其它标签处理
+//				_handleOthers(sqlBuilder, name, type)) {
+//			}
+//		}
+//		sqlBuilder.deleteLastChar().deleteLastChar();
+//		sqlBuilder.appendLn("\nfrom ads." + tempTableName + " a left join ETL_VW_TB_CALLER_AREACODE b on substring(a.USERID_V , 1, 7) = b.CALLERHEAD");
+//		log.info("\n" + sqlBuilder.toString());
+//		//dynamicExecuteService.execute(DsEnum.clickhouse, sqlBuilder.toString(), "临时表导入视图生成");
 	}
 
 	/**
@@ -172,9 +171,9 @@ public class FlatTagHandler {
 		//按手机号 查找分段
 		Set<String> setps=new HashSet<>();
 		StrBuilder sqlBuilder=new StrBuilder();
-		sqlBuilder.appendLn(String.format("with intDiv((select COUNT() from ads.%s),%d) as step",tempTableName,clickhouseDbCluster.length))
+		sqlBuilder.appendLn(String.format("with intDiv((select COUNT() from ads.%s),%d) as step", TagKit.getTagTempTable(tempTableName,month),clickhouseDbCluster.length))
 				.appendLn("select USERID_V,(rowNumberInAllBlocks()+1) rownum from (")
-				.appendLn(String.format("select USERID_V from ads.%s order by USERID_V ",tempTableName)+")")
+				.appendLn(String.format("select USERID_V from ads.%s order by USERID_V ",TagKit.getTagTempTable(tempTableName,month))+")")
 				.append("where rownum in (");
 		IntStream.range(1, clickhouseDbCluster.length).forEachOrdered(idx -> {setps.add(String.format("step*%d",idx));});
 		sqlBuilder.append(String.join(",", setps));
@@ -197,13 +196,13 @@ public class FlatTagHandler {
 							.appendLn(")")
 							.appendLn("select")
 							.appendLn(String.join(",", columnList))
-							.appendLn("\nfrom ads." + tempTableName)
+							.appendLn("\nfrom ads." + TagKit.getTagTempTable(tempTableName,month))
 							.appendLn("where THEMONTH_V = '"+month+"' "+userIdSql);
 					_shellToCluster(new ShellCommond(clickhouseDbCluster[i],i==0,builder.toString()));
 				}else{
 					builder.append("echo 'select ")
 							.append(String.join(",", columnList))
-							.append(String.format(" from ads.%s where THEMONTH_V = toString(%s) %s' | ",tempTableName,month,userIdSql))
+							.append(String.format(" from ads.%s where THEMONTH_V = toString(%s) %s' | ",TagKit.getTagTempTable(tempTableName,month),month,userIdSql))
 							.append(String.format(" curl %s:8123?database=ads -u%s:%s -d @- | ",firstDb,clickhouseUserName,clickhousePassWord))
 							.append(String.format(" docker run -i --rm yandex/clickhouse-client --host %s --port 9007 --user %s --password %s --input_format_allow_errors_num=100",clickhouseDbCluster[i],clickhouseUserName,clickhousePassWord))
 							.append(String.format(" --query='INSERT INTO ads.%s( ",cluTableName))
@@ -243,8 +242,8 @@ public class FlatTagHandler {
 	/**
 	 * 创建Flat表 [初始化执行一次]
 	 */
-	public void initCreateFlatTable() {
-		List<JSONObject> columnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name,`type` from system.columns where `table`='" + tempTableName + "'");
+	public void initCreateFlatTable(String month) {
+		List<JSONObject> columnList = dynamicExecuteService.list(DsEnum.clickhouse, "select name,`type` from system.columns where `table`='" + TagKit.getTagTempTable(tempTableName,month) + "'");
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn("create table ads." + cluTableName + " on cluster " + CLUSTER_NAME + " (")
 				  .appendLn("`THEMONTH_V` String, ").appendLn("`USERID_V` String,");
@@ -264,8 +263,8 @@ public class FlatTagHandler {
 	/**
 	 * 校验字段列是否有改变
 	 */
-	private void _checkAndDealColumn() {
-		List<JSONObject> checkTableDiffList = _getDiffColumnList();
+	private void _checkAndDealColumn(String month) {
+		List<JSONObject> checkTableDiffList = _getDiffColumnList(month);
 		List<String> addColumns = new ArrayList<>();
 //		List<String> delColumns = new ArrayList<>();
 		for(JSONObject dataObj : checkTableDiffList) {
@@ -284,7 +283,7 @@ public class FlatTagHandler {
 			//是否需要增删字段； 是->先drop视图
 			if(!addColumns.isEmpty()) { //|| !delColumns.isEmpty()
 				//1.drop视图
-				String dropViewSql = String.format("drop table %s ON cluster %s", flatAllViewName, CLUSTER_NAME);
+				String dropViewSql = String.format("drop table IF EXISTS %s ON cluster %s", flatAllViewName, CLUSTER_NAME);
 				log.info("\n" + dropViewSql);
 				dynamicExecuteService.execute(DsEnum.clickhouse, dropViewSql, "删除视图完成");
 				log.info("删除视图完成，开始等待8s处理增删列...");
@@ -307,12 +306,12 @@ public class FlatTagHandler {
 		}
 	}
 	
-	private List<JSONObject> _getDiffColumnList(){
+	private List<JSONObject> _getDiffColumnList(String month){
 		StrBuilder checkTableDiffSql = new StrBuilder()
 				.appendLn("select temp.name as temp_name,temp.type as tp,clu.name as clu_name from")
-				.appendLn("(select name,`type` from system.columns where `table`='" + tempTableName +(useDebug==true?"_dev":"") +"')temp")
+				.appendLn("(select name,`type` from system.columns where `table`='" + TagKit.getTagTempTable(tempTableName,month)  +"')temp")
 				.appendLn("full outer join")
-				.appendLn("(select name,`type` from system.columns where `table`='" + cluTableName +(useDebug==true?"_dev":"")+ "')clu")
+				.appendLn("(select name,`type` from system.columns where `table`='" + cluTableName + "')clu")
 				.appendLn("on temp.name=clu.name");
 		return dynamicExecuteService.list(DsEnum.clickhouse, checkTableDiffSql.toString());
 	}

src/main/java/com/platomix/userprofile/handler/core/ActivityJoinTagHandler.java

@@ -38,7 +38,7 @@ public class ActivityJoinTagHandler extends AbstractTagHandler {
 		
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_ACT", "USERID_ACT"));
-		sqlBuilder.append("select MEMBER_PHONE as USERID_ACT,").append("'"+month+"'").appendLn(" as THEMONTH_ACT,");
+		sqlBuilder.append("select").append("'"+month+"'").appendLn(" as THEMONTH_ACT,").append(" MEMBER_PHONE as USERID_ACT,").append("'"+month+"'").appendLn(" as THEMONTH_ACT,");
 
 		String activitySql = userprofileTagService.genSelectSql("参与活动", "title", "ACTIVITY_UUID", null);
 		sqlBuilder.appendLn(activitySql);

src/main/java/com/platomix/userprofile/handler/core/AppPreferenceTagHandler.java

@@ -51,17 +51,20 @@ public class AppPreferenceTagHandler extends AbstractTagHandler {
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_APP_PREF", "USERID_APP_PREF"));
 		sqlBuilder.appendLn("with NULLIF(APP_NAME, '') as APPNAME")
-				    .append("select DEVICE_NUMBER as USERID_APP_PREF,").append("'"+month+"'").appendLn(" as THEMONTH_APP_PREF,");
+				    .append("select").append("'"+month+"'").appendLn(" as THEMONTH_APP_PREF,").append("DEVICE_NUMBER as USERID_APP_PREF,");
 		
 		//格式化数据
-		List<String> tagIdList = tagList.stream().map(UserprofileTag::getTagId).collect(Collectors.toList());
 		Map<String, String> tagName2IdMap = tagList.stream().collect(Collectors.toMap(UserprofileTag::getTagName, UserprofileTag::getTagId, (t, t1) -> t));
-		String tagIdListStr = JSON.toJSONString(tagIdList).replaceAll("\"", "\'"), 
+//		List<String> tagIdList = tagList.stream().map(UserprofileTag::getTagId).collect(Collectors.toList());
+//		String tagIdListStr = JSON.toJSONString(tagIdList).replaceAll("\"", "\'"),
+//				tagName2IdMapStr = JSON.toJSONString(tagName2IdMap);
+		List<String> tagNameList = tagList.stream().map(UserprofileTag::getTagName).collect(Collectors.toList());
+		String tagNameListStr = JSON.toJSONString(tagNameList).replaceAll("\"", "\'"),
 				tagName2IdMapStr = JSON.toJSONString(tagName2IdMap);
 		
-		sqlBuilder.appendLn("arrayMap(x -> JSONExtractString('"+tagName2IdMapStr+"', x), TAG_APP_PREF_C) TAG_APP_PREF,")
-				  .appendLn("groupUniqArray(if(has("+tagIdListStr+", APPNAME) = 1, APPNAME, '其他')) as TAG_APP_PREF_C,")
-				  .appendLn("TAG_APP_PREF_C as TAG_APP_PREF_V")
+		sqlBuilder.appendLn("arrayMap(x -> JSONExtractString('"+tagName2IdMapStr+"', x), `TAG_APP_PREF.C`) `TAG_APP_PREF.E`,")
+				  .appendLn("groupUniqArray(if(has("+tagNameListStr+", APPNAME) = 1, APPNAME, '其他')) as `TAG_APP_PREF.C`,")
+				  .appendLn("`TAG_APP_PREF.C` as `TAG_APP_PREF.V`")
 				  .appendLn("from ads.XXH_ALL_APP_MONTH")
 				  .appendLn("where MONTH_ID >= '"+threeMonthAgo+"' and MONTH_ID <= '"+month+"' and ONE_FLUX > 1024")
 				  .appendLn("group by DEVICE_NUMBER");

src/main/java/com/platomix/userprofile/handler/core/CollectAllTagHandler.java

@@ -11,6 +11,8 @@ import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 
+import com.platomix.userprofile.utils.TagKit;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.util.StringUtils;
 
@@ -30,15 +32,16 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 @Service
 public class CollectAllTagHandler extends AbstractTagHandler{
-	
-	final static String ZT_TABLE_NAME = "ads.Z_USER_TAG_FLAT_temp";
+
+	@Value("${dflow.flat-temp-table-name}")
+	private String tempTableName;
 	/**
 	 * before invoke 执行
 	 * @return
 	 */
 	@Override
 	public void beforeInvoke(String month) {
-		//dynamicExecuteService.execute(DsEnum.clickhouse, String.format("drop table if exists %s",ZT_TABLE_NAME), "删除temp原始数据");
+		dynamicExecuteService.execute(DsEnum.clickhouse, String.format("drop table if exists %s", TagKit.getTagTempTable(tempTableName,month)), "删除temp原始数据");
 		try {
 			TimeUnit.SECONDS.sleep(5);
 		} catch (InterruptedException e) {
@@ -48,7 +51,7 @@ public class CollectAllTagHandler extends AbstractTagHandler{
 	@Override
 	public void invoke(String month, String mobileLeft, String mobileRight) {
 		StrBuilder sqlBuilder = new StrBuilder();
-		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_V", "USERID_V"));
+		sqlBuilder.appendLn(super.buildTableSql(TagKit.getTagTempTable(tempTableName,month), "THEMONTH_V", "USERID_V"));
 		sqlBuilder.appendLn("select '"+month+"' as THEMONTH_V,");
 		//动态拼接SQL
 		List<JSONObject> dataList = dynamicExecuteService.list(DsEnum.clickhouse, "select TABLE_NAME,JOINSTR,JOIN_TABLE from ads.CFG_ZT_TAG_JOIN order by id");

src/main/java/com/platomix/userprofile/handler/core/ComplainTagHandler.java

@@ -46,7 +46,7 @@ public class ComplainTagHandler extends AbstractTagHandler {
 		
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_COMPLAIN", "USERID_COMPLAIN"));
-		sqlBuilder.append("select ACCEPT_PHONENUM as USERID_COMPLAIN,").append("'"+month+"'").appendLn(" as THEMONTH_COMPLAIN,");
+		sqlBuilder.append("select").append("'"+month+"'").appendLn(" as THEMONTH_COMPLAIN,").append("ACCEPT_PHONENUM as USERID_COMPLAIN,");
 		//标签范围值
 		/**
 		 * (case when TAG_COMPLAIN_V = 0 then 'B112_22_001'

src/main/java/com/platomix/userprofile/handler/core/ForecastCancelTagHandler.java

@@ -31,7 +31,7 @@ public class ForecastCancelTagHandler extends AbstractTagHandler {
 	public void invoke(String month, String mobileLeft, String mobileRight) {
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_CEL", "USERID_CEL"));
-		sqlBuilder.append("select USERID as USERID_CEL,").append("'"+month+"'").appendLn(" as THEMONTH_CEL,");
+		sqlBuilder.append("select").append("'"+month+"'").appendLn(" as THEMONTH_CEL,").append("USERID as USERID_CEL,");
 		
 		String cancelSql = userprofileTagService.genSelectSql("退订预测", "CANCEL_P1", "CANCEL_P2", null, "", "tagValue");
 		sqlBuilder.appendLn(cancelSql);

src/main/java/com/platomix/userprofile/handler/core/KuaishouTagHandler.java

@@ -43,7 +43,7 @@ public class KuaishouTagHandler extends AbstractTagHandler {
 		
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_KS", "USERID_KS"));
-		sqlBuilder.append("select DEVICE_NUMBER as USERID_KS,").append("'"+month+"'").appendLn(" as THEMONTH_KS,");
+		sqlBuilder.append("select ").append("'"+month+"'").appendLn(" as THEMONTH_KS,").append("DEVICE_NUMBER as USERID_KS,").append("'"+month+"'").appendLn(" as THEMONTH_KS,");
 		//标签范围值
 		/**
 		 * (case when IS_ORDER=2 then 'C100_02_001'

src/main/java/com/platomix/userprofile/handler/core/OrderSumDateTagHandler.java

@@ -43,7 +43,7 @@ public class OrderSumDateTagHandler extends AbstractTagHandler {
 		
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_OS", "USERID_OS"));
-		sqlBuilder.append("select USERID as USERID_OS,").append("'"+month+"'").appendLn(" as THEMONTH_OS,");
+		sqlBuilder.append("select").append("'"+month+"'").appendLn(" as THEMONTH_OS,").append("USERID as USERID_OS,");
 		
 		//标签范围值
 		/**

src/main/java/com/platomix/userprofile/handler/core/RmfTagHandler.java

@@ -43,7 +43,7 @@ public class RmfTagHandler extends AbstractTagHandler {
 		
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_RFM", "USERID_RFM"));
-		sqlBuilder.append("select USERID as USERID_RFM,").append("'"+month+"'").appendLn(" as THEMONTH_RFM,");
+		sqlBuilder.append("select").append("'"+month+"'").appendLn(" as THEMONTH_RFM,").append("USERID as USERID_RFM,");
 		//标签范围值
 		/**
 		 * (case when CLASS =1 then 'A112_03_001'

src/main/java/com/platomix/userprofile/handler/core/UserOrderStateTagHandler.java

@@ -5,6 +5,7 @@ package com.platomix.userprofile.handler.core;
 
 import java.time.LocalDate;
 import java.time.format.DateTimeFormatter;
+import java.util.concurrent.TimeUnit;
 
 import org.springframework.stereotype.Service;
 
@@ -22,41 +23,54 @@ import lombok.extern.slf4j.Slf4j;
 @Slf4j
 @Service
 public class UserOrderStateTagHandler extends AbstractTagHandler {
-	
+
 	final static String ZT_TABLE_NAME = "ads.ZT_USER_ORDER_STATE";
+	final static String Z_TEMP_TABLE_USERS = " ads.ztemp_relation_users";
 	final static long _STEP=50000000L;
-//	@Autowired
-//	private UserprofileTagService userprofileTagService;
+
+	/**
+	 * 开始前 删除用户缓存表
+	 * @param month
+	 */
+	@Override
+	public void beforeInvoke(String month) {
+		dynamicExecuteService.execute(DsEnum.clickhouse, String.format("drop table if exists %s",Z_TEMP_TABLE_USERS), "用户订购状态标签-缓存表");
+	}
 
 	@Override
 	public void invoke(String month, String mobileLeft, String mobileRight) {
 		// 查询所有标签
-//		List<UserprofileTag> tagList = userprofileTagService.list();
-//		Map<String, String> tagMap = tagList.stream().collect(Collectors.toMap(UserprofileTag::getTagId, UserprofileTag::getTagName, (t, t1) -> t));
-
-//		String lrMobileSql = mobileLeft == null || mobileRight == null ? "" : String.format(" and USERID>='%s' and USERID<'%s'", mobileLeft, mobileRight);
+		String lrDeviceNumberSql = mobileLeft == null || mobileRight == null ? "" : String.format(" Where DEVICE_NUMBER>='%s' and DEVICE_NUMBER<'%s'", mobileLeft, mobileRight);
+		String lrUserIdSql = mobileLeft == null || mobileRight == null ? "" : String.format("  USERID>='%s' and USERID<'%s'", mobileLeft, mobileRight);
 		
 		// 获取月份开始日期
 		String monthStart = month.substring(0, 4) + "-" + month.substring(4) + "-01";
 		LocalDate monthStartDate = LocalDate.parse(monthStart, DateTimeFormatter.ISO_LOCAL_DATE);
 		String nextMonth = monthStartDate.plusMonths(1).format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
-		
+
+		StrBuilder sqlBuilderTemp = new StrBuilder();
+		sqlBuilderTemp.appendLn(super.buildTableSql(Z_TEMP_TABLE_USERS,  "USERID"));
+		sqlBuilderTemp.append("select distinct USERID from ads.ETL_TD_ORDER_RELATIONS_YEAR where "+lrUserIdSql);
+		log.info("\n" + sqlBuilderTemp.toString());
+		dynamicExecuteService.execute(DsEnum.clickhouse, sqlBuilderTemp.toString(), "用户订购状态标签生成-缓存表准备");
+
+
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "THEMONTH_BL", "USERID_BL"));
-		sqlBuilder.append("select DEVICE_NUMBER as USERID_BL,").append("'"+month+"'").appendLn(" as THEMONTH_BL,")
+		sqlBuilder.append("select ").append("'"+month+"'").appendLn(" as THEMONTH_BL,").append("DEVICE_NUMBER as USERID_BL,")
 				  .appendLn("toDate('1970-01-01') as cTime,")
 				  .appendLn("'B112_19_003' as EVENT_USER_OSTATE,")
 				  .appendLn("'未订过' as EVENT_USER_OSTATE_C,")
 				  .appendLn("toDate('1970-01-01') as EVENT_USER_OSTATE_V from (")
 				  .appendLn(" select DISTINCT DEVICE_NUMBER from")
 				  .appendLn(" (")
-				  .appendLn("	select DEVICE_NUMBER from ads.XXH_ALL_MONTH")
+				  .appendLn("	select DEVICE_NUMBER from ads.XXH_ALL_MONTH "+lrDeviceNumberSql)
 				  .appendLn("	union all")
-				  .appendLn("   select DEVICE_NUMBER from ads.XXH_KUAISHOU_MONTH xkm")
+				  .appendLn("   select DEVICE_NUMBER from ads.XXH_KUAISHOU_MONTH "+lrDeviceNumberSql)
 				  .appendLn(" )")
 				  .appendLn(")s")
 				  .appendLn("where s.DEVICE_NUMBER not in (")
-				  .appendLn("select USERID from ads.ztemp_relation_users  --ads.ETL_TD_ORDER_RELATIONS")
+				  .appendLn("select USERID from ads.ztemp_relation_users  where "+lrUserIdSql)
 				  .appendLn(")")
 				  .appendLn("union all")
 				  .appendLn("SELECT '"+month+"' as THEMONTH_BL,USERID as USERID_BL,")
@@ -67,13 +81,19 @@ public class UserOrderStateTagHandler extends AbstractTagHandler {
 				  .appendLn("FROM (")
 				  .appendLn("	SELECT USERID,if(CANCELTIME='1970-01-01', toDate('"+nextMonth+"'),CANCELTIME) as CANCELTIME1")
 				  .appendLn("	FROM ads.ETL_TD_ORDER_RELATIONS_YEAR")
-				  .appendLn("	where THEMONTH ='"+month+"'")
+				  .appendLn("	where THEMONTH ='"+month+"' and "+lrUserIdSql)
 				  .appendLn("	ORDER BY CANCELTIME1 DESC")
 				  .appendLn(") GROUP BY USERID");
 		log.info("\n" + sqlBuilder.toString());
 		dynamicExecuteService.execute(DsEnum.clickhouse, sqlBuilder.toString(), "用户订购状态标签生成");
 	}
-	
+
+
+
+	@Override
+	public boolean executeInBatchs() {
+		return true;
+	}
 	@Override
 	public long calcMobileRangeStep() {
 		return _STEP;

src/main/java/com/platomix/userprofile/handler/core/XxhLongTagHandler.java

@@ -33,14 +33,14 @@ public class XxhLongTagHandler extends AbstractTagHandler {
 	public void invoke(String month, String mobileLeft, String mobileRight) {
 		String lrMobileSql = mobileLeft == null || mobileRight == null ? "" : String.format(" and DEVICE_NUMBER>='%s' and DEVICE_NUMBER<'%s'", mobileLeft, mobileRight);
 		StrBuilder sqlBuilder = new StrBuilder();
-		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, null, "DEVICE_NUMBER"));
+		sqlBuilder.appendLn(super.buildTableSql(XXH_TABLE_NAME, null, "DEVICE_NUMBER"));
 		sqlBuilder.appendLn("select DEVICE_NUMBER ,any(PROV_DESC) PROV_DESC,");
 		sqlBuilder.appendLn("any(SERVICE_TYPE) SERVICE_TYPE,any(CUST_SEX) CUST_SEX,any(CERT_AGE) CERT_AGE,any(PRODUCT_CLASS_DESC) PRODUCT_CLASS_DESC ");
 		sqlBuilder.appendLn("from ");
 		sqlBuilder.appendLn("(select  MONTH_ID, DEVICE_NUMBER ,if(s1.PROV_DESC=='',b.PROVINCE,s1.PROV_DESC) as PROV_DESC,SERVICE_TYPE ,CUST_SEX,CERT_AGE,PRODUCT_CLASS_DESC from ( ");
 		sqlBuilder.appendLn("	select MONTH_ID, DEVICE_NUMBER ,PROV_DESC,SERVICE_TYPE ,CUST_SEX,CERT_AGE,PRODUCT_CLASS_DESC from ");
 		sqlBuilder.appendLn("	ads.XXH_ALL_MONTH ");
-		sqlBuilder.appendLn(String.format("	where  MONTH_ID >='202101' and  DEVICE_NUMBER not in (select DEVICE_NUMBER from ads.XXH_KUAISHOU_MONTH where 1=1 '%s')  '%s' ",lrMobileSql,lrMobileSql));
+		sqlBuilder.appendLn(String.format("	where  MONTH_ID >='202101' and  DEVICE_NUMBER not in (select DEVICE_NUMBER from ads.XXH_KUAISHOU_MONTH where 1=1 %s)  %s ",lrMobileSql,lrMobileSql));
 		sqlBuilder.appendLn("  union all ");
 		sqlBuilder.appendLn(" select MONTH_ID,DEVICE_NUMBER,  PROV_DESC, NULL as SERVICE_TYPE,if(CUST_SEX==1,'男','女') as CUST_SEX, ");
 		sqlBuilder.appendLn("  multiIf((2021-CERT_AGE)>=2010,'10后',(2021-CERT_AGE)>=2000,'00后',(2021-CERT_AGE)>=1990,'90后',(2021-CERT_AGE)>=1980,'80后', ");
@@ -87,7 +87,7 @@ public class XxhLongTagHandler extends AbstractTagHandler {
 		}
 
 		StrBuilder sqlBuilder = new StrBuilder();
-		sqlBuilder.appendLn(super.buildTableSql(XXH_TABLE_NAME, null, "DEVICE_NUMBER_XXH_L"));
+		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, null, "DEVICE_NUMBER_XXH_L"));
 		sqlBuilder.appendLn("select DEVICE_NUMBER as DEVICE_NUMBER_XXH_L,");
 		//性别
 		String sexSql = userprofileTagService.genSelectSql("性别", "CUST_SEX", null, null);

src/main/java/com/platomix/userprofile/handler/core/XxhTagHandler.java

@@ -42,7 +42,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 		String lrMobileSql = mobileLeft == null || mobileRight == null ? "" : String.format(" and DEVICE_NUMBER>='%s' and DEVICE_NUMBER<'%s'", mobileLeft, mobileRight);
 		StrBuilder sqlBuilder = new StrBuilder();
 		sqlBuilder.appendLn(super.buildTableSql(ZT_TABLE_NAME, "MONTH_ID", "DEVICE_NUMBER"));
-		sqlBuilder.append("select DEVICE_NUMBER,").append("'"+month+"'").appendLn(" as MONTH_ID,");
+		sqlBuilder.append("select ").append("'"+month+"'").appendLn(" as MONTH_ID,").append("DEVICE_NUMBER,");
 		int year = LocalDate.now().getYear();
 		
 		//大网标签-入网时长
@@ -120,7 +120,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 	/**
 	 * 套餐价格
 	 * @param sqlBuilder
-	 * @param tagMap
+	 * @param tagListMap
 	 */
 	public void appendTagPackagePic(StrBuilder sqlBuilder, Map<String, List<UserprofileTag>> tagListMap) {
 		/**
@@ -140,7 +140,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 	/**
 	 * 大网标签-月流量消耗
 	 * @param sqlBuilder
-	 * @param tagMap
+	 * @param tagListMap
 	 */
 	public void appendEventFlux(StrBuilder sqlBuilder, Map<String, List<UserprofileTag>> tagListMap) {
 		/**
@@ -161,7 +161,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 	/**
 	 * 大网标签-月消费
 	 * @param sqlBuilder
-	 * @param tagMap
+	 * @param tagListMap
 	 */
 	public void appendEventConsume(StrBuilder sqlBuilder, Map<String, List<UserprofileTag>> tagListMap) {
 		/**
@@ -181,7 +181,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 	/**
 	 * 大网标签-视频流量消耗
 	 * @param sqlBuilder
-	 * @param tagMap
+	 * @param tagListMap
 	 */
 	public void appendTotalFluxVideo(StrBuilder sqlBuilder, Map<String, List<UserprofileTag>> tagListMap) {
 		/**
@@ -203,7 +203,7 @@ public class XxhTagHandler extends AbstractTagHandler{
 	/**
 	 * 大网标签-是否出账
 	 * @param sqlBuilder
-	 * @param tagMap
+	 * @param tagListMap
 	 */
 	public void appendEventIsAcct(StrBuilder sqlBuilder, Map<String, List<UserprofileTag>> tagListMap) {
 		/**

src/main/java/com/platomix/userprofile/utils/TagKit.java

@@ -55,4 +55,14 @@ public class TagKit {
 		}
 		return tplContent;
 	}
+
+	/**
+	 * 获取临时表名
+	 * @param tableName
+	 * @param month
+	 * @return
+	 */
+	public static String getTagTempTable(String tableName,String month){
+		return tableName+"_"+month;
+	}
 }

src/main/resources/application-dev.yml

@@ -51,7 +51,7 @@ spring:
 #                  initial-size: 0
             clickhouse:
                driver-class-name: com.clickhouse.jdbc.ClickHouseDriver
-               url: jdbc:clickhouse://10.0.17.152:8123/ads?socket_timeout=30000000&numPartitions=1&rewriteBatchedStatements=true
+               url: jdbc:clickhouse://10.0.17.152:8123/ads?socket_timeout=30000000&batchsize=50000&numPartitions=1&rewriteBatchedStatements=true
                username: default
                password: ltbiydhao66
 dflow:

src/main/resources/application-test.yml

@@ -39,16 +39,16 @@ spring:
 #                  initial-size: 0
             clickhouse:
                driver-class-name: ru.yandex.clickhouse.ClickHouseDriver
-               url: jdbc:clickhouse://10.0.17.152:8123/ads?socket_timeout=30000000&numPartitions=1&rewriteBatchedStatements=true
+               url: jdbc:clickhouse://10.0.17.152:8123/ads?socket_timeout=30000000&batchsize=50000&numPartitions=1&rewriteBatchedStatements=true
                username: default
                password: ltbiydhao66
 dflow:
-   use-debug: false
+   use-debug: true
    #如果设置为true 会执行删除 更新 插入操作！！，flase只会打印sql
    run-execute: true
-   flat-temp-table-name: Z_USER_TAG_FLAT_temp
-   flat-clu-table-name: Z_USER_TAG_FLAT_clu1
-   flat-all-view-name: ads.Z_USER_TAG_FLAT_ALL
+   flat-temp-table-name: Z_USER_TAG_FLAT_temp_dev
+   flat-clu-table-name: Z_USER_TAG_FLAT_clu1_dev
+   flat-all-view-name: ads.Z_USER_TAG_FLAT_ALL_dev
    
 dbcluster:
    #确保第一个是（源数据地址） 向（目标地址）分发

src/main/resources/application.yml

@@ -1,5 +1,5 @@
 server:
-  port: 8080
+  port: 8087
 
 spring:
    profiles: