|
@@ -7,6 +7,8 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
|
|
|
import cn.com.ctop.job.bytedance.data.utils.Check;
|
|
import cn.com.ctop.job.bytedance.data.utils.Check;
|
|
|
import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
|
|
import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
|
|
|
import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
|
|
import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
|
|
|
|
|
+import cn.com.ctop.job.bytedance.utils.doris_sink.DorisMultiTableStreamLoadManager;
|
|
|
|
|
+import cn.com.ctop.job.bytedance.utils.doris_sink.RecordWithMeta;
|
|
|
import com.alibaba.fastjson.JSONArray;
|
|
import com.alibaba.fastjson.JSONArray;
|
|
|
import com.alibaba.fastjson.JSONObject;
|
|
import com.alibaba.fastjson.JSONObject;
|
|
|
import lombok.extern.slf4j.Slf4j;
|
|
import lombok.extern.slf4j.Slf4j;
|
|
@@ -24,15 +26,10 @@ import java.util.*;
|
|
|
@EnableScheduling
|
|
@EnableScheduling
|
|
|
public class BytedanceAccountReportServiceImpl implements IBytedanceAccountReportService {
|
|
public class BytedanceAccountReportServiceImpl implements IBytedanceAccountReportService {
|
|
|
@Autowired
|
|
@Autowired
|
|
|
- private BytedanceAccountReportMapper accountReportMapper;
|
|
|
|
|
-
|
|
|
|
|
- /*private static final int MAX_BATCH_SIZE = 5000;
|
|
|
|
|
- private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
|
|
|
|
|
|
|
+ private DorisMultiTableStreamLoadManager dorisLoadManager;
|
|
|
|
|
|
|
|
- private final List<BytedanceAdvertiserReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
|
|
|
|
|
- private final List<BytedanceAdvertiserReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
|
|
|
|
|
- private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
|
|
|
|
|
- private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();*/
|
|
|
|
|
|
|
+ @Autowired
|
|
|
|
|
+ private BytedanceAccountReportMapper accountReportMapper;
|
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
@Override
|
|
@@ -231,6 +228,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
|
|
|
advertiserReportHour.setStatTimeDay(statTimeDay);
|
|
advertiserReportHour.setStatTimeDay(statTimeDay);
|
|
|
advertiserReportHour.setHour(hour);
|
|
advertiserReportHour.setHour(hour);
|
|
|
hours.add(advertiserReportHour);
|
|
hours.add(advertiserReportHour);
|
|
|
|
|
+
|
|
|
|
|
+ dorisLoadManager.add(new RecordWithMeta("media_api","bytedance_advertiser_report_hour",advertiserReportHour));
|
|
|
}
|
|
}
|
|
|
} else {
|
|
} else {
|
|
|
log.info("账户id:{},日期:{}广告时报(账户粒度)请求结果为空", advertiserId, startDate);
|
|
log.info("账户id:{},日期:{}广告时报(账户粒度)请求结果为空", advertiserId, startDate);
|
|
@@ -253,8 +252,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
|
|
|
}
|
|
}
|
|
|
if (!Check.isNull(hours)) {
|
|
if (!Check.isNull(hours)) {
|
|
|
accountReportMapper.replaceAccountReportHour(hours);
|
|
accountReportMapper.replaceAccountReportHour(hours);
|
|
|
- accountReportMapper.replaceAccountReportHourSD(hours);
|
|
|
|
|
- //advertiserReportHourBuffer.addAll(hours);
|
|
|
|
|
|
|
+ //accountReportMapper.replaceAccountReportHourSD(hours);
|
|
|
}
|
|
}
|
|
|
if (hours.size() >= 100) {
|
|
if (hours.size() >= 100) {
|
|
|
try {
|
|
try {
|
|
@@ -350,7 +348,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
|
|
|
} catch (InterruptedException e) {
|
|
} catch (InterruptedException e) {
|
|
|
throw new RuntimeException(e);
|
|
throw new RuntimeException(e);
|
|
|
}
|
|
}
|
|
|
- getBytedanceAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
|
|
|
|
|
|
|
+ getBytedanceAccountDaily(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
|
|
|
} else {
|
|
} else {
|
|
|
log.info("账户ID:{},日期:{},广告日报(账户粒度)数据请求成功", advertiserId, startDate);
|
|
log.info("账户ID:{},日期:{},广告日报(账户粒度)数据请求成功", advertiserId, startDate);
|
|
|
}
|
|
}
|