系列文章目錄

物流實時數倉：采集通道搭建
物流實時數倉：數倉搭建

---

---

前言

現在我們開始進行數倉的搭建，我們用Kafka來代替數倉的ods層。
基本流程為使用Flink從MySQL讀取數據然后寫入Kafka中

---

一、IDEA環境準備

1.pom.xml

寫入項目需要的配置

<properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><java.version>1.8</java.version><flink.version>1.17.0</flink.version><hadoop.version>3.2.3</hadoop.version><flink-cdc.version>2.3.0</flink-cdc.version></properties><dependencies><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka</artifactId><version>${flink.version}</version></dependency><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.68</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version><exclusions><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-reload4j</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-api</artifactId><version>1.7.25</version></dependency><dependency><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId><version>1.7.25</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-to-slf4j</artifactId><version>2.14.0</version></dependency><dependency><groupId>com.ververica</groupId><artifactId>flink-connector-mysql-cdc</artifactId><version>${flink-cdc.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-runtime</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-loader</artifactId><version>${flink.version}</version></dependency></dependencies>

基本上項目需要的所有jar包都有了，不夠以后在加。

2.目錄創建

按照以上目錄結構進行目錄創建

二、代碼編寫

1.log4j.properties

log4j.rootLogger=error,stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n

2.CreateEnvUtil.java

這個文件中有兩個方法
創建初始化Flink的env
Flink連接mysql的MySqlSource

package com.atguigu.tms.realtime.utils;import com.esotericsoftware.minlog.Log;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.connect.json.DecimalFormat;
import org.apache.kafka.connect.json.JsonConverterConfig;import java.util.HashMap;public class CreateEnvUtil {public static StreamExecutionEnvironment getStreamEnv(String[] args) {// 1.1 指定流處理環境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 2.檢查點相關設置// 2.1 開啟檢查點env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);// 2.2 設置檢查點的超時時間env.getCheckpointConfig().setCheckpointTimeout(120000L);// 2.3 設置job取消之后 檢查點是否保留env.getCheckpointConfig().setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);// 2.4 設置兩個檢查點之間的最小時間間隔env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000L);// 2.5 設置重啟策略env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1), Time.seconds(3)));// 2.6 設置狀態后端env.setStateBackend(new HashMapStateBackend());env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/tms/ck");// 2.7 設置操作hdfs用戶// 獲取命令行參數ParameterTool parameterTool = ParameterTool.fromArgs(args);String hdfsUserName = parameterTool.get("hadoop-user-name", "atguigu");System.setProperty("HADOOP_USER_NAME", hdfsUserName);return env;}public static MySqlSource<String> getMysqlSource(String option, String serverId, String[] args) {ParameterTool parameterTool = ParameterTool.fromArgs(args);String mysqlHostname = parameterTool.get("hadoop-user-name", "hadoop102");int mysqlPort = Integer.parseInt(parameterTool.get("mysql-port", "3306"));String mysqlUsername = parameterTool.get("mysql-username", "root");String mysqlPasswd = parameterTool.get("mysql-passwd", "000000");option = parameterTool.get("start-up-option", option);serverId = parameterTool.get("server-id", serverId);// 創建配置信息 Map 集合，將 Decimal 數據類型的解析格式配置 k-v 置于其中HashMap config = new HashMap<>();config.put(JsonConverterConfig.DECIMAL_FORMAT_CONFIG, DecimalFormat.NUMERIC.name());// 將前述 Map 集合中的配置信息傳遞給 JSON 解析 Schema，該 Schema 將用于 MysqlSource 的初始化JsonDebeziumDeserializationSchema jsonDebeziumDeserializationSchema =new JsonDebeziumDeserializationSchema(false, config);MySqlSourceBuilder<String> builder = MySqlSource.<String>builder().hostname(mysqlHostname).port(mysqlPort).username(mysqlUsername).password(mysqlPasswd).deserializer(jsonDebeziumDeserializationSchema);switch (option) {// 讀取實時數據case "dwd":String[] dwdTables = new String[]{"tms.order_info","tms.order_cargo","tms.transport_task","tms.order_org_bound"};return builder.databaseList("tms").tableList(dwdTables).startupOptions(StartupOptions.latest()).serverId(serverId).build();// 讀取維度數據case "realtime_dim":String[] realtimeDimTables = new String[]{"tms.user_info","tms.user_address","tms.base_complex","tms.base_dic","tms.base_region_info","tms.base_organ","tms.express_courier","tms.express_courier_complex","tms.employee_info","tms.line_base_shift","tms.line_base_info","tms.truck_driver","tms.truck_info","tms.truck_model","tms.truck_team"};return builder.databaseList("tms").tableList(realtimeDimTables).startupOptions(StartupOptions.initial()).serverId(serverId).build();}Log.error("不支持操作類型");return null;}
}

3.KafkaUtil.java

該文件中有一個方法，創建Flink連接Kafka需要的Sink

package com.atguigu.tms.realtime.utils;import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.kafka.clients.producer.ProducerConfig;public class KafkaUtil {private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";public static KafkaSink<String> getKafkaSink(String topic, String transIdPrefix, String[] args) {// 將命令行參數對象封裝為 ParameterTool 類對象ParameterTool parameterTool = ParameterTool.fromArgs(args);// 提取命令行傳入的 key 為 topic 的配置信息，并將默認值指定為方法參數 topic// 當命令行沒有指定 topic 時，會采用默認值topic = parameterTool.get("topic", topic);// 如果命令行沒有指定主題名稱且默認值為 null 則拋出異常if (topic == null) {throw new IllegalArgumentException("主題名不可為空：命令行傳參為空且沒有默認值!");}// 獲取命令行傳入的 key 為 bootstrap-servers 的配置信息，并指定默認值String bootstrapServers = parameterTool.get("bootstrap-severs", KAFKA_SERVER);// 獲取命令行傳入的 key 為 transaction-timeout 的配置信息，并指定默認值String transactionTimeout = parameterTool.get("transaction-timeout", 15 * 60 * 1000 + "");return KafkaSink.<String>builder().setBootstrapServers(bootstrapServers).setRecordSerializer(KafkaRecordSerializationSchema.builder().setTopic(topic).setValueSerializationSchema(new SimpleStringSchema()).build()).setDeliveryGuarantee(DeliveryGuarantee.AT_LEAST_ONCE).setTransactionalIdPrefix(transIdPrefix).setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, transactionTimeout).build();}public static KafkaSink<String> getKafkaSink(String topic, String[] args) {return getKafkaSink(topic, topic + "_trans", args);}
}

4.OdsApp.java

Ods層的app創建，負責讀取和寫入數據

package com.atguigu.tms.realtime.app.ods;import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tms.realtime.utils.CreateEnvUtil;
import com.atguigu.tms.realtime.utils.KafkaUtil;
import com.esotericsoftware.minlog.Log;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;public class OdsApp {public static void main(String[] args) throws Exception {// 1.獲取流處理環境并指定檢查點StreamExecutionEnvironment env = CreateEnvUtil.getStreamEnv(args);env.setParallelism(4);// 2 使用FlinkCDC從MySQL中讀取數據-事實數據String dwdOption = "dwd";String dwdServerId = "6030";String dwdsourceName = "ods_app_dwd_source";mysqlToKafka(dwdOption, dwdServerId, dwdsourceName, env, args);// 3 使用FlinkCDC從MySQL中讀取數據-維度數據String realtimeDimOption = "realtime_dim";String realtimeDimServerId = "6040";String realtimeDimsourceName = "ods_app_realtimeDim_source";mysqlToKafka(realtimeDimOption, realtimeDimServerId, realtimeDimsourceName, env, args);env.execute();}public static void mysqlToKafka(String option, String serverId, String sourceName, StreamExecutionEnvironment env, String[] args) {MySqlSource<String> MySqlSource = CreateEnvUtil.getMysqlSource(option, serverId, args);SingleOutputStreamOperator<String> dwdStrDS = env.fromSource(MySqlSource, WatermarkStrategy.noWatermarks(), sourceName).setParallelism(1).uid(option + sourceName);// 3 簡單ETLSingleOutputStreamOperator<String> processDS = dwdStrDS.process(new ProcessFunction<String, String>() {@Overridepublic void processElement(String jsonStr, ProcessFunction<String, String>.Context ctx, Collector<String> out) {try {JSONObject jsonObj = JSONObject.parseObject(jsonStr);if (jsonObj.getJSONObject("after") != null && !"d".equals(jsonObj.getString("op"))) {
//                                System.out.println(jsonObj);Long tsMs = jsonObj.getLong("ts_ms");jsonObj.put("ts", tsMs);jsonObj.remove("ts_ms");String jsonString = jsonObj.toJSONString();out.collect(jsonString);}} catch (Exception e) {Log.error("從Flink-CDC得到的數據不是一個標準的json格式",e);}}}).setParallelism(1);// 4 按照主鍵進行分組，避免出現亂序KeyedStream<String, String> keyedDS = processDS.keyBy((KeySelector<String, String>) jsonStr -> {JSONObject jsonObj = JSON.parseObject(jsonStr);return jsonObj.getJSONObject("after").getString("id");});//將數據寫入KafkakeyedDS.sinkTo(KafkaUtil.getKafkaSink("tms_ods", sourceName + "_transPre", args)).uid(option + "_ods_app_sink");}
}

三、代碼測試

在虛擬機啟動我們需要的組件，目前需要hadoop、zk、kafka和MySQL。

先開一個消費者進行消費。

bin/kafka-console-consumer.sh --bootstrap-server hadoop102:9092 --topic tms_ods

然后運行OdsApp.java
他會先讀取維度數據，因為維度數據需要全量更新之前的數據。

當他消費結束后，我們運行jar包，獲取事實數據。

java -jar tms-mock-2023-01-06.jar 

如果能消費到新數據，代表通道沒問題，ODS層創建完成。

---

總結

至此ODS搭建完成。