flink-sql-connector-kafka-1.15.1.jar直接在maven仓库下

flink读取mysql binlog写入kafka

• 创建mysql表

CREATE TABLE student_binlog (
 id INT NOT NULL,
 name STRING,
 age INT,
 class STRING,
 PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
 'connector' = 'mysql-cdc',
 'hostname' = 'mysqlserver',
 'port' = '3308',
 'username' = 'root',
 'password' = '123456',
 'database-name' = 'test',
 'table-name' = 'student'
);

• 创建kafka表

create table student_binlog_sink_kafka(
 id INT NOT NULL,
 name STRING,
 age INT,
 class STRING,
 PRIMARY KEY (`id`) NOT ENFORCED
) with (
  'connector'='upsert-kafka',
  'topic'='data_test',
  'properties.bootstrap.servers' = 'kafka1:9092',
  'properties.group.id' = 'testGroup',
  'key.format'='json',
  'value.format'='json'
);

• 将mysql binlog日志写入kafka

insert into student_binlog_sink_kafka select * from student_binlog;

查看Flink的Web UI，可以看到刚才提交的job

开启tableau方式查询表

set 'sql-client.execution.result-mode' = 'tableau';select * from student_binlog_sink_kafka;

往mysql的student表插入和更新数据测试下

INSERT INTO student VALUES(1,'张三',16,'高一3班');
COMMIT;
INSERT INTO student VALUES(2,'李四',18,'高三3班');
COMMIT;
UPDATE student SET NAME='李四四' WHERE id = 2;
COMMIT;

flink读取kafka数据并写入hudi数据湖

• 创建Kafka源表

CREATE TABLE student_binlog_source_kafka (
 id INT NOT NULL,
 name STRING,
 age INT,
 class STRING
)
WITH(
    'connector' = 'kafka',
    'topic'='data_test',
    'properties.bootstrap.servers' = 'kafka1:9092',
    'properties.group.id' = 'testGroup',
    'scan.startup.mode' = 'earliest-offset',
    'format' = 'json'
);

• 创建hudi目标表

CREATE TABLE student_binlog_sink_hudi (
 id INT NOT NULL,
 name STRING,
 age INT,
 class STRING,
 PRIMARY KEY (`id`) NOT ENFORCED
)
PARTITIONED BY (`class`)
WITH (
  'connector' = 'hudi',
  'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/student_binlog_sink_hudi',
  'table.type' = 'MERGE_ON_READ',
  'write.option' = 'insert',
  'write.precombine.field' = 'class'
);

• 将kafka数据写入hudi表

insert into student_binlog_sink_hudi select * from student_binlog_source_kafka;

mysql中student表新增加2条数据

INSERT INTO student VALUES(3,'韩梅梅',16,'高二2班');
INSERT INTO student VALUES(4,'李雷',16,'高二2班');
COMMIT;

查看HDFS中已经有相应的分区和数据了

调优

Memory

Parallelism

Compaction

只适用于online compaction

集成Hive

hudi源表对应一份hdfs数据，可以通过spark，flink 组件或者hudi客户端将hudi表的数据映射为hive外部表，基于该外部表， hive可以方便的进行实时视图，读优化视图以及增量视图的查询。

集成步骤

这里以hive3.1.3(关于hive可以详细查看前面的文章)、 hudi 0.12.1为例， 其他版本类似

将hudi-hadoop-mr-bundle-0.9.0xxx.jar , hudi-hive-sync-bundle-0.9.0xx.jar 放到hiveserver 节点的lib目录下

cd /home/commons/apache-hive-3.1.3-bin
cp -rf /home/commons/hudi-release-0.12.1/packaging/hudi-hadoop-mr-bundle/target/hudi-hadoop-mr-bundle-0.12.1.jar lib/
cp -rf /home/commons/hudi-release-0.12.1/packaging/hudi-hive-sync-bundle/target/hudi-hive-sync-bundle-0.12.1.jar lib/

按照需求选择合适的方式并重启hive

nohup hive --service metastore &
nohup hive --service hiveserver2 &

连接jdbc hive2测试,显示所有数据库

Flink同步Hive

Flink hive sync 现在支持两种 hive sync mode, 分别是 hms 和 jdbc 模式。 其中 hms 只需要配置 metastore uris；而 jdbc 模式需要同时配置 jdbc 属性 和 metastore uris，具体配置示例如下

CREATE TABLE t7(
  id int,
  num int,
  ts int,
  primary key (id) not enforced
)
PARTITIONED BY (num)
with(
  'connector'='hudi',
  'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t7',
  'table.type'='COPY_ON_WRITE', 
  'hive_sync.enable'='true', 
  'hive_sync.table'='h7', 
  'hive_sync.db'='default', 
  'hive_sync.mode' = 'hms',
  'hive_sync.metastore.uris' = 'thrift://hadoop2:9083'
);
insert into t7 values(1,1,1);

Hive Catalog

Flink官网的找到对应文档版本找到connector-hive，下载flink-sql-connector-hive-3.1.2_2.12-1.15.1.jar,上传到flink的lib目录下，建表示例

CREATE CATALOG hive_catalog WITH (
    'type' = 'hive',
    'default-database' = 'default',
    'hive-conf-dir' = '/home/commons/apache-hive-3.1.3-bin/conf/'
);

use catalog hive_catalog;
CREATE TABLE t8(
  id int,
  num int,
  ts int,
  primary key (id) not enforced
)
PARTITIONED BY (num)
with(
  'connector'='hudi',
  'path' = 'hdfs://hadoop1:9000/tmp/hudi_flink/t8',
  'table.type'='COPY_ON_WRITE', 
  'hive_sync.enable'='true', 
  'hive_sync.table'='h8', 
  'hive_sync.db'='default', 
  'hive_sync.mode' = 'hms',
  'hive_sync.metastore.uris' = 'thrift://hadoop2:9083'
);

本人博客网站IT小神 www.itxiaoshen.com