项目初始化

qg-data-dc/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>cn.quantgroup</groupId>
+    <artifactId>qg-data-dc</artifactId>
+    <version>0.0.1-SNAPSHOT</version>
+    <name>qg-data-dc</name>
+    <packaging>jar</packaging>
+    <description>Demo project for Spring Boot</description>
+
+    <parent>
+        <groupId>cn.quantgroup</groupId>
+        <artifactId>qg-rt-dc</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
+    <properties>
+        <java.version>1.8</java.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>cn.quantgroup</groupId>
+            <artifactId>qg-data-service</artifactId>
+            <version>V1.0</version>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+
+</project>

qg-data-dc/src/main/java/cn/quantgroup/qgdatadc/QgDataDcApplication.java

+package cn.quantgroup.qgdatadc;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+import com.ctrip.framework.apollo.spring.annotation.EnableApolloConfig;
+import com.ctrip.framework.apollo.spring.config.ApolloPropertySourceInitializer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.domain.EntityScan;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
+import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.transaction.annotation.EnableTransactionManagement;
+
+//@EnableJpaRepositories(basePackages = {"cn.quantgroup.qgrtdc.repository.jpa"})
+//@EntityScan(basePackages = {"cn.quantgroup.qgrtdc.repository.jpa.entity"})
+@EnableTransactionManagement
+@EnableConfigurationProperties
+@EnableCaching
+@EnableAsync
+@EnableAspectJAutoProxy
+@Slf4j
+@EnableApolloConfig
+@SpringBootApplication
+public class QgDataDcApplication {
+
+    public static void main(String[] args) {
+        JSON.DEFAULT_GENERATE_FEATURE |= SerializerFeature.WriteEnumUsingToString.getMask();
+        SpringApplication springApplication = new SpringApplication(QgDataDcApplication.class);
+        springApplication.addInitializers(new ApolloPropertySourceInitializer());
+        springApplication.run(args);
+        log.info("QUANTGROUP-数据实时清洗-系统启动完成 ^_^ ");
+    }
+
+}

qg-data-dc/src/main/resources/META-INF/app.properties

+app.id=qg-data-dc
+namespace=application,tech.service.urls,tech.common,tech.sleuth,tech.deploy,tech.msg.sdk
\ No newline at end of file

qg-data-dc/src/main/resources/application.properties

+

qg-data-dc/src/test/java/cn/quantgroup/qgdatadc/QgDataDcApplicationTests.java

+package cn.quantgroup.qgdatadc;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class QgDataDcApplicationTests {
+
+    @Test
+    public void contextLoads() {
+    }
+
+}

qg-data-service/pom.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>cn.quantgroup</groupId>
+    <artifactId>qg-data-service</artifactId>
+    <version>0.1.9-SNAPSHOT</version>
+    <packaging>jar</packaging>
+    <name>qg-data-service</name>
+
+    <parent>
+        <groupId>cn.quantgroup</groupId>
+        <artifactId>commons-parent</artifactId>
+        <version>0.2.4</version>
+    </parent>
+
+    <properties>
+        <java.version>1.8</java.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.zaxxer</groupId>
+            <artifactId>HikariCP</artifactId>
+            <version>2.5.1</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.46</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.lkb.data</groupId>
+            <artifactId>lkb-data-service</artifactId>
+            <version>1.7.5.1-3b-SNAPSHOT</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.5</version>
+        </dependency>
+
+    </dependencies>
+
+    <!--<build>-->
+        <!--<plugins>-->
+            <!--<plugin>-->
+                <!--<groupId>org.springframework.boot</groupId>-->
+                <!--<artifactId>spring-boot-maven-plugin</artifactId>-->
+            <!--</plugin>-->
+        <!--</plugins>-->
+    <!--</build>-->
+
+    <build>
+        <plugins>
+            <plugin>
+                <artifactId>maven-assembly-plugin</artifactId>
+                <version>2.5.3</version>
+                <configuration>
+                    <descriptorRefs>
+                        <descriptorRef>jar-with-dependencies</descriptorRef>
+                    </descriptorRefs>
+                </configuration>
+                <executions>
+                    <execution>
+                        <id>make-assembly</id>
+                        <!-- this is used for inheritance merges -->
+                        <phase>package</phase>
+                        <!-- bind to the packaging phase -->
+                        <goals>
+                            <goal>single</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+            <!--<plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-source-plugin</artifactId>
+                <version>2.4</version>
+                <executions>
+                    <execution>
+                        <id>attach-sources</id>
+                        <phase>compile</phase>
+                        <goals>
+                            <goal>jar-no-fork</goal>
+                            &lt;!&ndash; 类似执行mvn source:jar &ndash;&gt;
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>-->
+        </plugins>
+    </build>
+
+    <distributionManagement>
+        <repository>
+            <id>Releases</id>
+            <name>LKB Releases Repository</name>
+            <url>http://repo.quantgroup.cn/nexus/content/repositories/lkb-releases</url>
+        </repository>
+        <snapshotRepository>
+            <id>Snapshots</id>
+            <name>LKB Snapshots Repository</name>
+            <url>http://repo.quantgroup.cn/nexus/content/repositories/lkb-snapshots</url>
+        </snapshotRepository>
+    </distributionManagement>
+
+</project>

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/QgDataServiceApplication.java

+package cn.quantgroup.qgdataservice;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.EnableAspectJAutoProxy;
+import org.springframework.scheduling.annotation.EnableAsync;
+
+@SpringBootApplication
+public class QgDataServiceApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(QgDataServiceApplication.class, args);
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/config/datasource/TidbDataSource.java

+package cn.quantgroup.qgdataservice.config.datasource;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+
+/**
+ * @Author fengjunkai
+ * @Date 2019-05-14 16:32
+ */
+public enum TidbDataSource {
+
+    HIK_DATA_SOURCE;
+
+    public javax.sql.DataSource dataSource;
+
+    TidbDataSource() {
+        if (dataSource == null) {
+            System.out.println("======创建TIDB数据库连接======");
+            HikariConfig config = new HikariConfig();
+//            config.setJdbcUrl("jdbc:mysql://10.17.115.6:4010/rc_comservice_data_pool_v2?useUnicode=true&characterEncoding=UTF8&rewriteBatchedStatements=true");
+//            config.setUsername("rc_comservice_data_pool_v2_w");
+//            config.setPassword("w9pr8IPJkLmUSBe4");
+            config.setJdbcUrl("jdbc:mysql://172.30.220.9:3306/rc_comservice_data_pool_v2?useUnicode=true&characterEncoding=UTF8&rewriteBatchedStatements=true");
+//            config.setJdbcUrl("jdbc:mysql://172.30.220.9:3306/rc_comservice_data_pool_v2?useUnicode=true&characterEncoding=UTF8");useServerPrepStmts=true
+            config.setUsername("qa");
+            config.setPassword("qatest");
+            config.setDriverClassName("com.mysql.jdbc.Driver");
+            config.setMaximumPoolSize(12);
+            config.setMinimumIdle(6);
+            config.addDataSourceProperty("cachePrepStmts", "true");
+            config.addDataSourceProperty("prepStmtCacheSize", "250");
+            config.addDataSourceProperty("prepStmtCacheSqlLimit", "2048");
+            dataSource = new HikariDataSource(config);
+        }
+
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/constant/Constant.java

+package cn.quantgroup.qgdataservice.constant;
+
+/**
+ * @Author fengjunkai
+ * @Date 2019-05-14 18:19
+ */
+public class Constant {
+
+    public static class SQL{
+
+        public static String USER_INFO_ITEM = "INSERT IGNORE INTO comservice_i_spider_user_info (uuid,realName,registerDate,idCard,phoneRemain,phone,addr,merry,cardType,cardNo,sex,loginName,userSource,timestamp) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?)";  //14
+
+        public static String MOBILE_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_mobile_call_info (uuid,cTime,tradeAddr,tradeWay,tradeType,receiverPhone,tradeTime,taocan,onlinePay,phone,iscm,timestamp) values (?,?,?,?,?,?,?,?,?,?,?,?)"; //12
+
+        public static String MOBILE_SMS_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_mobile_sms_info (uuid,allPay,createTs,phone,receiverPhone,sentAddr,sentTime,tradeway,timestamp) values (?,?,?,?,?,?,?,?,?)";  //9
+
+        public static String MOBILE_FLOW_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_mobile_flow_info (uuid,cTime,cheapService,communicationFees,onlineTime,onlineType,phone,totalFlow,tradeAddr,timestamp) values (?,?,?,?,?,?,?,?,?,?)";  //10
+
+        public static String TELECOM_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_telecom_call_info (uuid,tradeType,cTime,tradeTime,callWay,receiverPhone,tradeAddr,basePay,longPay,infoPay,otherPay,allPay,phone,iscm,timestamp) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"; //15
+
+        public static String TELECOM_SMS_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_telecom_sms_info (uuid,allpay,businessType,createTs,phone,receiverPhone,sentTime,timestamp) values (?,?,?,?,?,?,?,?)"; //8
+
+        public static String TELECOM_FLOW_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_telecom_flow_info (uuid,beginTime,business,fee,flow,iscm,location,netType,phone,tradeTime,timestamp) values (?,?,?,?,?,?,?,?,?,?,?)"; //11
+
+        public static String UNICOM_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_unicom_call_info (uuid,businessType,cTime,tradeTime,callType,receiverPhone,tradeAddr,tradeType,basePay,ldPay,otherPay,totalPay,phone,iscm,reductionPay,timestamp) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";  //16
+
+        public static String UNICOM_SMS_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_unicom_sms_info (uuid,allPay,createTs,phone,receiverPhone,sentTime,tradeType,timestamp) values (?,?,?,?,?,?,?,?)"; //8
+
+        public static String UNICOM_FLOW_DETAIL_INFOS = "INSERT IGNORE INTO comservice_i_unicom_flow_info (uuid,allFlow,allPay,createTs,phone,startTime,tradeAddr,tradeType,timestamp) values (?,?,?,?,?,?,?,?,?)";//9
+
+        public static String PHONE_BILL_INFOS = "INSERT IGNORE INTO comservice_i_phone_bill_info (uuid,phoneNo,billDay,name,amount,dependCycle,billPackage,content,timestamp) values (?,?,?,?,?,?,?,?,?)";//9
+
+        public static String PHONE_RECHARGE_INFOS = "INSERT IGNORE INTO comservice_i_phone_payment_info (uuid,phone,payamount,paytime,paymethod,paychannel,payspare,timestamp) values (?,?,?,?,?,?,?,?)";//8
+
+    }
+
+
+    public static class OPERATOR{
+
+        public static final String MOBILE_USERSOURCE_CALL = "MOBILE_USERSOURCE_CALL";
+
+        public static final String TELECOM_USERSOURCE_CALL = "TELECOM_USERSOURCE_CALL";
+
+        public static final String UNICOM_USERSOURCE_CALL = "UNICOM_USERSOURCE_CALL";
+
+        public static final String MOBILE_USERSOURCE_SMS = "MOBILE_USERSOURCE_SMS";
+
+        public static final String TELECOM_USERSOURCE_SMS = "TELECOM_USERSOURCE_SMS";
+
+        public static final String UNICOM_USERSOURCE_SMS = "UNICOM_USERSOURCE_SMS";
+
+        public static final String MOBILE_USERSOURCE_FLOW = "MOBILE_USERSOURCE_FLOW";
+
+        public static final String TELECOM_USERSOURCE_FLOW = "TELECOM_USERSOURCE_FLOW";
+
+        public static final String UNICOM_USERSOURCE_FLOW = "UNICOM_USERSOURCE_FLOW";
+
+        public static final String PHONE_CALL_BILL = "PHONE_CALL_BILL";
+
+//        public static final String PHONE_FLOW_BILL = "PHONE_FLOW_BILL";
+
+        public static final String PHONE_RECHARGE_RECORDS = "PHONE_RECHARGE_RECORDS";
+
+
+    }
+
+    public static class BATCH{
+        public static final int BATCH_SIZE = 1000;
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/service/hbase/OperatorInfoService.java

+package cn.quantgroup.qgdataservice.service.hbase;
+
+import cn.quantgroup.qgdataservice.service.tidb.OperatorInfoCleaningService;
+import com.lkb.data.hbase.dataservice.SpiderUserItemDataService;
+import com.lkb.data.hbase.dataservice.operators.PhoneBillDataService;
+import com.lkb.data.hbase.dataservice.operators.PhonePaymentDataService;
+import com.lkb.data.hbase.dataservice.operators.mobile.MobileCallInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.mobile.MobileFlowInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.mobile.MobileSmsInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.telecom.TelecomCallInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.telecom.TelecomFlowInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.telecom.TelecomSmsInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.unicom.UnicomCallInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.unicom.UnicomFlowInfoDataService;
+import com.lkb.data.hbase.dataservice.operators.unicom.UnicomSmsInfoDataService;
+import com.lkb.data.hbase.row.SpiderUserItemDataRow;
+import com.lkb.data.hbase.row.operators.PhoneBillDataRow;
+import com.lkb.data.hbase.row.operators.PhonePaymentDataRow;
+import com.lkb.data.hbase.row.operators.mobile.MobileCallInfoRow;
+import com.lkb.data.hbase.row.operators.mobile.MobileFlowInfoRow;
+import com.lkb.data.hbase.row.operators.mobile.MobileSmsInfoRow;
+import com.lkb.data.hbase.row.operators.telecom.TelecomCallInfoRow;
+import com.lkb.data.hbase.row.operators.telecom.TelecomFlowInfoRow;
+import com.lkb.data.hbase.row.operators.telecom.TelecomSmsInfoRow;
+import com.lkb.data.hbase.row.operators.unicom.UnicomCallInfoRow;
+import com.lkb.data.hbase.row.operators.unicom.UnicomFlowInfoRow;
+import com.lkb.data.hbase.row.operators.unicom.UnicomSmsInfoRow;
+
+import java.util.List;
+
+/**
+ * 三大运营商数据清洗
+ *
+ * @Author fengjunkai
+ * @Date 2019-05-14 11:42
+ */
+public class OperatorInfoService {
+
+    /**
+     * 保存并清洗用户详情信息
+     *
+     * @param spiderUserItemDataRow
+     * @param uuid
+     */
+    public static void storageUserInfoItem(SpiderUserItemDataRow spiderUserItemDataRow, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            SpiderUserItemDataService.putAndCheck(spiderUserItemDataRow);
+        else
+            SpiderUserItemDataService.put(spiderUserItemDataRow);
+
+        OperatorInfoCleaningService.cleaningAndSaveUserInfo(spiderUserItemDataRow, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗移动通话详单
+     *
+     * @param mobileCallInfoRows
+     * @param uuid
+     */
+    public static void storageMobileCallDetailInfos(List<MobileCallInfoRow> mobileCallInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            MobileCallInfoDataService.putAndCheck(mobileCallInfoRows);
+        else
+            MobileCallInfoDataService.put(mobileCallInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveMobileCallDetailInfo(mobileCallInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗移动短信详单
+     *
+     * @param mobileSmsInfoRows
+     * @param uuid
+     */
+    public static void storageMobileSmsDetail(List<MobileSmsInfoRow> mobileSmsInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            MobileSmsInfoDataService.putAndCheck(mobileSmsInfoRows);
+        else
+            MobileSmsInfoDataService.put(mobileSmsInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveMobileSmsDetainInfo(mobileSmsInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗移动流量详单
+     *
+     * @param mobileFlowInfoRows
+     * @param uuid
+     */
+    public static void storageMobileFlowDetailInfos(List<MobileFlowInfoRow> mobileFlowInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            MobileFlowInfoDataService.putAndCheck(mobileFlowInfoRows);
+        else
+            MobileFlowInfoDataService.put(mobileFlowInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveMobileFlowDetailInfo(mobileFlowInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗电信通话详单
+     *
+     * @param telecomCallInfoRows
+     * @param uuid
+     */
+    public static void storageTelecomCallDetailInfos(List<TelecomCallInfoRow> telecomCallInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            TelecomCallInfoDataService.putAndCheck(telecomCallInfoRows);
+        else
+            TelecomCallInfoDataService.put(telecomCallInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveTelecomCallDetailInfo(telecomCallInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗电信短信详单
+     *
+     * @param telecomSmsInfoRows
+     * @param uuid
+     */
+    public static void storageTelecomSmsDetailInfos(List<TelecomSmsInfoRow> telecomSmsInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            TelecomSmsInfoDataService.putAndCheck(telecomSmsInfoRows);
+        else
+            TelecomSmsInfoDataService.put(telecomSmsInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveTelecomSmsDetailInfo(telecomSmsInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗电信流量详单
+     *
+     * @param telecomFlowInfoRows
+     * @param uuid
+     */
+    public static void storageTelecomFlowDetailInfos(List<TelecomFlowInfoRow> telecomFlowInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            TelecomFlowInfoDataService.putAndCheck(telecomFlowInfoRows);
+        else
+            TelecomFlowInfoDataService.put(telecomFlowInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveTelecomFlowDetailInfo(telecomFlowInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗联通通话详单
+     *
+     * @param unicomCallInfoRows
+     * @param uuid
+     */
+    public static void storageUnicomCallDetailInfos(List<UnicomCallInfoRow> unicomCallInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            UnicomCallInfoDataService.putAndCheck(unicomCallInfoRows);
+        else
+            UnicomCallInfoDataService.put(unicomCallInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveUnicomCallDetailInfo(unicomCallInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗联通短信详单
+     *
+     * @param unicomSmsInfoRows
+     * @param uuid
+     */
+    public static void storageUnicomSmsDetailInfos(List<UnicomSmsInfoRow> unicomSmsInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            UnicomSmsInfoDataService.putAndCheck(unicomSmsInfoRows);
+        else
+            UnicomSmsInfoDataService.put(unicomSmsInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveUnicomSmsDetailInfo(unicomSmsInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗联通流量详单
+     *
+     * @param unicomFlowInfoRows
+     * @param uuid
+     */
+    public static void storageUnicomFlowDetailInfos(List<UnicomFlowInfoRow> unicomFlowInfoRows, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            UnicomFlowInfoDataService.putAndCheck(unicomFlowInfoRows);
+        else
+            UnicomFlowInfoDataService.put(unicomFlowInfoRows);
+
+        OperatorInfoCleaningService.cleaningAndSaveUnicomFlowDetailInfo(unicomFlowInfoRows, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗通话账单
+     *
+     * @param phoneBillDataRow
+     * @param uuid
+     */
+    public static void storageBillInfos(PhoneBillDataRow phoneBillDataRow, String uuid, boolean putAndCheck, boolean ka) {
+
+        if (putAndCheck)
+            PhoneBillDataService.putAndCheck(phoneBillDataRow);
+        else
+            PhoneBillDataService.put(phoneBillDataRow);
+
+        OperatorInfoCleaningService.cleaningAndSavePhoneBillInfo(phoneBillDataRow, uuid, ka);
+
+    }
+
+    /**
+     * 保存并清洗手机充值记录
+     *
+     * @param phonePaymentDataRow
+     * @param uuid
+     */
+    public static void storageRechargeInfos(PhonePaymentDataRow phonePaymentDataRow, String uuid, boolean putAndCheck, boolean ka) {
+
+        PhonePaymentDataService.put(phonePaymentDataRow);
+
+        OperatorInfoCleaningService.cleaningAndSaveRechargeInfo(phonePaymentDataRow, uuid, ka);
+
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/utils/JdbcExecuters.java

+package cn.quantgroup.qgdataservice.utils;
+
+import cn.quantgroup.qgdataservice.constant.Constant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.*;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static cn.quantgroup.qgdataservice.config.datasource.TidbDataSource.HIK_DATA_SOURCE;
+
+/**
+ * Created by renfeng on 2019/4/22.
+ */
+public class JdbcExecuters {
+
+    private static final Logger log = LoggerFactory.getLogger(JdbcExecuters.class);
+
+    /**
+     * 批量插入
+     *
+     * @param sql
+     */
+    public static void batchUpdateExecute(String sql, String userSource, String phone) {
+        Connection conn = null;
+        Statement st = null;
+        try {
+            conn = HIK_DATA_SOURCE.dataSource.getConnection();
+            st = conn.createStatement();
+            try {
+                st.executeUpdate(sql);
+            } catch (SQLException e) {
+                log.error("执行sql异常, sql: {} ", sql, e);
+            } catch (Exception e) {
+                log.error("未知异常, sql: {} ", sql, e);
+            }
+        } catch (Exception e) {
+            log.error("执行sqlException异常, sqls: {} ", sql, e);
+        } finally {
+            close(conn, st, null);
+        }
+    }
+
+
+    /**
+     * 批量插入
+     *
+     * @param sqls
+     */
+    public static void batchUpdateExecute(List<String> sqls) {
+        Connection conn = null;
+        Statement st = null;
+        try {
+            conn = HIK_DATA_SOURCE.dataSource.getConnection();
+            st = conn.createStatement();
+            conn.setAutoCommit(false);
+            for (String sql : sqls) {
+                try {
+                    st.executeUpdate(sql);
+                } catch (SQLException e) {
+                    log.error("执行sql异常, sql: {} ", sql, e);
+                } catch (Exception e) {
+                    log.error("未知异常, sql: {} ", sql, e);
+                }
+            }
+            conn.commit();
+        } catch (Exception e) {
+            log.error("执行sqlException异常, sqls: {} ", sqls, e);
+        } finally {
+            close(conn, st, null);
+        }
+    }
+
+    public static int batchExecute(List<List<String>> list, String sql) {
+        Connection conn = null;
+        PreparedStatement ps = null;
+        AtomicInteger atomicInteger = new AtomicInteger();
+        try {
+
+            conn = HIK_DATA_SOURCE.dataSource.getConnection();
+            ps = conn.prepareStatement(sql);
+            conn.setAutoCommit(false);
+
+            for (int i = 0; i < list.size(); i++) {
+
+                List<String> params = list.get(i);
+
+                for (int j = 0; j < params.size(); j++) {
+                    ps.setString(j + 1, params.get(j));
+                }
+
+                ps.addBatch();
+                atomicInteger.getAndIncrement();
+
+                if (i > 0 && i % Constant.BATCH.BATCH_SIZE == 0) {
+                    ps.executeBatch();
+                    conn.commit();
+                    ps.clearBatch();
+                }
+            }
+
+            ps.executeBatch();
+            conn.commit();
+
+        } catch (Exception e) {
+            log.error("清洗数据批量插入数据异常", e);
+        } finally {
+            close(conn, ps, null);
+        }
+        return atomicInteger.get();
+    }
+
+//    public static void batchExecute(List<String> list) {
+//        Connection conn = null;
+//        PreparedStatement ps = null;
+//        try {
+//
+////            String sql = "INSERT IGNORE INTO comservice_i_unicom_call_info (uuid,businessType,cTime,tradeTime,callType,receiverPhone,tradeAddr,tradeType,basePay,ldPay,otherPay,totalPay,phone,iscm,reductionPay,timestamp) " +
+////                    "values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)";
+//
+//            conn = HIK_DATA_SOURCE.dataSource.getConnection();
+//            ps = conn.prepareStatement("");
+//            conn.setAutoCommit(false);
+//            for (int i = 0; i < list.size(); i++) {
+//                ps.addBatch(list.get(i));
+//                if (i>0 && i % Constant.BATCH.BATCH_SIZE == 0) {
+//                    ps.executeBatch();
+//                    ps.clearBatch();
+//                }
+//            }
+//            ps.executeBatch();
+//            conn.commit();
+//            ps.clearBatch();
+//
+//
+//        } catch (Exception e) {
+//            log.error("清洗数据批量插入数据异常", e);
+//        } finally {
+//            close(conn, ps, null);
+//        }
+//    }
+
+    /**
+     * 关闭资源
+     *
+     * @param conn
+     * @param st
+     * @param rs
+     */
+    private static void close(Connection conn, Statement st, ResultSet rs) {
+        try {
+            if (rs != null)
+                rs.close();
+            if (st != null)
+                st.close();
+            if (conn != null)
+                conn.close();
+        } catch (SQLException e) {
+            e.printStackTrace();
+        }
+
+    }
+
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/utils/NumberUtils.java

+package cn.quantgroup.qgdataservice.utils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.math.BigDecimal;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @Author fengjunkai
+ * @Date 2019-05-14 17:51
+ */
+public class NumberUtils {
+
+    private static final Logger log = LoggerFactory.getLogger(NumberUtils.class);
+
+    public static String formatPhoneRemain(Float remain, String userSource, String phone) {
+
+        if (Objects.isNull(remain)) return null;
+
+        try {
+
+            BigDecimal bigDecimal = new BigDecimal(remain);
+            BigDecimal phoneRemain = bigDecimal.setScale(2, BigDecimal.ROUND_HALF_UP);
+            return String.valueOf(phoneRemain);
+
+        } catch (Exception e) {
+            log.error("运营商 ( {} ) 解析用户余额异常, phone: {} , enterStr: {} ", userSource, phone, remain, e);
+            return null;
+        }
+
+    }
+
+    public static String getUseTimeFormat(String time, String userSource, String phone) {
+        if (StringUtils.isBlank(time)) {
+            return null;
+        }
+
+        if (StringUtils.startsWith(time, "-")) {
+            return null;
+        }
+
+        try {
+            long allTime = 0, hour2Second = 0, min2Second = 0, second = 0;
+            if (time.contains(":")) {
+                if (time.length() == 7 && time.indexOf(":") == 1) {
+                    time = "0" + time;
+                }
+                if (time.length() == 8) {
+                    second = new Long(time.substring(6, 8));
+                    min2Second = new Long(time.substring(3, 5));
+                    hour2Second = new Long(time.substring(0, 2));
+                } else if (time.split(":").length == 3) {
+                    String[] timeStrs = time.split(":");
+                    second = new Long(timeStrs[2]);
+                    min2Second = new Long(timeStrs[1]);
+                    hour2Second = new Long(timeStrs[0]);
+                }
+                allTime = hour2Second * 3600 + min2Second * 60 + second;
+                return String.valueOf(allTime);
+            } else {
+                if (time.contains("时")) {
+                    if (time.contains("小时"))
+                        hour2Second = Long.parseLong(time.split("小时")[0].trim()) * 60 * 60;
+                    else
+                        hour2Second = Long.parseLong(time.split("时")[0].trim()) * 60 * 60;
+                }
+                if (time.contains("分")) {
+                    if (time.contains("时")) {
+                        min2Second = Long.parseLong(subStr("时", "分", time)) * 60;
+                    } else {
+                        String strs[] = time.split("分");
+                        min2Second = Long.parseLong(strs[0].trim()) * 60;
+                        try {
+                            if (strs.length > 1 && !strs[1].contains("秒") && StringUtils.isNumeric(strs[1])) {
+                                second = Long.parseLong(strs[1].trim());
+                            }
+                        } catch (Exception e) {
+                            e.printStackTrace();
+                        }
+                    }
+                }
+                if (time.contains("秒")) {
+                    if (time.contains("分")) {
+                        if (time.contains("分钟"))
+                            second = Long.parseLong(subStr("分钟", "秒", time).trim());
+                        else
+                            second = Long.parseLong(subStr("分", "秒", time).trim());
+                    } else if (time.contains("时")) {
+                        if (time.contains("小时"))
+                            second = Long.parseLong(subStr("小时", "秒", time).trim());
+                        else
+                            second = Long.parseLong(subStr("时", "秒", time).trim());
+                    } else {
+                        second = Long.parseLong(time.split("秒")[0].trim());
+                    }
+                }
+                allTime = hour2Second + min2Second + second;
+                if (allTime == 0) {
+                    if (time != null) {
+                        try {
+                            if (time.contains(".")) {
+                                time = time.substring(0, time.indexOf("."));
+                            }
+                            allTime = Long.parseLong(time.trim().replaceAll("\\D+?", ""));
+                        } catch (NumberFormatException e) {
+                            e.printStackTrace();
+                        }
+                    }
+
+                }
+                return String.valueOf(allTime);
+            }
+        } catch (Exception e) {
+            log.error("运营商 ( {} ) 解析用户通话使用时间异常, phone: {} , enterStr: {} ", userSource, phone, time, e);
+        }
+        return time;
+    }
+
+    public static String flowFormat(String flow_temp, String userSource, String phone) {
+        try {
+
+            if (StringUtils.isBlank(flow_temp) || StringUtils.startsWith(flow_temp, "-")) {
+                return null;
+            }
+
+            BigDecimal mb2Kb = new BigDecimal("0");
+            BigDecimal gb2Kb = new BigDecimal("0");
+            BigDecimal kb = new BigDecimal("0");
+            BigDecimal thou = new BigDecimal("1024");
+            flow_temp = flow_temp.replaceAll(" ", "").replace(" ", "").replace("(", "").replace(")", "").replace("（", "").replace("）", "");
+            // 全角空格这么霸道...by Pat。Liu
+            flow_temp = flow_temp.replaceAll("　", "");
+            flow_temp = flow_temp.replaceAll("-", "");
+            if (flow_temp.contains("GB") || flow_temp.contains("MB") || flow_temp.contains("KB")) {
+                if (flow_temp.contains("GB")) {
+                    gb2Kb = new BigDecimal(flow_temp.split("GB")[0].trim()).multiply(thou).multiply(thou);
+                }
+                if (flow_temp.contains("MB")) {
+                    if (flow_temp.contains("GB"))
+                        mb2Kb = new BigDecimal(flow_temp.split("GB")[1].split("MB")[0].trim()).multiply(thou);
+                    else
+                        mb2Kb = new BigDecimal(flow_temp.split("MB")[0].trim()).multiply(thou);
+                }
+                if (flow_temp.contains("KB")) {
+                    if (flow_temp.contains("MB")) {
+                        kb = new BigDecimal(subStr("MB", "KB", flow_temp).trim());
+                    } else if (flow_temp.contains("GB")) {
+                        kb = new BigDecimal(subStr("GB", "KB", flow_temp).trim());
+                    } else {
+                        kb = new BigDecimal(flow_temp.split("KB")[0].trim());
+                    }
+                }
+            } else if (isNumeric(flow_temp)) {
+                BigDecimal bigDecimal = new BigDecimal(flow_temp).setScale(2, BigDecimal.ROUND_HALF_UP);
+                return String.valueOf(bigDecimal);
+            } else {
+                if (flow_temp.contains("G")) {
+                    gb2Kb = new BigDecimal(flow_temp.split("G")[0].trim()).multiply(thou).multiply(thou);
+                }
+                if (flow_temp.contains("M")) {
+                    if (flow_temp.contains("G"))
+                        mb2Kb = new BigDecimal(flow_temp.split("G")[1].split("M")[0].trim()).multiply(thou);
+                    else
+                        mb2Kb = new BigDecimal(flow_temp.split("M")[0].trim()).multiply(thou);
+                }
+                if (flow_temp.contains("K")) {
+                    if (flow_temp.contains("M")) {
+                        kb = new BigDecimal(subStr("M", "K", flow_temp).trim());
+                    } else if (flow_temp.contains("G")) {
+                        kb = new BigDecimal(subStr("G", "K", flow_temp).trim());
+                    } else {
+                        kb = new BigDecimal(flow_temp.split("K")[0].trim());
+                    }
+                }
+            }
+
+            BigDecimal d = gb2Kb.add(mb2Kb).add(kb).setScale(2, BigDecimal.ROUND_HALF_UP);
+            return String.valueOf(d);
+        } catch (Exception e) {
+            log.error("运营商 ( {} ) 解析用户使用总流量异常, phone: {} , enterStr: {} ", userSource, phone, flow_temp, e);
+            return null;
+        }
+    }
+
+    public static String amountToFormat(String amount, String userSource, String phone) {
+        try {
+            String amountTmep = amount;
+            if (isNumeric(amountTmep)) {
+                if (amountTmep.startsWith("-") || amount.contains("E") || amount.contains("e")) {
+                    return new BigDecimal(0.00).setScale(2, BigDecimal.ROUND_HALF_UP).toString();
+                }
+                return new BigDecimal(amountTmep).setScale(2, BigDecimal.ROUND_HALF_UP).toString();
+            }
+        } catch (Exception e) {
+            log.error("运营商 ( {} ) 解析用户账单当月费用异常, phone: {} , enterStr: {} ", userSource, phone, amount, e);
+        }
+
+        return amount;
+    }
+
+    public static String subStr(String stext, String etext, String text) {
+        int sindex = text.indexOf(stext);
+        if (sindex >= 0) {
+            int eindex = text.indexOf(etext, sindex);
+            if (eindex >= 0) {
+                String ctext = text.substring(sindex + stext.length(), eindex);
+                return ctext;
+            }
+        }
+        return "";
+    }
+
+    public static boolean isNumeric(String str) {
+        Pattern pattern = Pattern.compile("-?[0-9]+(.[0-9]+)?");
+        Matcher isNum = pattern.matcher(str);
+        if (!isNum.matches()) {
+            return false;
+        }
+        return true;
+    }
+
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/utils/PhoneUtils.java

+package cn.quantgroup.qgdataservice.utils;
+
+
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.regex.Pattern;
+
+/**
+ * @Author fengjunkai
+ * @Date 2019-05-14 18:08
+ */
+public class PhoneUtils {
+
+    private static final Logger log = LoggerFactory.getLogger(PhoneUtils.class);
+
+    public static String getPhoneNoByRegx(String enterStr, String userSource, String phone) {
+
+        if (StringUtils.isBlank(enterStr)) {
+            return null;
+        }
+
+        try {
+            String phoneNo = Pattern.compile("[^0-9]").matcher(enterStr).replaceAll("");
+
+            if (StringUtils.startsWith(phoneNo, "86125832")) {
+                return phoneNo.replaceFirst("86125832", "");
+            } else if (StringUtils.startsWith(phoneNo, "0086125832")) {
+                return StringUtils.replaceFirst(phoneNo, "0086125832", "");
+            } else if (13 == phoneNo.length() && StringUtils.startsWith(phoneNo, "86")) {
+                return phoneNo.replaceFirst("86", "");
+            } else if (13 == phoneNo.length() && StringUtils.startsWith(phoneNo, "001")) {
+                return phoneNo.replaceFirst("00", "");
+            } else if (StringUtils.startsWith(phoneNo, "+86")) {
+                return phoneNo.replaceFirst("\\+86", "");
+            } else if (StringUtils.startsWith(phoneNo, "0086")) {
+                return phoneNo.replaceFirst("0086", "");
+            }
+
+        } catch (Exception e) {
+            log.error("运营商 ( {} ) 解析用户余额异常, phone: {} , enterStr: {} ", userSource, phone, enterStr, e);
+        }
+
+        return enterStr;
+
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/utils/SqlUtils.java

+package cn.quantgroup.qgdataservice.utils;
+
+import java.util.List;
+
+/**
+ * @Author fengjunkai
+ * @Date 2019-05-14 18:18
+ */
+public class SqlUtils {
+
+    public static String getExecuteSql(List<String> list, String sqlStr) {
+        String formatStr = "";
+        for (int i = 0; i < list.size(); i++) {
+            String string = list.get(i);
+            if (string == null) {
+                formatStr += (string + (i == list.size() - 1 ? "" : ","));
+            } else {
+                formatStr += ("'" + string + "'" + (i == list.size() - 1 ? "" : ","));
+            }
+        }
+        String executeSql = String.format(sqlStr, formatStr);
+        return executeSql;
+    }
+
+}

qg-data-service/src/main/java/cn/quantgroup/qgdataservice/utils/ThreadPoolExecutorUtils.java

+
+package cn.quantgroup.qgdataservice.utils;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+public class ThreadPoolExecutorUtils {
+
+    private static final Logger log = LoggerFactory.getLogger(ThreadPoolExecutorUtils.class);
+
+    private static ThreadPoolExecutorUtils threadPool = new ThreadPoolExecutorUtils();
+
+    private ThreadPoolExecutor executor = null;
+
+    private ThreadPoolExecutorUtils() {
+        executor = new ThreadPoolExecutor(50, 200, 50, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(20));
+    }
+
+    public static ThreadPoolExecutorUtils getThreadPoolUtil() {
+        return threadPool;
+    }
+
+    public void execute(Runnable r) {
+        try {
+            int activeCount = executor.getActiveCount();
+            int size = executor.getQueue().size();
+            log.info("数据清洗线程池活跃数: {} , queueSize: {} ", activeCount, size);
+        } catch (Exception e) {
+            log.info("TASK_POOL KA Exception : {}", e);
+        }
+        executor.execute(r);
+    }
+
+    public void shutDown() {
+        executor.shutdown();
+    }
+
+}

qg-data-service/src/main/resources/application.properties

+

qg-data-service/src/test/java/cn/quantgroup/qgdataservice/QgDataServiceApplicationTests.java

+package cn.quantgroup.qgdataservice;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class QgDataServiceApplicationTests {
+
+    @Test
+    public void contextLoads() {
+    }
+
+}