FusionInsight MRS Flink DataStream API讀寫Hudi實踐

来源:https://www.cnblogs.com/huaweiyun/archive/2022/11/14/16888362.html
-Advertisement-
Play Games

摘要：目前Hudi只支持FlinkSQL進行數據讀寫，但是在實際項目開發中一些客戶存在使用Flink DataStream API讀寫Hudi的訴求。 本文分享自華為雲社區《FusionInsight MRS Flink DataStream API讀寫Hudi實踐》，作者： yangxiao_mr ...


摘要：目前Hudi只支持FlinkSQL進行數據讀寫，但是在實際項目開發中一些客戶存在使用Flink DataStream API讀寫Hudi的訴求。

本文分享自華為雲社區《FusionInsight MRS Flink DataStream API讀寫Hudi實踐》，作者： yangxiao_mrs 。

目前Hudi只支持FlinkSQL進行數據讀寫，但是在實際項目開發中一些客戶存在使用Flink DataStream API讀寫Hudi的訴求。

該實踐包含三部分內容：

1）HoodiePipeline.java ，該類將Hudi內核讀寫介面進行封裝，提供Hudi DataStream API。

2）WriteIntoHudi.java ，該類使用 DataStream API將數據寫入Hudi。

3）ReadFromHudi.java ，該類使用 DataStream API讀取Hudi數據。

1.HoodiePipeline.java 將Hudi內核讀寫介面進行封裝，提供Hudi DataStream API。關鍵實現邏輯：

第一步：將原來Hudi流表的列名、主鍵、分區鍵set後，通過StringBuilder拼接成create table SQL。

第二步：將該hudi流表註冊到catalog中。

第三步：將DynamicTable轉換為DataStreamProvider後，進行數據produce或者consume。

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.internal.TableEnvironmentImpl;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.ObjectPath;
import org.apache.flink.table.catalog.exceptions.TableNotExistException;
import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
import org.apache.flink.table.connector.source.DataStreamScanProvider;
import org.apache.flink.table.connector.source.ScanTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.table.HoodieTableFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 *  A tool class to construct hoodie flink pipeline.
 *
 *  <p>How to use ?</p>
 *  Method {@link #builder(String)} returns a pipeline builder. The builder
 *  can then define the hudi table columns, primary keys and partitions.
 *
 *  <p>An example:</p>
 *  <pre>
 *    HoodiePipeline.Builder builder = HoodiePipeline.builder("myTable");
 *    DataStreamSink<?> sinkStream = builder
 *        .column("f0 int")
 *        .column("f1 varchar(10)")
 *        .column("f2 varchar(20)")
 *        .pk("f0,f1")
 *        .partition("f2")
 *        .sink(input, false);
 *  </pre>
 */
public class HoodiePipeline {

  /**
   * Returns the builder for hoodie pipeline construction.
   */
  public static Builder builder(String tableName) {
    return new Builder(tableName);
  }

    /**
     * Builder for hudi source/sink pipeline construction.
     */
    public static class Builder {
      private final String tableName;
      private final List<String> columns;
      private final Map<String, String> options;

      private String pk;
      private List<String> partitions;

      private Builder(String tableName) {
        this.tableName = tableName;
        this.columns = new ArrayList<>();
        this.options = new HashMap<>();
        this.partitions = new ArrayList<>();
      }

      /**
       * Add a table column definition.
       *
       * @param column the column format should be in the form like 'f0 int'
       */
      public Builder column(String column) {
        this.columns.add(column);
        return this;
      }

      /**
       * Add primary keys.
       */
      public Builder pk(String... pks) {
        this.pk = String.join(",", pks);
        return this;
      }

      /**
       * Add partition fields.
       */
      public Builder partition(String... partitions) {
        this.partitions = new ArrayList<>(Arrays.asList(partitions));
        return this;
      }

      /**
       * Add a config option.
       */
      public Builder option(ConfigOption<?> option, Object val) {
        this.options.put(option.key(), val.toString());
        return this;
      }

      public Builder option(String key, Object val) {
        this.options.put(key, val.toString());
        return this;
      }

      public Builder options(Map<String, String> options) {
        this.options.putAll(options);
        return this;
      }

      public DataStreamSink<?> sink(DataStream<RowData> input, boolean bounded) {
        TableDescriptor tableDescriptor = getTableDescriptor();
        return HoodiePipeline.sink(input, tableDescriptor.getTableId(), tableDescriptor.getCatalogTable(), bounded);
      }

      public TableDescriptor getTableDescriptor() {
        EnvironmentSettings environmentSettings = EnvironmentSettings
            .newInstance()
            .build();
        TableEnvironmentImpl tableEnv = TableEnvironmentImpl.create(environmentSettings);
        String sql = getCreateHoodieTableDDL(this.tableName, this.columns, this.options, this.pk, this.partitions);
        tableEnv.executeSql(sql);
        String currentCatalog = tableEnv.getCurrentCatalog();
        CatalogTable catalogTable = null;
        String defaultDatabase = null;
        try {
            Catalog catalog = tableEnv.getCatalog(currentCatalog).get();
            defaultDatabase = catalog.getDefaultDatabase();
            catalogTable = (CatalogTable) catalog.getTable(new ObjectPath(defaultDatabase, this.tableName));
        } catch (TableNotExistException e) {
            throw new HoodieException("Create table " + this.tableName + " exception", e);
        }
        ObjectIdentifier tableId = ObjectIdentifier.of(currentCatalog, defaultDatabase, this.tableName);
        return new TableDescriptor(tableId, catalogTable);
      }

      public DataStream<RowData> source(StreamExecutionEnvironment execEnv) {
        TableDescriptor tableDescriptor = getTableDescriptor();
        return HoodiePipeline.source(execEnv, tableDescriptor.tableId, tableDescriptor.getCatalogTable());
      }
    }

    private static String getCreateHoodieTableDDL(
      String tableName,
      List<String> fields,
      Map<String, String> options,
      String pkField,
      List<String> partitionField) {
      StringBuilder builder = new StringBuilder();
      builder.append("create table ")
          .append(tableName)
          .append("(\n");
      for (String field : fields) {
        builder.append("  ")
              .append(field)
              .append(",\n");
      }
      builder.append("  PRIMARY KEY(")
          .append(pkField)
          .append(") NOT ENFORCED\n")
          .append(")\n");
      if (!partitionField.isEmpty()) {
        String partitons = partitionField
            .stream()
            .map(partitionName -> "`" + partitionName + "`")
            .collect(Collectors.joining(","));
        builder.append("PARTITIONED BY (")
            .append(partitons)
            .append(")\n");
      }
      builder.append("with ('connector' = 'hudi'");
      options.forEach((k, v) -> builder
          .append(",\n")
          .append("  '")
          .append(k)
          .append("' = '")
          .append(v)
          .append("'"));
      builder.append("\n)");

      System.out.println(builder.toString());
      return builder.toString();
    }

    /**
     * Returns the data stream sink with given catalog table.
     *
     * @param input        The input datastream
     * @param tablePath    The table path to the hoodie table in the catalog
     * @param catalogTable The hoodie catalog table
     * @param isBounded    A flag indicating whether the input data stream is bounded
     */
    private static DataStreamSink<?> sink(DataStream<RowData> input, ObjectIdentifier tablePath, CatalogTable catalogTable, boolean isBounded) {
      DefaultDynamicTableContext context = new DefaultDynamicTableContext(tablePath, catalogTable,
          Configuration.fromMap(catalogTable.getOptions()), Thread.currentThread().getContextClassLoader(), false);
      HoodieTableFactory hoodieTableFactory = new HoodieTableFactory();
      return ((DataStreamSinkProvider) hoodieTableFactory.createDynamicTableSink(context)
          .getSinkRuntimeProvider(new SinkRuntimeProviderContext(isBounded)))
          .consumeDataStream(input);
    }

    /**
     * Returns the data stream source with given catalog table.
     *
     * @param execEnv      The execution environment
     * @param tablePath    The table path to the hoodie table in the catalog
     * @param catalogTable The hoodie catalog table
     */
    private static DataStream<RowData> source(StreamExecutionEnvironment execEnv, ObjectIdentifier tablePath, CatalogTable catalogTable) {
      DefaultDynamicTableContext context = new DefaultDynamicTableContext(tablePath, catalogTable,
          Configuration.fromMap(catalogTable.getOptions()), Thread.currentThread().getContextClassLoader(), false);
      HoodieTableFactory hoodieTableFactory = new HoodieTableFactory();
      DataStreamScanProvider dataStreamScanProvider = (DataStreamScanProvider) ((ScanTableSource) hoodieTableFactory
          .createDynamicTableSource(context))
          .getScanRuntimeProvider(new ScanRuntimeProviderContext());
      return  dataStreamScanProvider.produceDataStream(execEnv);
    }

    /***
     *  A POJO that contains tableId and resolvedCatalogTable.
     */
    public static class TableDescriptor {
      private ObjectIdentifier tableId;
      private CatalogTable catalogTable;

      public TableDescriptor(ObjectIdentifier tableId, CatalogTable catalogTable) {
          this.tableId = tableId;
          this.catalogTable = catalogTable;
      }

      public ObjectIdentifier getTableId() {
          return tableId;
      }

      public CatalogTable getCatalogTable() {
            return catalogTable;
        }
    }

    private static class DefaultDynamicTableContext implements DynamicTableFactory.Context {

      private final ObjectIdentifier objectIdentifier;
      private final CatalogTable catalogTable;
      private final ReadableConfig configuration;
      private final ClassLoader classLoader;
      private final boolean isTemporary;

      DefaultDynamicTableContext(
        ObjectIdentifier objectIdentifier,
        CatalogTable catalogTable,
        ReadableConfig configuration,
        ClassLoader classLoader,
        boolean isTemporary) {
        this.objectIdentifier = objectIdentifier;
        this.catalogTable = catalogTable;
        this.configuration = configuration;
        this.classLoader = classLoader;
        this.isTemporary = isTemporary;
      }

      @Override
      public ObjectIdentifier getObjectIdentifier() {
        return objectIdentifier;
      }

      @Override
      public CatalogTable getCatalogTable() {
        return catalogTable;
      }

      @Override
      public ReadableConfig getConfiguration() {
        return configuration;
      }

      @Override
      public ClassLoader getClassLoader() {
        return classLoader;
      }

      @Override
      public boolean isTemporary() {
            return isTemporary;
        }
    }
}

2.WriteIntoHudi.java 使用 DataStream API將數據寫入Hudi。關鍵實現邏輯：

第一步：Demo中的數據源來自datagen connector Table。

第二步：使用toAppendStream將Table轉化為Stream。

第三步：build hudi sink stream後寫入Hudi。

在項目實踐中也可以直接使用DataStream源寫入Hudi。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.configuration.FlinkOptions;

import java.util.HashMap;
import java.util.Map;

public class WriteIntoHudi {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        env.getCheckpointConfig().setCheckpointInterval(10000);

        tableEnv.executeSql("CREATE TABLE datagen (\n"
            + "  uuid varchar(20),\n"
            + "  name varchar(10),\n"
            + "  age int,\n"
            + "  ts timestamp(3),\n"
            + "  p varchar(20)\n"
            + ") WITH (\n"
            + "  'connector' = 'datagen',\n"
            + "  'rows-per-second' = '5'\n"
            + ")");

        Table table = tableEnv.sqlQuery("SELECT * FROM datagen");

        DataStream<RowData> dataStream = tableEnv.toAppendStream(table, RowData.class);
        String targetTable = "hudiSinkTable";

        String basePath = "hdfs://hacluster/tmp/flinkHudi/hudiTable";

        Map<String, String> options = new HashMap<>();
        options.put(FlinkOptions.PATH.key(), basePath);
        options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
        options.put(FlinkOptions.PRECOMBINE_FIELD.key(), "ts");
        options.put(FlinkOptions.INDEX_BOOTSTRAP_ENABLED.key(), "true");

        HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
            .column("uuid VARCHAR(20)")
            .column("name VARCHAR(10)")
            .column("age INT")
            .column("ts TIMESTAMP(3)")
            .column("p VARCHAR(20)")
            .pk("uuid")
            .partition("p")
            .options(options);

        builder.sink(dataStream, false); // The second parameter indicating whether the input data stream is bounded
        env.execute("Api_Sink");
    }
}

3.ReadFromHudi.java 使用 DataStream API讀取Hudi數據。關鍵實現邏輯：

第一步：build hudi source stream讀取hudi數據。

第二步：使用fromDataStream將stream轉化為table。

第三步：將Hudi table的數據使用print connector列印輸出。

在項目實踐中也可以直接讀取Hudi數據後寫入sink DataStream。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.configuration.FlinkOptions;

import java.util.HashMap;
import java.util.Map;

public class ReadFromHudi {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        String targetTable = "hudiSourceTable";
        String basePath = "hdfs://hacluster/tmp/flinkHudi/hudiTable";

        Map<String, String> options = new HashMap<>();
        options.put(FlinkOptions.PATH.key(), basePath);
        options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
        options.put(FlinkOptions.READ_AS_STREAMING.key(), "true"); // this option enable the streaming read
        options.put("read.streaming.start-commit", "20210316134557"); // specifies the start commit instant time

        HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
            .column("uuid VARCHAR(20)")
            .column("name VARCHAR(10)")
            .column("age INT")
            .column("ts TIMESTAMP(3)")
            .column("p VARCHAR(20)")
            .pk("uuid")
            .partition("p")
            .options(options);

        DataStream<RowData> rowDataDataStream = builder.source(env);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        Table table = tableEnv.fromDataStream(rowDataDataStream,"uuid, name, age, ts, p");

        tableEnv.registerTable("hudiSourceTable",table);

        tableEnv.executeSql("CREATE TABLE print("
            + "   uuid varchar(20),\n"
            + "   name varchar(10),\n"
            + "   age int,\n"
            + "   ts timestamp(3),\n"
            + "   p varchar(20)\n"
            + ") WITH (\n"
            + " 'connector' = 'print'\n"
            + ")");

        tableEnv.executeSql("insert into print select * from hudiSourceTable");
        env.execute("Api_Source");
    }
}

4.在項目實踐中如果有解析Kafka複雜Json的需求：

1）使用FlinkSQL: https://bbs.huaweicloud.com/forum/thread-153494-1-1.html

2）使用Flink DataStream MapFunction實現。

點擊關註，第一時間瞭解華為雲新鮮技術~


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 第四章 linux字元設備驅動一
    前言 字元設備是Linux驅動中三大設備之一，字元（char）設備是個能夠像位元組流（類似文件）一樣被訪問的設備，由字元設備驅動程式來實現這種特性。字元設備驅動程式通常至少要實現open、close、read和write的系統調用。字元終端（/dev/console）和串口（/dev/ttyS0以及類 ...
  • 配置sshd_使用CA簽名證書登錄_更新sshd服務端的通訊密鑰
    一、Installing RabbitMQ-3.10.2 on CentOS 7.9 1 地址 https://www.rabbitmq.com https://github.com/rabbitmq/rabbitmq-server https://github.com/rabbitmq/rabbi ...
  • ElasticSearch 常見問題
    ElasticSearch 常見問題 丈夫有淚不輕彈，只因未到傷心處。 1、說說 es 的一些調優手段。 僅索引層面調優手段： 1.1、設計階段調優 （1）根據業務增量需求，採取基於日期模板創建索引，通過 roll over API 滾動索引； （2）使用別名進行索引管理； （3）每天凌晨定時對索引 ...
  • Oracle中新建數據表的兩種方法
    首發微信公眾號：SQL資料庫運維 原文鏈接：https://mp.weixin.qq.com/s?__biz=MzI1NTQyNzg3MQ==&mid=2247485212&idx=1&sn=450e9e94fa709b5eeff0de371c62072b&chksm=ea37536cdd40da7 ...
  • 五分鐘掌握CloudCanal的數據校驗與數據訂正
    簡述 CloudCanal除了提供最核心的數據遷移和同步能力以外，還提供數據校驗和數據訂正兩種非常實用的能力。這兩種功能為用戶保障數據遷移同步鏈路的數據質量提供了非常大的便利性。例如對端資料庫因為各種原因產生一些異常寫入導致的數據不一致或者丟失，用戶均可以使用CloudCanal提供的數據校驗和數據 ...
  • MySQL科學計數法展示解惑
    GreatSQL社區原創內容未經授權不得隨意使用，轉載請聯繫小編並註明來源。 GreatSQL是MySQL的國產分支版本，使用上與MySQL一致。 作者： 好好先生 一、問題引入 今天遇到一個很奇怪的問題，在MySQL客戶端輸入，用不同科學計數法表示的數值，展示效果卻截然不同： mysql> sel ...
  • SQL Server 時間算差值-年限
    項目中需要計算使用年限，按月份算。剛開始踩了坑，不足1年應該按1年算。記錄下~ 和當前時間比較，用DATEDIFF函數DateDiff(month,比較的時間,getdate())先算出月份，再除以12算年份 查看代碼 --月份差值 2.083333 select CONVERT(decimal,D ...
  • TDSQL攜手金蝶雲·蒼穹，發佈“國產資料庫聯合解決方案”
    11月11日，騰訊雲資料庫與金蝶雲 · 蒼穹發佈“國產資料庫聯合解決方案”，騰訊雲資料庫全面支持蒼穹平臺的技術與應用設計，通過一體化的“PaaS+SaaS”解決方案，一站式解決企業國產化難題，最高可支持億級賬戶規模量與日均億級交易處理。 目前，騰訊雲資料庫TDSQL和金蝶雲 · 蒼穹PaaS平臺已完 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類