ソースを参照

头条数据清洗

yumeng 4 年 前
コミット
3fc7528d0a

+ 19 - 9
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedanceEtlVideoJob.java

@@ -39,9 +39,9 @@ public class BytedanceEtlVideoJob {
 
     @XxlJob("bytedanceEtlVideoJob")
     public void execute() throws Exception {
-        Date startDate = DateUtils.addDay(new Date(),-2);
-        for(int i=0;i<=2;i++){
-            Date getDate = DateUtils.addDay(startDate,i);
+        Date startDate = DateUtils.addDay(new Date(), -2);
+        for (int i = 0; i <= 2; i++) {
+            Date getDate = DateUtils.addDay(startDate, i);
             bytedanceVideoEtlInfoService.etlBytedanceVideoInfo(getDate);
         }
     }
@@ -49,9 +49,9 @@ public class BytedanceEtlVideoJob {
     //TODO 待废弃
     @XxlJob("bytedanceEtlVideoProjectJob")
     public void bytedanceEtlVideoProjectJob() throws Exception {
-        Date startDate = DateUtils.addDay(new Date(),-2);
-        for(int i=0;i<=2;i++){
-            Date getDate = DateUtils.addDay(startDate,i);
+        Date startDate = DateUtils.addDay(new Date(), -2);
+        for (int i = 0; i <= 2; i++) {
+            Date getDate = DateUtils.addDay(startDate, i);
             bytedanceVideoEtlInfoService.etlBytedanceVideoInfoByProjectId(getDate);
         }
     }
@@ -59,6 +59,7 @@ public class BytedanceEtlVideoJob {
 
     /**
      * 头条视频报表url异常 -修改
+     *
      * @throws Exception
      */
     @XxlJob("updBytedanceVideoJob")
@@ -67,7 +68,7 @@ public class BytedanceEtlVideoJob {
         String timeStr = sdf.format(new Date());
         List<UserAllocation> userAllocationList = userAllocationService.list();
         for (UserAllocation userAllocation : userAllocationList) {
-            etlBytedanceReportVideoDailyService.updBytedanceVideo(userAllocation.getAccountId(),userAllocation.getProjectId(),timeStr);
+            etlBytedanceReportVideoDailyService.updBytedanceVideo(userAllocation.getAccountId(), userAllocation.getProjectId(), timeStr);
         }
     }
 
@@ -76,6 +77,7 @@ public class BytedanceEtlVideoJob {
 
     /**
      * 清洗头条剪辑
+     *
      * @throws Exception
      */
     @XxlJob("etlBytedanceVideoClipInfoJob")
@@ -87,6 +89,7 @@ public class BytedanceEtlVideoJob {
 
     /**
      * 清洗头条视频报表编导数据
+     *
      * @throws Exception
      */
     @XxlJob("etlBytedanceVideoPlanInfoJob")
@@ -96,9 +99,9 @@ public class BytedanceEtlVideoJob {
     }
 
 
-
     /**
      * 清洗头条拍摄
+     *
      * @throws Exception
      */
     @XxlJob("etlBytedanceVideoShotInfoJob")
@@ -108,14 +111,21 @@ public class BytedanceEtlVideoJob {
 
     /**
      * 头条视频报表数据任务
+     *
      * @throws Exception
      */
     @XxlJob("formatBytedance")
     public void etlBytedanceVideoJob() throws Exception {
+
+        String nowDate = DateUtils.getNowDate("yyyy-MM-dd");
+        String endDate = DateUtils.getAnotherDay("yyyy-MM-dd", nowDate, -1);
+        String startDate = DateUtils.getAnotherDay("yyyy-MM-dd", nowDate, -2);
         //查询所有账户-头条
         List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
         tokens.forEach(token -> executorService.submit(() -> {
-            etlBytedanceReportVideoDailyService.etlBytedanceVideo(token.getAccountId());
+            etlBytedanceReportVideoDailyService.etlBytedanceVideo(token.getAccountId(), startDate);
+            etlBytedanceReportVideoDailyService.etlBytedanceVideo(token.getAccountId(), endDate);
         }));
     }
+
 }

+ 1 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/batch/service/impl/KuaishouInterfaceServiceImpl.java

@@ -848,6 +848,7 @@ public class KuaishouInterfaceServiceImpl implements IKuaishouInterfaceService {
     }
 
 
+
     /**
      * 创建广告计划
      *

+ 285 - 64
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/report/mapper/xml/EtlBytedanceReportVideoDailyMapper.xml

@@ -2,7 +2,8 @@
 <!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
 <mapper namespace="cn.com.ctop.toutiao.modules.report.mapper.EtlBytedanceReportVideoDailyMapper">
 
-    <select id="getCtopMaterialAscriptionInfo" resultType="cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo">
+    <select id="getCtopMaterialAscriptionInfo"
+            resultType="cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo">
         SELECT DISTINCT
             md5,
             cma.clip_id,
@@ -23,7 +24,8 @@
 		cma.plan_name is not null;
     </select>
 
-    <update id="updateOneBytedanceVideoInfo"  parameterType="cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo">
+    <update id="updateOneBytedanceVideoInfo"
+            parameterType="cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo">
         UPDATE etl_report_bytedance_video
         <trim prefix="set" suffixOverrides=",">
             <if test="etlReportBytedanceVideo.clipId!=null">clip_id=#{etlReportBytedanceVideo.clipId},</if>
@@ -61,67 +63,287 @@
     </select>
     <select id="etlBytedanceVideoInfo"
             resultType="cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo">
-             select
-                      (case when cmi.url is not null then cmi.url else '' end) 'url',
-                          (CASE when cmi.cover_url is null then ''
-                                ELSE cmi.cover_url end) 'cover_url',
-                          (case when cmi.material_name is null then '' else cmi.material_name end)  'material_name',
-                          (case when cbvi.signature is null then '' else cbvi.signature end)'md5',
-                          report.material_id 'material_id',
-                          cp.project_name 'project_name',
-                          cp.id as 'project_id',
-                          case when etlVideo.clip_name is null then '' else etlVideo.clip_name end 'clip_name',
-                          case when etlVideo.plan_name is null then '' else etlVideo.plan_name end 'plan_name',
-                          case when etlVideo.shot_name is null then '' else etlVideo.shot_name end 'shot_name',
-                          case when cmi.create_time is not null then cmi.create_time else '' end 'material_create_time',
-                          sum(report.cost) 'cost',
-                          sum(report.click) 'click',
-                          (case when sum(report.show_material)>0 and sum(report.click)>0 then round(sum(report.click)/sum(report.show_material)*100) else 0 end) 'ctr',
-                          sum(report.show_material) 'material_show',
-                          sum(report.convert_material) 'convert_material',
-                          (case when sum(report.convert_material)>0 and sum(report.click)>0 then round(sum(report.convert_material)/sum(report.click)*100) else 0 end) 'convert_rate',
-                          (case when sum(report.cost) >0 and sum(report.show_material)>0 then round(sum(report.cost)/sum(report.show_material)*1000,2) else 0 end) 'cpm',
-                          (case when sum(report.cost) >0 and sum(report.click)>0 then round(sum(report.cost)/sum(report.click),2) else 0 end) 'cpc',
-                          (case when sum(report.cost) >0 and sum(report.convert_material)>0 then round(sum(report.cost)/sum(report.convert_material),2) else 0 end) 'convert_cost',
-                          sum(report.download_start) 'download_start',
-                          sum(report.download_finish) 'download_finish',
-                          sum(report.active) 'active',
-                          sum(report.register) 'register',
-                          sum(report.next_day_open) 'next_day_open',
-                          sum(report.pay_count) 'active_pay_amount',
-                          (case when sum(report.cost) >0 and sum(report.pay_count)>0 then round(sum(report.cost)/sum(report.pay_count),2) else 0 end)  'active_pay_cost',
-                          (case when sum(report.cost) >0 and sum(report.next_day_open)>0 then round(sum(report.cost)/sum(report.next_day_open),2) else 0 end) 'next_day_open_cost',
-                          (case when sum(report.cost) >0 and sum(report.register)>0 then round(sum(report.cost)/sum(report.register),2) else 0 end) 'active_register_cost',
-                          (case when sum(report.cost) >0 and sum(report.active)>0 then round(sum(report.cost)/sum(report.active),2) else 0 end) 'active_cost',
-                          (case when sum(report.cost) >0 and sum(report.download_finish)>0 then round(sum(report.cost)/sum(report.download_finish),2) else 0 end) 'download_finish_cost',
-                          (case when sum(report.cost) >0 and sum(report.download_start)>0 then round(sum(report.cost)/sum(report.download_start),2) else 0 end) 'download_start_cost',
-                          (case when sum(report.download_start) >0 and sum(report.click)>0 then round(sum(report.download_start)/sum(report.click)*100,2) else 0 end) 'download_start_rate',
-                          (case when sum(report.download_finish) >0 and sum(report.download_start)>0 then round(sum(report.download_finish)/sum(report.download_start)*100,2) else 0 end) 'download_finish_rate',
-                          (case when sum(report.active) >0 and sum(report.click)>0 then round(sum(report.active)/sum(report.click)*100,2) else 0 end) 'active_rate',
-                          (case when sum(report.register) >0 and sum(report.active)>0 then round(sum(report.register)/sum(report.active)*100,2) else 0 end) 'active_register_rate',
-                          (case when sum(report.active) >0 and sum(report.next_day_open)>0 then round(sum(report.next_day_open)/sum(report.active)*100,2) else 0 end) 'next_day_open_rate',
-                          sum(report.form) 'form',
-                          (case when sum(report.cost) >0 and sum(report.valid_play)>0 then round(sum(report.cost)/sum(report.valid_play),2) else 0 end) 'valid_play_cost',
-                          sum(report.play75_feed_break) 'play75_feed_break',
-                          sum(report.play100_feed_break) 'play100_feed_break',
-                          sum(report.play25_feed_break) 'play25_feed_break',
-                          sum(report.play50_feed_break) 'play50_feed_break',
-                          sum(report.share_material) 'share_material',
-                          sum(report.follow) 'follow',
-                          sum(report.like_material) 'like_material',
-                          report.stat_datetime 'stat_datetime',
-                      report.account_id 'account_id'
-  from ctop_bytedance_report_material_daily report
-           left join ctop_user_allocation cua on cua.account_id = report.account_id
-           left join (select  account_id,material_id,signature from ctop_bytedance_video_info  where  account_id = #{accountId}) cbvi
-                     on report.account_id = cbvi.account_id and report.material_id = cbvi.material_id
-           left join ctop_bytedance_video_etl_info etlVideo on etlVideo.video_code = cbvi.signature
-           left join ctop_material_info cmi on cmi.code = cbvi.signature
-           left join ctop_project cp on cp.id = cua.project_id
-  where 1 = 1
-    and report.stat_datetime = #{date}
-    and report.account_id = #{accountId}
-  group by report.material_id,report.account_id;
+             SELECT *  from (
+			SELECT
+			t.*,
+			(case when t1.url is not null then t1.url else '' end) 'url',
+			(CASE when t1.cover_url is null then ''  ELSE t1.cover_url end) 'cover_url',
+			(case when t1.material_name is null then '' else t1.material_name end)  'material_name',
+			(case when t1.create_time is not null then t1.create_time else '' end) 'material_create_time'
+			FROM
+			(
+			SELECT
+			(CASE when cbvi.signature is null then ''  ELSE cbvi.signature end) 'md5',
+			CONCAT(report.material_id, '-') 'material_id',
+			cp.project_name 'project_name',
+			cp.id AS 'project_id',
+			CASE
+		WHEN etlVideo.clip_name IS NULL THEN
+			''
+		ELSE
+			etlVideo.clip_name
+		END 'clip_name',
+		CASE
+	WHEN etlVideo.plan_name IS NULL THEN
+		''
+	ELSE
+		etlVideo.plan_name
+	END 'plan_name',
+	CASE
+	WHEN etlVideo.shot_name IS NULL THEN
+	''
+	ELSE
+	etlVideo.shot_name
+	END 'shot_name',
+sum(report.cost) 'cost',
+ sum(report.click) 'click',
+ (
+	CASE
+	WHEN sum(report.show_material) > 0
+	AND sum(report.click) > 0 THEN
+		round(
+			sum(report.click) / sum(report.show_material) * 100
+		)
+	ELSE
+		0
+	END
+) 'ctr',
+ sum(report.show_material) 'material_show',
+ sum(report.convert_material) 'convert_material',
+ (
+	CASE
+	WHEN sum(report.convert_material) > 0
+	AND sum(report.click) > 0 THEN
+		round(
+			sum(report.convert_material) / sum(report.click) * 100
+		)
+	ELSE
+		0
+	END
+) 'convert_rate',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.show_material) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.show_material) * 1000,
+			2
+		)
+	ELSE
+		0
+	END
+) 'cpm',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.click) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.click),
+			2
+		)
+	ELSE
+		0
+	END
+) 'cpc',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.convert_material) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.convert_material),
+			2
+		)
+	ELSE
+		0
+	END
+) 'convert_cost',
+ sum(report.download_start) 'download_start',
+ sum(report.download_finish) 'download_finish',
+ sum(report.active) 'active',
+ sum(report.register) 'register',
+ sum(report.next_day_open) 'next_day_open',
+ sum(report.pay_count) 'active_pay_amount',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.pay_count) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.pay_count),
+			2
+		)
+	ELSE
+		0
+	END
+) 'active_pay_cost',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.next_day_open) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.next_day_open),
+			2
+		)
+	ELSE
+		0
+	END
+) 'next_day_open_cost',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.register) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.register),
+			2
+		)
+	ELSE
+		0
+	END
+) 'active_register_cost',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.active) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.active),
+			2
+		)
+	ELSE
+		0
+	END
+) 'active_cost',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.download_finish) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.download_finish),
+			2
+		)
+	ELSE
+		0
+	END
+) 'download_finish_cost',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.download_start) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.download_start),
+			2
+		)
+	ELSE
+		0
+	END
+) 'download_start_cost',
+ (
+	CASE
+	WHEN sum(report.download_start) > 0
+	AND sum(report.click) > 0 THEN
+		round(
+			sum(report.download_start) / sum(report.click) * 100,
+			2
+		)
+	ELSE
+		0
+	END
+) 'download_start_rate',
+ (
+	CASE
+	WHEN sum(report.download_finish) > 0
+	AND sum(report.download_start) > 0 THEN
+		round(
+			sum(report.download_finish) / sum(report.download_start) * 100,
+			2
+		)
+	ELSE
+		0
+	END
+) 'download_finish_rate',
+ (
+	CASE
+	WHEN sum(report.active) > 0
+	AND sum(report.click) > 0 THEN
+		round(
+			sum(report.active) / sum(report.click) * 100,
+			2
+		)
+	ELSE
+		0
+	END
+) 'active_rate',
+ (
+	CASE
+	WHEN sum(report.register) > 0
+	AND sum(report.active) > 0 THEN
+		round(
+			sum(report.register) / sum(report.active) * 100,
+			2
+		)
+	ELSE
+		0
+	END
+) 'active_register_rate',
+ (
+	CASE
+	WHEN sum(report.active) > 0
+	AND sum(report.next_day_open) > 0 THEN
+		round(
+			sum(report.next_day_open) / sum(report.active) * 100,
+			2
+		)
+	ELSE
+		0
+	END
+) 'next_day_open_rate',
+ sum(report.form) 'form',
+ (
+	CASE
+	WHEN sum(report.cost) > 0
+	AND sum(report.valid_play) > 0 THEN
+		round(
+			sum(report.cost) / sum(report.valid_play),
+			2
+		)
+	ELSE
+		0
+	END
+) 'valid_play_cost',
+ sum(report.play75_feed_break) 'play75_feed_break',
+ sum(report.play100_feed_break) 'play100_feed_break',
+ sum(report.play25_feed_break) 'play25_feed_break',
+ sum(report.play50_feed_break) 'play50_feed_break',
+ sum(report.share_material) 'share_material',
+ sum(report.follow) 'follow',
+ sum(report.like_material) 'like_material',
+ report.stat_datetime 'stat_datetime',
+ report.account_id 'account_id'
+	FROM
+	ctop_bytedance_report_material_daily report
+	LEFT JOIN ctop_user_allocation cua ON cua.account_id = report.account_id
+	LEFT JOIN (
+	SELECT
+		account_id,
+		material_id,
+		signature
+	FROM
+		ctop_bytedance_video_info
+	WHERE
+		account_id = #{accountId}
+ 	GROUP BY material_id
+	) cbvi ON report.account_id = cbvi.account_id
+	AND report.material_id = cbvi.material_id
+	LEFT JOIN ctop_bytedance_video_etl_info etlVideo ON etlVideo.video_code = cbvi.signature
+	LEFT JOIN ctop_project cp ON cp.id = cua.project_id
+	WHERE
+	1 = 1
+	AND report.stat_datetime = #{date}
+	AND report.account_id = #{accountId}
+	GROUP BY
+	report.material_id,
+	report.account_id
+	) t
+	LEFT JOIN ctop_material_info t1 ON t.md5 = t1.`code` ) t2
+	GROUP BY t2.md5
     </select>
     <select id="replaceEtlVideoInfo">
         REPLACE INTO etl_report_bytedance_video(
@@ -234,5 +456,4 @@
     </update>
 
 
-
 </mapper>

+ 1 - 1
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/report/service/IEtlBytedanceReportVideoDailyService.java

@@ -16,5 +16,5 @@ public interface IEtlBytedanceReportVideoDailyService extends IService<EtlByteda
     Result updBytedanceVideo(Long accountId,Long projectId,String timeStr) throws Exception;
 
 
-    void etlBytedanceVideo(Long accountId);
+    void etlBytedanceVideo(Long accountId,String date);
 }

+ 13 - 23
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/report/service/impl/EtlBytedanceReportVideoDailyServiceImpl.java

@@ -1,23 +1,15 @@
 package cn.com.ctop.toutiao.modules.report.service.impl;
 
 import cn.com.ctop.toutiao.modules.link.entity.ETLReportBytedanceVideo;
-import cn.com.ctop.toutiao.modules.material.entity.EtlBytedanceVideoInfo;
 import cn.com.ctop.toutiao.modules.report.entity.EtlBytedanceReportVideoDaily;
 import cn.com.ctop.toutiao.modules.report.mapper.EtlBytedanceReportVideoDailyMapper;
 import cn.com.ctop.toutiao.modules.report.service.IEtlBytedanceReportVideoDailyService;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
-import com.google.api.client.util.Data;
-import com.google.common.base.Function;
 import lombok.extern.slf4j.Slf4j;
-import org.checkerframework.checker.nullness.qual.Nullable;
 import org.jeecg.common.api.vo.Result;
-import org.jeecg.common.util.DateUtils;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import javax.annotation.Resource;
-import java.util.ArrayList;
-import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -42,20 +34,20 @@ public class EtlBytedanceReportVideoDailyServiceImpl extends ServiceImpl<EtlByte
     private EtlBytedanceReportVideoDailyMapper etlBytedanceReportVideoDailyMapper;
 
     @Override
-    public Result updBytedanceVideo(Long accountId,Long projectId,String timeStr) throws Exception{
-        CompletableFuture future = CompletableFuture.supplyAsync(()->{
-           return etlBytedanceReportVideoDailyMapper.getBytedanceVideoInfo(accountId,projectId,timeStr);
+    public Result updBytedanceVideo(Long accountId, Long projectId, String timeStr) throws Exception {
+        CompletableFuture future = CompletableFuture.supplyAsync(() -> {
+            return etlBytedanceReportVideoDailyMapper.getBytedanceVideoInfo(accountId, projectId, timeStr);
         }).handle(new BiFunction<List<Map<String, Object>>, Throwable, Object>() {
             @Override
             public Object apply(List<Map<String, Object>> list, Throwable throwable) {
-                if (throwable != null){
+                if (throwable != null) {
                     return Result.error(throwable.getMessage());
                 }
-                list.forEach(map ->{
-                    String coverUrl = map.get("posterUrl")+"";
-                    String url = map.get("videoUrl")+"";
-                    String md5 = map.get("md5")+"";
-                    etlBytedanceReportVideoDailyMapper.updBytedanceVideoByMd5(md5,coverUrl,url);
+                list.forEach(map -> {
+                    String coverUrl = map.get("posterUrl") + "";
+                    String url = map.get("videoUrl") + "";
+                    String md5 = map.get("md5") + "";
+                    etlBytedanceReportVideoDailyMapper.updBytedanceVideoByMd5(md5, coverUrl, url);
                 });
                 return Result.ok("修改成功。");
             }
@@ -65,10 +57,10 @@ public class EtlBytedanceReportVideoDailyServiceImpl extends ServiceImpl<EtlByte
 
     static ExecutorService executorService = Executors.newFixedThreadPool(20);
     static CountDownLatch countDownLatch = null;
+
     @Override
-    public void etlBytedanceVideo(Long accountId){
-        String date = DateUtils.formatDate(DateUtils.addDay(new Date(), -1));
-        List<ETLReportBytedanceVideo> list = etlBytedanceReportVideoDailyMapper.etlBytedanceVideoInfo(date,accountId);
+    public void etlBytedanceVideo(Long accountId, String date) {
+        List<ETLReportBytedanceVideo> list = etlBytedanceReportVideoDailyMapper.etlBytedanceVideoInfo(date, accountId);
         if (null != list && !list.isEmpty()) {
             countDownLatch = new CountDownLatch(list.size());
             list.forEach(videoInfo -> executorService.submit(() -> {
@@ -85,11 +77,9 @@ public class EtlBytedanceReportVideoDailyServiceImpl extends ServiceImpl<EtlByte
             } catch (InterruptedException e) {
                 e.printStackTrace();
             }
-            log.info("数据清洗完成");
+            log.info("数据清洗完成,accountId:{},date:{}", accountId, date);
         }
     }
 
 
-
-
 }