创建FlinkServer作业写入数据至Hudi表

操作场景

本指南通过使用FlinkServer写FlinkSQL对接Hudi。FlinkSQL读写Hudi时，不支持定义TINYINT、SMALLINT和TIME类型。

Flink对Hudi表的COW表、MOR表类型读写支持详情见表1。

前提条件

• 集群已安装HDFS、Yarn、Flink和Hudi等服务。
• 包含Hudi服务的客户端已安装，例如安装路径为：/opt/client。
• Flink要求1.12.2及以后版本，Hudi要求0.9.0及以后版本。
• 参考创建FlinkServer权限角色创建一个具有FlinkServer管理员权限的用户用于访问Flink WebUI，如：flink_admin。并且用户需要添加hadoop、hive、kafkaadmin用户组，以及Manager_administrator角色。

创建作业步骤

1. 使用flink_admin登录Manager，选择“集群 > 服务 > Flink”，在“Flink WebUI”右侧，单击链接，访问Flink的WebUI。
2. 参考如何创建FlinkServer作业，新建Flink SQL流作业，在作业开发界面进行如下作业配置。然后输入SQL，执行SQL校验通过后，启动作业。如下SQL示例将作为3个作业分别添加，依次运行。

   需勾选“基础参数”中的“开启CheckPoint”，“时间间隔（ms）”可设置为“60000”，“模式”可使用默认值。

   

   • 由于FlinkSQL作业在触发CheckPoint时才会往Hudi表中写数据，所以需要在Flink WebUI界面中开启CheckPoint。CheckPoint间隔根据业务需要调整，建议间隔调大。
   • 如果CheckPoint间隔太短，数据来不及刷新会导致作业异常；建议CheckPoint间隔为分钟级。
   • FlinkSQL作业写MOR表时需要做异步compaction，控制compaction间隔的参数，见Hudi官网：https://hudi.apache.org/docs/configurations.html
   • MRS 3.2.1及以后版本默认Hudi写表是Flink状态索引，如果需要使用bucket索引需要在Hudi写表中添加参数：

     'index.type'='BUCKET',
     'hoodie.bucket.index.num.buckets'='Hudi表中每个分区划分桶的个数',
     'hoodie.bucket.index.hash.field'='recordkey.field'

     • hoodie.bucket.index.num.buckets：Hudi表中每个分区划分桶的个数，每个分区内的数据通过Hash方式放入每个桶内。建表或第一次写入数据时设置后不能修改，否则更新数据会存在异常。
     • hoodie.bucket.index.hash.field：进行分桶时计算Hash值的字段，必须为主键的子集，默认为Hudi表的主键。该参数不填则默认为recordkey.field。
   • MRS 3.2.1及以后版本对于同一张Hudi表，可以被Flink、Spark引擎的bucket索引交叉混写。
   1. 作业1：FlinkSQL流式写入MOR表。

      CREATE TABLE stream_mor(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) PARTITIONED BY (`p`) WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://hacluster/tmp/hudi/stream_mor',
      'table.type' = 'MERGE_ON_READ',
      'hoodie.datasource.write.recordkey.field' = 'uuid',
      'write.precombine.field' = 'ts',
      'write.tasks' = '4'
      );
      
      CREATE TABLE kafka(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) WITH (
      'connector' = 'kafka',
      'topic' = 'writehudi',
      'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
      'properties.group.id' = 'testGroup1',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json',
      'properties.sasl.kerberos.service.name' = 'kafka',--普通模式集群不需要该参数，同时删除上一行的逗号
      'properties.security.protocol' = 'SASL_PLAINTEXT',--普通模式集群不需要该参数
      'properties.kerberos.domain.name' = 'hadoop.系统域名'--普通模式集群不需要该参数
      );
      
      insert into
      stream_mor
      select
      *
      from
      kafka;

   2. 作业2：FlinkSQL流式写入COW表

      CREATE TABLE stream_write_cow(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) PARTITIONED BY (`p`) WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://hacluster/tmp/hudi/stream_cow',
      'hoodie.datasource.write.recordkey.field' = 'uuid',
      'write.precombine.field' = 'ts',
      'write.tasks' = '4'
      );
      
      CREATE TABLE kafka(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) WITH (
      'connector' = 'kafka',
      'topic' = 'writehudi',
      'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
      'properties.group.id' = 'testGroup1',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json',
      'properties.sasl.kerberos.service.name' = 'kafka',--普通模式集群不需要该参数，同时删除上一行的逗号
      'properties.security.protocol' = 'SASL_PLAINTEXT',--普通模式集群不需要该参数
      'properties.kerberos.domain.name' = 'hadoop.系统域名'--普通模式集群不需要该参数
      );
      
      insert into
      stream_write_cow
      select
      *
      from
      kafka;

   3. 作业3：FlinkSQL流式读取MOR和COW表并合并数据输出到Kafka。注意作业3需要等待作业1和作业2均启动后，状态显示为“运行中”后再执行SQL校验和启动作业（否则SQL校验可能提示错误，找不到Hudi表目录）。

      CREATE TABLE stream_mor(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) PARTITIONED BY (`p`) WITH (
      'connector' = 'hudi',  
      'path' = 'hdfs://hacluster/tmp/hudi/stream_mor',  
      'table.type' = 'MERGE_ON_READ',
      'hoodie.datasource.write.recordkey.field' = 'uuid',
      'write.precombine.field' = 'ts',
      'read.tasks' = '4',
      'read.streaming.enabled' = 'true',
      'read.streaming.check-interval' = '5',
      'read.streaming.start-commit' = 'earliest'
      );
      CREATE TABLE stream_write_cow(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) PARTITIONED BY (`p`) WITH (
      'connector' = 'hudi',
      'path' = 'hdfs://hacluster/tmp/hudi/stream_cow',
      'hoodie.datasource.write.recordkey.field' = 'uuid',
      'write.precombine.field' = 'ts',
      'read.tasks' = '4',
      'read.streaming.enabled' = 'true',
      'read.streaming.check-interval' = '5',
      'read.streaming.start-commit' = 'earliest'
      );
      
      CREATE TABLE kafka(
      uuid VARCHAR(20),
      name VARCHAR(10),
      age INT,
      ts INT,
      `p` VARCHAR(20)
      ) WITH (
      'connector' = 'kafka',
      'topic' = 'readhudi',
      'properties.bootstrap.servers' = 'Kafka的Broker实例业务IP:Kafka端口号',
      'properties.group.id' = 'testGroup1',
      'scan.startup.mode' = 'latest-offset',
      'format' = 'json',
      'properties.sasl.kerberos.service.name' = 'kafka',--普通模式集群不需要该参数，同时删除上一行的逗号
      'properties.security.protocol' = 'SASL_PLAINTEXT',--普通模式集群不需要该参数
      'properties.kerberos.domain.name' = 'hadoop.系统域名'--普通模式集群不需要该参数
      );
      
      insert into 
      kafka 
      select 
      * 
      from 
      stream_mor union all select * from stream_write_cow;

   

   • Kafka端口号：
     • 集群的“认证模式”为“安全模式”时为“sasl.port”的值，默认为“21007”。
     • 集群的“认证模式”为“普通模式”时为“port”的值，默认为“9092”。如果配置端口号为9092，则需要配置“allow.everyone.if.no.acl.found”参数为true，具体操作如下：

       登录FusionInsight Manager系统，选择“集群 > 服务 > Kafka > 配置 > 全部配置”，搜索“allow.everyone.if.no.acl.found”配置，修改参数值为true，保存配置即可。

   • 系统域名：可登录FusionInsight Manager，选择“系统 > 权限 > 域和互信”，查看“本端域”参数，即为当前系统域名。

3. 参考管理Kafka Topic中的消息，向kafka中写入数据。

   sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic 主题名称 --producer.config 客户端目录/Kafka/kafka/config/producer.properties

   例如本示例使用主题名称为“writehudi”：

   sh kafka-console-producer.sh --broker-list Kafka角色实例所在节点的IP地址:Kafka端口号 --topic writehudi --producer.config /opt/client/Kafka/kafka/config/producer.properties

   输入消息内容：

   {"uuid": "1","name":"a01","age":10,"ts":10,"p":"1"}
   {"uuid": "2","name":"a02","age":20,"ts":20,"p":"2"}

   输入完成后按回车发送消息。

4. 消费kafka topic的数据，读取Flink流读Hudi表的结果。

   sh kafka-console-consumer.sh --bootstrap-server Kafka的Broker实例所在节点的IP地址:Kafka端口号 --topic 主题名称 --consumer.config 客户端目录/Kafka/kafka/config/consumer.properties --from-beginning

   例如本示例使用主题名称为“readhudi”：

   sh kafka-console-consumer.sh --bootstrap-server Kafka的Broker实例所在节点的IP地址:Kafka端口号 --topic readhudi --consumer.config /opt/client/Kafka/kafka/config/consumer.properties --from-beginning

   读取结果如下（顺序不固定）：

   {"uuid": "1","name":"a01","age":10,"ts":10,"p":"1"}
   {"uuid": "2","name":"a02","age":20,"ts":20,"p":"2"}
   {"uuid": "1","name":"a01","age":10,"ts":10,"p":"1"}
   {"uuid": "2","name":"a02","age":20,"ts":20,"p":"2"}

FlinkSQL Lookup Join Hudi使用须知

适用于MRS 3.5.0及以后版本。

• 使用lookup.join.cache.ttl参数来控制维表数据的加载周期，默认值为60min。
• Hudi维表数据会被加载到Flink TaskManager Heap中，所以不推荐大于10万行记录的Hudi表作为维表。
• 维表的新增、更新数据需要等到下一次加载周期后，才能被加载进来参与计算。

CREATE TABLE hudimor(
  uuid VARCHAR(20),
  name VARCHAR(10),
  age INT,
  ts INT,
  `p` VARCHAR(20),
  PRIMARY KEY (uuid) NOT ENFORCED
) PARTITIONED BY (`p`) WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://hacluster/tmp/hudimor',
  'table.type' = 'MERGE_ON_READ',
  'hoodie.datasource.write.recordkey.field' = 'uuid',
  'write.precombine.field' = 'ts',
  'lookup.join.cache.ttl' = '60min'
);
CREATE TABLE datagen(uuid varchar(20), proctime as PROCTIME()) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '1'
);
CREATE TABLE blackhole (
  uuid VARCHAR(20),
  name VARCHAR(10),
  age INT,
  ts INT,
  `p` VARCHAR(20)
) WITH ('connector' = 'blackhole');
insert into
  blackhole
select
  t1.uuid as uuid,
  t2.name as name,
  t2.age as age,
  t2.ts as ts,
  t2.p as p
FROM
  datagen AS t1
  left JOIN hudimor FOR SYSTEM_TIME AS OF t1.proctime AS t2 ON t1.uuid = t2.uuid;

WITH主要参数说明

Flink On Hudi同步元数据到Hive

启动此特性后，Flink写数据至Hudi表将自动在Hive上创建出Hudi表并同步添加分区，然后供SparkSQL、Hive等服务读取Hudi表数据。

如下是支持的两种同步元数据方式，后续操作步骤以JDBC方式为示例：