天天看點
傳回

Kafka實時資料即席查詢應用與實踐

作者：dbaplus社群

Kafka中的實時資料是以Topic的概念進行分類存儲，而Topic的資料是有一定時效性的，比如儲存24小時、36小時、48小時等。而在定位一些實時資料的Case時，如果沒有對實時資料進行曆史歸檔，在排查問題時，沒有日志追述，會很難定位是哪個環節的問題。

一、背景

Kafka中的實時資料是以Topic的概念進行分類存儲，而Topic的資料是有一定時效性的，比如儲存24小時、36小時、48小時等。而在定位一些實時資料的Case時，如果沒有對實時資料進行曆史歸檔，在排查問題時，沒有日志追述，會很難定位是哪個環節的問題。是以，我們需要對處理的這些實時資料進行記錄歸檔并存儲。

二、内容

1.案例分析

這裡以i視訊和vivo短視訊實時資料為例，之前存在這樣的協作問題：

資料上遊内容方提供實時Topic（存放i視訊和vivo短視訊相關實時資料），資料側對實時資料進行邏輯處理後，發送給下遊工程去建庫實時索引，當任務執行一段時間後，工程側建索引偶爾會提出資料沒有發送過去的Case，前期由于沒有對資料做存儲，在定位問題的時候會比較麻煩，經常需求檢視實時日志，需要花費很長的時間來分析這些Case是出現在哪個環節。

為了解決這個問題，我們可以将實時Topic中的資料，在發送給其他Topic的時候，添加跟蹤機制，進行資料分流，Sink到存儲媒體（比如HDFS、Hive等）。這裡，我們選擇使用Hive來進行存儲，主要是查詢友善，支援SQL來快速查詢。如下圖所示：

Kafka實時資料即席查詢應用與實踐

在實作優化後的方案時，有兩種方式可以實作跟蹤機制，它們分别是Flink SQL寫Hive、Flink DataStream寫Hive。接下來，分别對這兩種實作方案進行介紹和實踐。

方案一：Flink SQL寫Hive

這種方式比較直接，可以在Flink任務裡面直接操作實時Topic資料後，将消費後的資料進行分流跟蹤，作為日志記錄寫入到Hive表中，具體實作步驟如下：

  • 構造Hive Catalog；
  • 建立Hive表；
  • 寫入實時資料到Hive表。

1）構造Hive Catalog

在構造Hive Catalog時，需要初始化Hive的相關資訊，部分代碼片段如下所示：

// 設定執行環境
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
 EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().build();
 StreamTableEnvironment tEnv = StreamTableEnvironment.create(env,settings);


 // 構造 Hive Catalog 名稱
 String name = "video-hive-catalog";
 // 初始化資料庫名
 String defaultDatabase = "comsearch";
 // Hive 配置檔案路徑位址
 String hiveConfDir = "/appcom/hive/conf";
 // Hive 版本号
 String version = "3.1.2";
 // 執行個體化一個 HiveCatalog 對象
 HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
 // 注冊HiveCatalog
 tEnv.registerCatalog(name, hive);
 // 設定目前 HiveCatalog
 tEnv.useCatalog(name);
 // 設定執行SQL為Hive
 tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
 // 使用資料庫
 tEnv.useDatabase("db1");

在以上代碼中，我們首先設定了 Flink 的執行環境和表環境，然後建立了一個 HiveCatalog，并将其注冊到表環境中。

2）建立Hive表

如果Hive表不存在，可以通過在程式中執行建表語句，具體SQL見表語句代碼如下所示：

-- 建立表語句 
tEnv.executeSql("CREATE TABLE IF NOT EXISTS TABLE `xxx_table`(
  `content_id` string,
  `status` int)
PARTITIONED BY (
  `dt` string,
  `h` string,
  `m` string)
stored as ORC
TBLPROPERTIES (
  'auto-compaction'='true',
  'sink.partition-commit.policy.kind'='metastore,success-file',
  'partition.time-extractor.timestamp-pattern'='$dt $h:$m:00'
)")

在建立Hive表時我們使用了IF NOT EXISTS關鍵字，如果Hive中該表不存在會自動在Hive上建立，也可以提前在Hive中建立好該表，Flink SQL中就無需再執行建表SQL，因為用了Hive的Catalog，Flink SQL運作時會找到表。這裡，我們設定了auto-compaction屬性為true，用來使小檔案自動合并，1.12版的新特性，解決了實時寫Hive産生的小檔案問題。同時，指定metastore值是專門用于寫入Hive的，也需要指定success-file值，這樣CheckPoint觸發完資料寫入磁盤後會建立_SUCCESS檔案以及Hive metastore上建立中繼資料，這樣Hive才能夠對這些寫入的資料可查。

3）寫入實時資料到Hive表

在準備完成2.2.1和2.2.2中的步驟後，接下來就可以在Flink任務中通過SQL來對實時資料進行操作了，具體實作代碼片段如下所示：

// 編寫業務SQL
 String insertSql = "insert into  xxx_table SELECT content_id, status, " +
                    " DATE_FORMAT(ts, 'yyyy-MM-dd'), DATE_FORMAT(ts, 'HH'), DATE_FORMAT(ts, 'mm') FROM xxx_rt";
 // 執行 Hive SQL
 tEnv.executeSql(insertSql);
 // 執行任務
 env.execute();

将消費後的資料進行分類，編寫業務SQL語句，将消費的資料作為日志記錄，發送到Hive表進行存儲，這樣Kafka中的實時資料就存儲到Hive了，友善使用Hive來對Kafka資料進行即席分析。

4）避坑技巧

使用這種方式在處理的過程中，如果配置使用的是EventTime，在程式中配置'sink.partition-commit.trigger'='partition-time'，最後會出現無法送出分區的情況。經過對源代碼PartitionTimeCommitTigger的分析，找到了出現這種異常情況的原因。

我們可以通過看

org.apache.flink.table.filesystem.stream.PartitionTimeCommitTigger#committablePartitionsorg.apache.flink.table.filesystem.stream.PartitionTimeCommitTigger#committablePartitions

中的一個函數，來說明具體的問題，部分源代碼片段如下：

// PartitionTimeCommitTigger源代碼函數代碼片段
@Override
public List<String> committablePartitions(long checkpointId) {
 if (!watermarks.containsKey(checkpointId)) {
  throw new IllegalArgumentException(String.format(
    "Checkpoint(%d) has not been snapshot. The watermark information is: %s.",
    checkpointId, watermarks));
 }


 long watermark = watermarks.get(checkpointId);
 watermarks.headMap(checkpointId, true).clear();


 List<String> needCommit = new ArrayList<>();
 Iterator<String> iter = pendingPartitions.iterator();
 while (iter.hasNext()) {
  String partition = iter.next();
  // 通過分區的值來擷取分區的時間
  LocalDateTime partTime = extractor.extract(
    partitionKeys, extractPartitionValues(new Path(partition)));
  // 判斷水印是否大于分區建立時間+延遲時間
  if (watermark > toMills(partTime) + commitDelay) {
   needCommit.add(partition);
   iter.remove();
  }
 }
 return needCommit;
}

通過分析上述代碼片段，我們可以知道系統通過分區值來抽取相應的分區來建立時間，然後進行比對，比如我們設定的時間 pattern 是 '$dt $h:$m:00' , 某一時刻我們正在往 /2022-02-26/18/20/ 這個分區下寫資料，那麼程式根據分區值，得到的 pattern 将會是2022-02-26 18:20:00，這個值在SQL中是根據 DATA_FORMAT 函數擷取的。

而這個值是帶有時區的，比如我們的時區設定為東八區，2022-02-26 18:20:00這個時間是東八區的時間，換成标準 UTC 時間是減去8個小時，也就是2022-02-26 10:20:00，而在源代碼中的 toMills 函數在處理這個東八區的時間時，并沒有對時區進行處理，把這個其實應該是東八區的時間當做了 UTC 時間來處理，這樣計算出來的值就比實際值大8小時，導緻一直沒有觸發分區的送出。

如果我們在資料源中構造的分區是 UTC 時間，也就是不帶分區的時間，那麼這個邏輯就是沒有問題的，但是這樣又不符合我們的實際情況，比如對于分區2022-02-26 18:20:00，我希望我的分區肯定是東八區的時間，而不是比東八區小8個小時的UTC時間2022-02-26 10:20:00。

在明白了原因之後，我們就可以針對上述異常情況進行優化我們的實作方案，比如自定義一個分區類、或者修改預設的時間分區類。比如，我們使用TimeZoneTableFunction類來實作一個自定義時區，部分參考代碼片段如下：

public class CustomTimeZoneTableFunction implements TimeZoneTableFunction {


  private transient DateTimeFormatter formatter;
  private String timeZoneId;


  public CustomTimeZoneTableFunction(String timeZoneId) {
    this.timeZoneId = timeZoneId;
  }


  @Override
  public void open(FunctionContext context) throws Exception {
    // 初始化 DateTimeFormatter 對象
    formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:00");
    formatter = formatter.withZone(ZoneId.of(timeZoneId));
  }


  @Override
  public void eval(Long timestamp, Collector<TimestampWithTimeZone> out) {
    // 将時間戳轉換為 LocalDateTime 對象
    LocalDateTime localDateTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(timestamp), ZoneOffset.UTC);
    // 将 LocalDateTime 對象轉換為指定時區下的 LocalDateTime 對象
    LocalDateTime targetDateTime = localDateTime.atZone(ZoneId.of(timeZoneId)).toLocalDateTime();
    // 将 LocalDateTime 對象轉換為 TimestampWithTimeZone 對象，并輸出到下遊
    out.collect(TimestampWithTimeZone.fromLocalDateTime(targetDateTime, ZoneId.of(timeZoneId)));
  }
}

方案二：Flink DataStream寫Hive

在一些特殊的場景下，Flink SQL如果無法實作我們複雜的業務需求，那麼我們可以考慮使用Flink DataStream寫Hive這種實作方案。比如如下業務場景，現在需要實作這樣一個業務需求，内容方将實時資料寫入到Kafka消息隊列中，然後由資料側通過Flink任務消費内容方提供的資料源，接着對消費的資料進行分流處理（這裡的步驟和Flink SQL寫Hive的步驟類似），每分鐘進行存儲到HDFS（MapReduce任務需要計算和重跑HDFS資料），然後通過MapReduce任務将HDFS上的這些日志資料生成Hive所需要格式，最後将這些Hive格式資料檔案加載到Hive表中。實作Kafka資料到Hive的即席分析功能，具體實作流程細節如下圖所示：

Kafka實時資料即席查詢應用與實踐

具體核心實作步驟如下：

  • 消費内容方Topic實時資料；
  • 生成資料預處理政策；
  • 加載資料；
  • 使用Hive SQL對Kafka資料進行即席分析。

1）消費内容方Topic實時資料

編寫消費Topic的Flink代碼，這裡不對Topic中的資料做邏輯處理，在後面統一交給MapReduce來做資料預處理，直接消費并存儲到HDFS上。具體實作代碼如下所示：

public class Kafka2Hdfs {


    public static void main(String[] args) {
        // 判斷參數是否有效
        if (args.length != 3) {
            LOG.error("kafka(server01:9092), hdfs(hdfs://cluster01/data/), flink(parallelism=2) must be exist.");
            return;
        }
        // 初始化Kafka連接配接位址和HDFS存儲位址以及Flink并行度
        String bootStrapServer = args[0];
        String hdfsPath = args[1];
        int parallelism = Integer.parseInt(args[2]);


        // 執行個體化一個Flink任務對象
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000);
        env.setParallelism(parallelism);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);


        // Flink消費Topic中的資料
        DataStream<String> transction = env.addSource(new FlinkKafkaConsumer010<>("test_bll_topic", new SimpleStringSchema(), configByKafkaServer(bootStrapServer)));


        // 執行個體化一個HDFS存儲對象
        BucketingSink<String> sink = new BucketingSink<>(hdfsPath);


        // 自定義存儲到HDFS上的檔案名，用小時和分鐘來命名，友善後面算政策
        sink.setBucketer(new DateTimeBucketer<String>("HH-mm"));
        // 設定存儲HDFS的檔案大小和存儲檔案時間頻率
        sink.setBatchSize(1024 * 1024 * 4);
        sink.setBatchRolloverInterval(1000 * 30);
        transction.addSink(sink);


        env.execute("Kafka2Hdfs");
    }




    // 初始化Kafka對象連接配接資訊
    private static Object configByKafkaServer(String bootStrapServer) {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", bootStrapServer);
        props.setProperty("group.id", "test_bll_group");
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }


}

注意事項：

  • 這裡我們把時間視窗設定小一些，每30s做一次Checkpoint，如果該批次的時間視窗沒有資料過來，就生成一個檔案落地到HDFS上；
  • 另外，我們重寫了Bucketer為DateTimeBucketer，邏輯并不複雜，在原有的方法上加一個年-月-日/時-分的檔案生成路徑，例如在HDFS上的生成路徑：xxxx/2022-02-26/00-00。

具體DateTimeBucketer實作代碼如下所示：

public class DateMinuteBucketer implements Bucketer<String> {
    private SimpleDateFormat baseFormatDay = new SimpleDateFormat("yyyy-MM-dd");
    private SimpleDateFormat baseFormatMin = new SimpleDateFormat("HH-mm");


    @Override
    public Path getBucketPath(Clock clock, Path basePath, String element) {
        return new Path(basePath + "/" + baseFormatDay.format(new Date()) + "/" + baseFormatMin.format(new Date()));
    }
}

2）生成資料預處理政策

這裡，我們需要對落地到HDFS上的檔案進行預處理，處理的邏輯是這樣的。比如，現在是2022-02-26 14:00，那麼我們需要将當天的13:55，13:56，13:57，13:58，13:59這最近5分鐘的資料處理到一起，并加載到Hive的最近5分鐘的一個分區裡面去。那麼，我們需要生成這樣一個邏輯政策集合，用HH-mm作為key，與之最近的5個檔案作為value，進行資料預處理合并。具體實作代碼步驟如下：

  • 步驟一：擷取小時循環政策；
  • 步驟二：擷取分鐘循環政策；
  • 步驟三：判斷是否為5分鐘的倍數；
  • 步驟四：對分鐘級别小于10的數字做0補齊（比如9補齊後變成09）；
  • 步驟五：對小時級别小于10的數字做0補齊（比如1補齊後變成01）；
  • 步驟六：生成時間範圍；
  • 步驟七：輸出結果。

其中，主要的邏輯是在生成時間範圍的過程中，根據小時和分鐘數的不同情況，生成不同的時間範圍，并輸出結果。在生成時間範圍時，需要注意前導0的處理，以及特殊情況（如小時為0、分鐘為0等）的處理。最後，将生成的時間範圍輸出即可。

根據上述步驟編寫對應的實作代碼，生成當天所有日期命名規則，預覽部分結果如下：

Kafka實時資料即席查詢應用與實踐

需要注意的是，如果發生了第二天00:00，那麼我們需要用到前一天的00-00=>23-59,23-58,23-57,23-56,23-55這5個檔案中的資料來做預處理。

3）加載資料

在完成2.3.1和2.3.2裡面的内容後，接下來，我們可以使用Hive的load指令直接加載HDFS上預處理後的檔案，把資料加載到對應的Hive表中，具體實作指令如下：

-- 加載資料到Hive表
load data inpath '<hdfs_path_hfile>' overwrite into table xxx.table partition(day='2022-02-26',hour='14',min='05')

4）即席分析

之後，我們使用Hive SQL來對Kafka資料進行即席分析，示例SQL如下所示：

-- 查詢某5分鐘分區資料
select * from xxx.table where day='2022-02-26' and hour='14' and min='05'

Flink SQL與 Flink DataStream如何選擇

Flink SQL 和 Flink DataStream 都是 Flink 中用于處理資料的核心元件，我們可以根據自己實際的業務場景來選擇使用哪一種元件。

Flink SQL 是一種基于 SQL 語言的資料處理引擎，它可以将 SQL 查詢語句轉換為 Flink 的資料流處理程式。相比于 Flink DataStream，Flink SQL 更加易于使用和維護，同時具有更快的開發速度和更高的代碼複用性。Flink SQL 适用于需要快速開發和部署資料處理任務的場景，比如資料倉庫、實時報表、資料清洗等。

Flink DataStream API是Flink資料流處理标準API，SQL是Flink後期版本提供的新的資料處理操作接口。SQL的引入為提高了Flink使用的靈活性。可以認為Flink SQL是一種通過字元串來定義資料流處理邏輯的描述語言。

是以，在選擇 Flink SQL 和 Flink DataStream 時，需要根據具體的業務需求和資料處理任務的特點來進行選擇。如果需要快速開發和部署任務，可以選擇使用 Flink SQL；如果需要進行更為深入和定制化的資料處理操作，可以選擇使用 Flink DataStream。同時，也可以根據實際情況，結合使用 Flink SQL 和 Flink DataStream 來完成複雜的資料處理任務。

三、 總結

在實際應用中，Kafka實時資料即席查詢可以用于多種場景，如實時監控、實時報警、實時統計、實時分析等。具體應用和實踐中，需要注意以下幾點：

  • 資料品質：Kafka實時資料即席查詢需要保證資料品質，避免資料重複、丢失或錯誤等問題，需要進行資料品質監控和調優。
  • 系統複雜性：Kafka實時資料即席查詢需要涉及到多個系統群組件，包括Kafka、資料處理引擎（比如Flink）、查詢引擎（比如Hive）等，需要對系統進行配置和管理，增加了系統的複雜性。
  • 安全性：Kafka實時資料即席查詢需要加強資料安全性保障，避免資料洩露或資料篡改等安全問題，做好Hive的權限管控。
  • 性能優化：Kafka實時資料即席查詢需要對系統進行性能優化，包括優化資料處理引擎、查詢引擎等，提高系統的性能和效率。

>>>>參考資料

  • https://github.com/apache/flink
  • https://flink.apache.org/

作者丨vivo 網際網路搜尋團隊-Deng Jie

來源丨公衆号：vivo網際網路技術（ID：vivoVMIC）

dbaplus社群歡迎廣大技術人員投稿，投稿郵箱：[email protected]

關于我們

dbaplus社群是圍繞Database、BigData、AIOps的企業級專業社群。資深大咖、技術幹貨，每天精品原創文章推送，每周線上技術分享，每月線下技術沙龍，每季度Gdevops&DAMS行業大會。

關注公衆号【dbaplus社群】，擷取更多原創技術文章和精選工具下載下傳

實時資料 kafka
上一篇: Js如何删除數組中某一個元素？
下一篇: asp.net core mvc (.net5)自帶容器生命周期

繼續閱讀