FusionInsight MRS Flink DataStream API读写Hudi实践

摘要：目前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。

1. import org.apache.flink.configuration.ConfigOption;
2. import org.apache.flink.configuration.Configuration;
3. import org.apache.flink.configuration.ReadableConfig;
4. import org.apache.flink.streaming.api.datastream.DataStream;
5. import org.apache.flink.streaming.api.datastream.DataStreamSink;
6. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7. import org.apache.flink.table.api.EnvironmentSettings;
8. import org.apache.flink.table.api.internal.TableEnvironmentImpl;
9. import org.apache.flink.table.catalog.Catalog;
10. import org.apache.flink.table.catalog.CatalogTable;
11. import org.apache.flink.table.catalog.ObjectIdentifier;
12. import org.apache.flink.table.catalog.ObjectPath;
13. import org.apache.flink.table.catalog.exceptions.TableNotExistException;
14. import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
15. import org.apache.flink.table.connector.source.DataStreamScanProvider;
16. import org.apache.flink.table.connector.source.ScanTableSource;
17. import org.apache.flink.table.data.RowData;
18. import org.apache.flink.table.factories.DynamicTableFactory;
19. import org.apache.flink.table.runtime.connector.sink.SinkRuntimeProviderContext;
20. import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
21. import org.apache.hudi.exception.HoodieException;
22. import org.apache.hudi.table.HoodieTableFactory;
24. import java.util.ArrayList;
25. import java.util.Arrays;
26. import java.util.HashMap;
27. import java.util.List;
28. import java.util.Map;
29. import java.util.stream.Collectors;
31. /**
32. *  A tool class to construct hoodie flink pipeline.
33. *
34. *  <p>How to use ?</p>
35. *  Method {@link #builder(String)} returns a pipeline builder. The builder
36. *  can then define the hudi table columns, primary keys and partitions.
37. *
38. *  <p>An example:</p>
39. *  <pre>
40. *    HoodiePipeline.Builder builder = HoodiePipeline.builder("myTable");
41. *    DataStreamSink<?> sinkStream = builder
42. *        .column("f0 int")
43. *        .column("f1 varchar(10)")
44. *        .column("f2 varchar(20)")
45. *        .pk("f0,f1")
46. *        .partition("f2")
47. *        .sink(input, false);
48. *  </pre>
49. */
50. public class HoodiePipeline {
52.   /**
53.    * Returns the builder for hoodie pipeline construction.
54.    */
55.   public static Builder builder(String tableName) {
56.     return new Builder(tableName);
57.   }
59.     /**
60.      * Builder for hudi source/sink pipeline construction.
61.      */
62.     public static class Builder {
63.       private final String tableName;
64.       private final List<String> columns;
65.       private final Map<String, String> options;
67.       private String pk;
68.       private List<String> partitions;
70.       private Builder(String tableName) {
71.         this.tableName = tableName;
72.         this.columns = new ArrayList<>();
73.         this.options = new HashMap<>();
74.         this.partitions = new ArrayList<>();
75.       }
77.       /**
78.        * Add a table column definition.
79.        *
80.        * @param column the column format should be in the form like 'f0 int'
81.        */
82.       public Builder column(String column) {
83.         this.columns.add(column);
84.         return this;
85.       }
87.       /**
88.        * Add primary keys.
89.        */
90.       public Builder pk(String... pks) {
91.         this.pk = String.join(",", pks);
92.         return this;
93.       }
95.       /**
96.        * Add partition fields.
97.        */
98.       public Builder partition(String... partitions) {
99.         this.partitions = new ArrayList<>(Arrays.asList(partitions));
100.         return this;
101.       }
103.       /**
104.        * Add a config option.
105.        */
106.       public Builder option(ConfigOption<?> option, Object val) {
107.         this.options.put(option.key(), val.toString());
108.         return this;
109.       }
111.       public Builder option(String key, Object val) {
112.         this.options.put(key, val.toString());
113.         return this;
114.       }
116.       public Builder options(Map<String, String> options) {
117.         this.options.putAll(options);
118.         return this;
119.       }
121.       public DataStreamSink<?> sink(DataStream<RowData> input, boolean bounded) {
122.         TableDescriptor tableDescriptor = getTableDescriptor();
123.         return HoodiePipeline.sink(input, tableDescriptor.getTableId(), tableDescriptor.getCatalogTable(), bounded);
124.       }
126.       public TableDescriptor getTableDescriptor() {
127.         EnvironmentSettings environmentSettings = EnvironmentSettings
128.             .newInstance()
129.             .build();
130.         TableEnvironmentImpl tableEnv = TableEnvironmentImpl.create(environmentSettings);
131.         String sql = getCreateHoodieTableDDL(this.tableName, this.columns, this.options, this.pk, this.partitions);
132.         tableEnv.executeSql(sql);
133.         String currentCatalog = tableEnv.getCurrentCatalog();
134.         CatalogTable catalogTable = null;
135.         String defaultDatabase = null;
136.         try {
137.             Catalog catalog = tableEnv.getCatalog(currentCatalog).get();
138.             defaultDatabase = catalog.getDefaultDatabase();
139.             catalogTable = (CatalogTable) catalog.getTable(new ObjectPath(defaultDatabase, this.tableName));
140.         } catch (TableNotExistException e) {
141.             throw new HoodieException("Create table " + this.tableName + " exception", e);
142.         }
143.         ObjectIdentifier tableId = ObjectIdentifier.of(currentCatalog, defaultDatabase, this.tableName);
144.         return new TableDescriptor(tableId, catalogTable);
145.       }
147.       public DataStream<RowData> source(StreamExecutionEnvironment execEnv) {
148.         TableDescriptor tableDescriptor = getTableDescriptor();
149.         return HoodiePipeline.source(execEnv, tableDescriptor.tableId, tableDescriptor.getCatalogTable());
150.       }
151.     }
153.     private static String getCreateHoodieTableDDL(
154.       String tableName,
155.       List<String> fields,
156.       Map<String, String> options,
157.       String pkField,
158.       List<String> partitionField) {
159.       StringBuilder builder = new StringBuilder();
160.       builder.append("create table ")
161.           .append(tableName)
162.           .append("(\n");
163.       for (String field : fields) {
164.         builder.append("  ")
165.               .append(field)
166.               .append(",\n");
167.       }
168.       builder.append("  PRIMARY KEY(")
169.           .append(pkField)
170.           .append(") NOT ENFORCED\n")
171.           .append(")\n");
172.       if (!partitionField.isEmpty()) {
173.         String partitons = partitionField
174.             .stream()
175.             .map(partitionName -> "`" + partitionName + "`")
176.             .collect(Collectors.joining(","));
177.         builder.append("PARTITIONED BY (")
178.             .append(partitons)
179.             .append(")\n");
180.       }
181.       builder.append("with ('connector' = 'hudi'");
182.       options.forEach((k, v) -> builder
183.           .append(",\n")
184.           .append("  '")
185.           .append(k)
186.           .append("' = '")
187.           .append(v)
188.           .append("'"));
189.       builder.append("\n)");
191.       System.out.println(builder.toString());
192.       return builder.toString();
193.     }
195.     /**
196.      * Returns the data stream sink with given catalog table.
197.      *
198.      * @param input        The input datastream
199.      * @param tablePath    The table path to the hoodie table in the catalog
200.      * @param catalogTable The hoodie catalog table
201.      * @param isBounded    A flag indicating whether the input data stream is bounded
202.      */
203.     private static DataStreamSink<?> sink(DataStream<RowData> input, ObjectIdentifier tablePath, CatalogTable catalogTable, boolean isBounded) {
204.       DefaultDynamicTableContext context = new DefaultDynamicTableContext(tablePath, catalogTable,
205.           Configuration.fromMap(catalogTable.getOptions()), Thread.currentThread().getContextClassLoader(), false);
206.       HoodieTableFactory hoodieTableFactory = new HoodieTableFactory();
207.       return ((DataStreamSinkProvider) hoodieTableFactory.createDynamicTableSink(context)
208.           .getSinkRuntimeProvider(new SinkRuntimeProviderContext(isBounded)))
209.           .consumeDataStream(input);
210.     }
212.     /**
213.      * Returns the data stream source with given catalog table.
214.      *
215.      * @param execEnv      The execution environment
216.      * @param tablePath    The table path to the hoodie table in the catalog
217.      * @param catalogTable The hoodie catalog table
218.      */
219.     private static DataStream<RowData> source(StreamExecutionEnvironment execEnv, ObjectIdentifier tablePath, CatalogTable catalogTable) {
220.       DefaultDynamicTableContext context = new DefaultDynamicTableContext(tablePath, catalogTable,
221.           Configuration.fromMap(catalogTable.getOptions()), Thread.currentThread().getContextClassLoader(), false);
222.       HoodieTableFactory hoodieTableFactory = new HoodieTableFactory();
223.       DataStreamScanProvider dataStreamScanProvider = (DataStreamScanProvider) ((ScanTableSource) hoodieTableFactory
224.           .createDynamicTableSource(context))
225.           .getScanRuntimeProvider(new ScanRuntimeProviderContext());
226.       return  dataStreamScanProvider.produceDataStream(execEnv);
227.     }
229.     /***
230.      *  A POJO that contains tableId and resolvedCatalogTable.
231.      */
232.     public static class TableDescriptor {
233.       private ObjectIdentifier tableId;
234.       private CatalogTable catalogTable;
236.       public TableDescriptor(ObjectIdentifier tableId, CatalogTable catalogTable) {
237.           this.tableId = tableId;
238.           this.catalogTable = catalogTable;
239.       }
241.       public ObjectIdentifier getTableId() {
242.           return tableId;
243.       }
245.       public CatalogTable getCatalogTable() {
246.             return catalogTable;
247.         }
248.     }
250.     private static class DefaultDynamicTableContext implements DynamicTableFactory.Context {
252.       private final ObjectIdentifier objectIdentifier;
253.       private final CatalogTable catalogTable;
254.       private final ReadableConfig configuration;
255.       private final ClassLoader classLoader;
256.       private final boolean isTemporary;
258.       DefaultDynamicTableContext(
259.         ObjectIdentifier objectIdentifier,
260.         CatalogTable catalogTable,
261.         ReadableConfig configuration,
262.         ClassLoader classLoader,
263.         boolean isTemporary) {
264.         this.objectIdentifier = objectIdentifier;
265.         this.catalogTable = catalogTable;
266.         this.configuration = configuration;
267.         this.classLoader = classLoader;
268.         this.isTemporary = isTemporary;
269.       }
271.       @Override
272.       public ObjectIdentifier getObjectIdentifier() {
273.         return objectIdentifier;
274.       }
276.       @Override
277.       public CatalogTable getCatalogTable() {
278.         return catalogTable;
279.       }
281.       @Override
282.       public ReadableConfig getConfiguration() {
283.         return configuration;
284.       }
286.       @Override
287.       public ClassLoader getClassLoader() {
288.         return classLoader;
289.       }
291.       @Override
292.       public boolean isTemporary() {
293.             return isTemporary;
294.         }
295.     }
296. }

复制代码

2.WriteIntoHudi.java 使用 DataStream API将数据写入Hudi。关键实现逻辑：
第一步：Demo中的数据源来自datagen connector Table。
第二步：使用toAppendStream将Table转化为Stream。
第三步：build hudi sink stream后写入Hudi。
在项目实践中也可以直接使用DataStream源写入Hudi。

1. import org.apache.flink.streaming.api.datastream.DataStream;
2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
3. import org.apache.flink.table.api.Table;
4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
5. import org.apache.flink.table.data.RowData;
6. import org.apache.hudi.common.model.HoodieTableType;
7. import org.apache.hudi.configuration.FlinkOptions;
9. import java.util.HashMap;
10. import java.util.Map;
12. public class WriteIntoHudi {
13.     public static void main(String[] args) throws Exception {
15.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
16.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
17.         env.getCheckpointConfig().setCheckpointInterval(10000);
19.         tableEnv.executeSql("CREATE TABLE datagen (\n"
20.             + "  uuid varchar(20),\n"
21.             + "  name varchar(10),\n"
22.             + "  age int,\n"
23.             + "  ts timestamp(3),\n"
24.             + "  p varchar(20)\n"
25.             + ") WITH (\n"
26.             + "  'connector' = 'datagen',\n"
27.             + "  'rows-per-second' = '5'\n"
28.             + ")");
30.         Table table = tableEnv.sqlQuery("SELECT * FROM datagen");
32.         DataStream<RowData> dataStream = tableEnv.toAppendStream(table, RowData.class);
33.         String targetTable = "hudiSinkTable";
35.         String basePath = "hdfs://hacluster/tmp/flinkHudi/hudiTable";
37.         Map<String, String> options = new HashMap<>();
38.         options.put(FlinkOptions.PATH.key(), basePath);
39.         options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
40.         options.put(FlinkOptions.PRECOMBINE_FIELD.key(), "ts");
41.         options.put(FlinkOptions.INDEX_BOOTSTRAP_ENABLED.key(), "true");
43.         HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
44.             .column("uuid VARCHAR(20)")
45.             .column("name VARCHAR(10)")
46.             .column("age INT")
47.             .column("ts TIMESTAMP(3)")
48.             .column("p VARCHAR(20)")
49.             .pk("uuid")
50.             .partition("p")
51.             .options(options);
53.         builder.sink(dataStream, false); // The second parameter indicating whether the input data stream is bounded
54.         env.execute("Api_Sink");
55.     }
56. }

复制代码

3.ReadFromHudi.java 使用 DataStream API读取Hudi数据。关键实现逻辑：
第一步：build hudi source stream读取hudi数据。
第二步：使用fromDataStream将stream转化为table。
第三步：将Hudi table的数据使用print connector打印输出。
在项目实践中也可以直接读取Hudi数据后写入sink DataStream。

1. import org.apache.flink.streaming.api.datastream.DataStream;
2. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
3. import org.apache.flink.table.api.Table;
4. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
5. import org.apache.flink.table.data.RowData;
6. import org.apache.hudi.common.model.HoodieTableType;
7. import org.apache.hudi.configuration.FlinkOptions;
9. import java.util.HashMap;
10. import java.util.Map;
12. public class ReadFromHudi {
13.     public static void main(String[] args) throws Exception {
14.         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
15.         String targetTable = "hudiSourceTable";
16.         String basePath = "hdfs://hacluster/tmp/flinkHudi/hudiTable";
18.         Map<String, String> options = new HashMap<>();
19.         options.put(FlinkOptions.PATH.key(), basePath);
20.         options.put(FlinkOptions.TABLE_TYPE.key(), HoodieTableType.MERGE_ON_READ.name());
21.         options.put(FlinkOptions.READ_AS_STREAMING.key(), "true"); // this option enable the streaming read
22.         options.put("read.streaming.start-commit", "20210316134557"); // specifies the start commit instant time
24.         HoodiePipeline.Builder builder = HoodiePipeline.builder(targetTable)
25.             .column("uuid VARCHAR(20)")
26.             .column("name VARCHAR(10)")
27.             .column("age INT")
28.             .column("ts TIMESTAMP(3)")
29.             .column("p VARCHAR(20)")
30.             .pk("uuid")
31.             .partition("p")
32.             .options(options);
34.         DataStream<RowData> rowDataDataStream = builder.source(env);
36.         StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
37.         Table table = tableEnv.fromDataStream(rowDataDataStream,"uuid, name, age, ts, p");
39.         tableEnv.registerTable("hudiSourceTable",table);
41.         tableEnv.executeSql("CREATE TABLE print("
42.             + "   uuid varchar(20),\n"
43.             + "   name varchar(10),\n"
44.             + "   age int,\n"
45.             + "   ts timestamp(3),\n"
46.             + "   p varchar(20)\n"
47.             + ") WITH (\n"
48.             + " 'connector' = 'print'\n"
49.             + ")");
51.         tableEnv.executeSql("insert into print select * from hudiSourceTable");
52.         env.execute("Api_Source");
53.     }
54. }

复制代码

4.在项目实践中如果有解析Kafka复杂Json的需求：
1）使用FlinkSQL: https://bbs.huaweicloud.com/forum/thread-153494-1-1.html
2）使用Flink DataStream MapFunction实现。
点击关注，第一时间了解华为云新鲜技术~

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！