|
|
@@ -1,11 +1,11 @@
|
|
|
package com.qucheng.game.data.oldsystem.ods;
|
|
|
|
|
|
+import com.alibaba.fastjson2.JSON;
|
|
|
+import com.alibaba.fastjson2.JSONObject;
|
|
|
import com.qucheng.game.data.oldsystem.Env;
|
|
|
import com.qucheng.game.data.oldsystem.pojo.FlinkAppConfigParam;
|
|
|
-import com.qucheng.game.data.oldsystem.pojo.MysqlConfigParam;
|
|
|
-import com.qucheng.game.data.oldsystem.pojo.TransportMap;
|
|
|
-import com.qucheng.game.data.oldsystem.serialization.MapDebeziumDeserializationSchema;
|
|
|
-import com.qucheng.game.data.oldsystem.sink.CdcMysqlTablesSink;
|
|
|
+import com.qucheng.game.data.oldsystem.serialization.CustomerDeserializationSchema;
|
|
|
+import com.qucheng.game.data.oldsystem.sink.ads_sink;
|
|
|
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
|
|
|
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
|
|
|
import org.apache.commons.lang3.StringUtils;
|
|
|
@@ -21,11 +21,14 @@ import org.apache.flink.streaming.api.environment.CheckpointConfig;
|
|
|
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
|
|
|
|
|
|
import java.util.Arrays;
|
|
|
-import java.util.HashMap;
|
|
|
-import java.util.Map;
|
|
|
+import java.util.List;
|
|
|
import java.util.Properties;
|
|
|
import java.util.concurrent.TimeUnit;
|
|
|
|
|
|
+/**
|
|
|
+ * ODS层数据填充
|
|
|
+ * 过滤字段及转换字段
|
|
|
+ */
|
|
|
public class ByteDailyCost {
|
|
|
private static final FlinkAppConfigParam appConfigParam = Env.isTest ? FlinkAppConfigParam.builder()
|
|
|
.savePath("")
|
|
|
@@ -82,7 +85,7 @@ public class ByteDailyCost {
|
|
|
env.getCheckpointConfig().setCheckpointStorage(appConfigParam.getSavePath());
|
|
|
}
|
|
|
|
|
|
- MySqlSource<TransportMap> mysqlCDCSource = MySqlSource.<TransportMap>builder()
|
|
|
+ MySqlSource<String> mysqlCDCSource = MySqlSource.<String>builder()
|
|
|
.hostname(props.getProperty("cdc.mysql.backup.host"))
|
|
|
.port(StringUtils.isBlank(props.getProperty("cdc.mysql.backup.port")) ? 3306 : Integer.parseInt(props.getProperty("cdc.mysql.backup.port")))
|
|
|
.username(props.getProperty("cdc.mysql.backup.username"))
|
|
|
@@ -91,48 +94,62 @@ public class ByteDailyCost {
|
|
|
.tableList((StringUtils.join(new String[]{
|
|
|
"ods.byte_t_ad_data_day"
|
|
|
}, ",")))
|
|
|
- .deserializer(new MapDebeziumDeserializationSchema())
|
|
|
+ .deserializer(new CustomerDeserializationSchema())
|
|
|
+ //5400 和 6400
|
|
|
.startupOptions(StartupOptions.initial())
|
|
|
.build();
|
|
|
|
|
|
- MysqlConfigParam sinkMysqlConfig = MysqlConfigParam.builder()
|
|
|
- .url(props.getProperty("mysql.quChengText.url"))
|
|
|
- .username(props.getProperty("mysql.quChengText.username"))
|
|
|
- .password(props.getProperty("mysql.quChengText.password"))
|
|
|
- .build();
|
|
|
+ DataStreamSource<String> dataStreamSource = env.fromSource(mysqlCDCSource, WatermarkStrategy.noWatermarks(), "ODS_Mysql_Source");
|
|
|
+
|
|
|
+ dataStreamSource.map(new FieldFilterMap()).addSink(new ads_sink());
|
|
|
|
|
|
- DataStreamSource<TransportMap> cdcSource = env.fromSource(mysqlCDCSource, WatermarkStrategy.noWatermarks(), "mysqlCDCSource");
|
|
|
- cdcSource.print();
|
|
|
- cdcSource.map(new FieldFilterMap())
|
|
|
- .addSink(new CdcMysqlTablesSink(sinkMysqlConfig));
|
|
|
|
|
|
- env.execute(ByteDailyCost.class.getSimpleName());
|
|
|
+ //4.启动任务
|
|
|
+ env.execute("ByteDailyCost");
|
|
|
}
|
|
|
|
|
|
- /**
|
|
|
- * 字段过滤和映射
|
|
|
- */
|
|
|
- public static class FieldFilterMap implements MapFunction<TransportMap, TransportMap> {
|
|
|
+ public static class FieldFilterMap implements MapFunction<String, JSONObject> {
|
|
|
@Override
|
|
|
- public TransportMap map(TransportMap transportMap) throws Exception {
|
|
|
- if (transportMap.getAfter() == null) {
|
|
|
- return transportMap;
|
|
|
- }
|
|
|
- Map<String, Object> after = transportMap.getAfter();
|
|
|
- if (transportMap.getTableName().equalsIgnoreCase("byte_t_ad_data_day")) {
|
|
|
- Map<String, Object> pojo = new HashMap<>(after.size());
|
|
|
- pojo.put("account_id",after.get("account_id"));
|
|
|
- pojo.put("date",after.get("day"));
|
|
|
- pojo.put("cost",after.get("cost"));
|
|
|
- pojo.put("view_count",after.get("show"));
|
|
|
- pojo.put("valid_click_count",after.get("click"));
|
|
|
- pojo.put("ctr",after.get("ctr"));
|
|
|
- after = pojo;
|
|
|
- transportMap.setTableName("daily_tt");
|
|
|
- transportMap.setPrimaryKeys(Arrays.asList("account_id", "day"));
|
|
|
- transportMap.setAfter(after);
|
|
|
+ public JSONObject map(String value) throws Exception {
|
|
|
+
|
|
|
+ JSONObject result = new JSONObject();
|
|
|
+ JSONObject data = JSON.parseObject(value);
|
|
|
+
|
|
|
+ String tableName = data.getString("tableName");
|
|
|
+ JSONObject after = data.getJSONObject("after");
|
|
|
+ JSONObject key = data.getJSONObject("key");
|
|
|
+ String afterFinal = "";
|
|
|
+ String keyFinal = "";
|
|
|
+ String tableNameFinal = "";
|
|
|
+
|
|
|
+ if (tableName.equals("byte_t_ad_data_day")) {
|
|
|
+
|
|
|
+ List<String> columns = Arrays.asList("account_id","ad_id", "day",
|
|
|
+ "cost", "show_count", "click", "ctr","active");
|
|
|
+
|
|
|
+ after.entrySet().removeIf(next -> !columns.contains(next.getKey()));
|
|
|
+
|
|
|
+ keyFinal = key.toString().replaceAll("\"day\":", "\"date\":");
|
|
|
+
|
|
|
+ afterFinal = after.toString().replaceAll("\"day\":", "\"date\":")
|
|
|
+ .replaceAll("\"show_count\":", "\"view_count\":")
|
|
|
+ .replaceAll("\"click\":", "\"valid_click_count\":")
|
|
|
+ .replaceAll("\"active\":", "\"from_follow_uv\":");
|
|
|
+
|
|
|
+
|
|
|
+ tableNameFinal = "daily_tt";
|
|
|
}
|
|
|
- return transportMap;
|
|
|
+
|
|
|
+ Object afterFinalJson = JSONObject.parse(afterFinal);
|
|
|
+ Object keyFinalJson = JSONObject.parse(keyFinal);
|
|
|
+
|
|
|
+ result.put("after", afterFinalJson);
|
|
|
+ result.put("key", keyFinalJson);
|
|
|
+ result.put("type", data.getString("type"));
|
|
|
+ result.put("tableName", tableNameFinal);
|
|
|
+ result.put("db", "quchen_text");
|
|
|
+
|
|
|
+ return result;
|
|
|
}
|
|
|
}
|
|
|
-}
|
|
|
+}
|
cola