flink13.2CDC-iceberg结合

1.根据网上文章，客户端使用flink1.11.4+iceberg-flink-runtime-0.11.1.jar （iceberg0.12新出，使用即报错）版本可正常操作。flink1.12.5 与flink1.13.2  都尝试过，皆报错（可能由于本人原因，尚未排查出错误原因）。

 2.代码端 flink cdc使用1.13.2 或者1.12.5 版本皆可，但pom配置某些包需降成1.11.1 不然会报缺包等错误。本次操作为使用flinkcdc（flink-connector-mysql-cdc 2.0.0 jar）与flink 13.2 结合，实时监控mysqlbinlog日志（需提前开启binlog日志功能，此处可自行百度，修改mysql配置文件即可），入库iceberg。此代码很多版本问题，版本不一致会出现各种错误，下面会本人使用pom文件和代码，亲测有效


3.pom文件

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

        xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0

    <groupId>org.example

    <artifactId>flink-mysql-iceberg

    <packaging>pom

    <version>1.0-SNAPSHOT

        <module>spark-hudi

        <java.version>1.8

        <maven.compiler.source>1.8

        <maven.compiler.target>1.8

        <spark.version>3.0.0

        <flink.version>1.12.5

        <scala.version>2.12

        <hadoop.version>3.1.3

            <groupId>org.slf4j

            <artifactId>slf4j-simple

            <version>1.7.25

            <groupId>org.projectlombok

            <artifactId>lombok

            <version>1.18.2

            <groupId>org.apache.flink

            <artifactId>flink-streaming-java_2.11

            <version>${flink.version}

            <groupId>org.apache.flink

            <artifactId>flink-json

            <version>${flink.version}

            <groupId>org.apache.flink

            <artifactId>flink-table-common

            <version>1.11.1

            <groupId>org.apache.flink

            <artifactId>flink-table-api-java

            <version>1.11.1

        <!-- 最新加入 -->

            <groupId>org.apache.flink

            <artifactId>flink-table-api-java-bridge_2.11

            <version>1.11.1

            <groupId>org.apache.flink

            <artifactId>flink-table-runtime-blink_2.11

            <version>1.11.1

            <groupId>com.alibaba

            <artifactId>fastjson

            <version>1.2.67

            <groupId>org.apache.flink

            <artifactId>flink-connector-jdbc_${scala.version}

            <version>${flink.version}

            <groupId>org.apache.flink

            <artifactId>flink-clients_2.11

            <version>${flink.version}

        <!--后面追加的jar包-->

            <groupId>com.ververica

            <artifactId>flink-connector-mysql-cdc

            <version>2.0.0

            <groupId>org.apache.hadoop

            <artifactId>hadoop-common

            <version>2.7.7

            <scope>compile

            <groupId>org.apache.hadoop

            <artifactId>hadoop-hdfs

            <version>2.7.7

            <groupId>org.apache.iceberg

            <artifactId>iceberg-flink-runtime

            <version>0.11.0

            <groupId>org.apache.iceberg

            <artifactId>iceberg-flink

            <version>0.11.0

            <scope>provided

                    <artifactId>slf4j-api

                    <groupId>org.slf4j

            <groupId>org.apache.flink

            <artifactId>flink-table-planner-blink_2.11

            <version>${flink.version}

            <scope>compile

                    <artifactId>slf4j-api

                    <groupId>org.slf4j

            <groupId>org.apache.flink

            <artifactId>flink-table-planner_2.11

            <version>${flink.version}

            <scope>compile

                    <artifactId>slf4j-api

                    <groupId>org.slf4j

            <groupId>org.apache.flink

            <artifactId>flink-table

            <version>${flink.version}

            <type>pom

            <scope>provided

            <groupId>mysql

            <artifactId>mysql-connector-java

            <version>8.0.16

            <scope>${scope}

            <groupId>org.apache.hadoop

            <artifactId>hadoop-mapreduce-client-core

            <version>3.1.3

        <!--新加入spark jar包-->

                <groupId>org.apache.maven.plugins

                <artifactId>maven-assembly-plugin

                <version>3.0.0

                        <descriptorRef>jar-with-dependencies

                        <id>make-assembly

                        <phase>package

                            <goal>single

</project>

以上4个类需降低到1.11.1版本

5.代码：

import com.ververica.cdc.connectors.mysql.MySqlSource;

import com.ververica.cdc.debezium.DebeziumDeserializationSchema;

import com.ververica.cdc.debezium.table.RowDataDebeziumDeserializeSchema;

import org.apache.flink.api.common.typeinfo.TypeInformation;

import org.apache.flink.streaming.api.CheckpointingMode;

import org.apache.flink.streaming.api.datastream.DataStream;

import org.apache.flink.streaming.api.datastream.DataStreamSource;

import org.apache.flink.streaming.api.environment.CheckpointConfig;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import org.apache.flink.table.api.DataTypes;

import org.apache.flink.table.api.TableColumn;

import org.apache.flink.table.api.TableSchema;

import org.apache.flink.table.data.RowData;

import org.apache.flink.table.runtime.types.TypeInfoDataTypeConverter;

import org.apache.flink.table.types.DataType;

import org.apache.flink.table.types.inference.TypeTransformations;

import org.apache.flink.table.types.logical.RowType;

import org.apache.flink.table.types.utils.DataTypeUtils;

import org.apache.hadoop.conf.Configuration;

import org.apache.iceberg.*;

import org.apache.iceberg.catalog.Catalog;

import org.apache.iceberg.catalog.Namespace;

import org.apache.iceberg.catalog.TableIdentifier;

import org.apache.iceberg.flink.CatalogLoader;

import org.apache.iceberg.flink.TableLoader;

import org.apache.iceberg.flink.sink.FlinkSink;

import org.apache.iceberg.types.Types;

import java.time.ZoneId;

import java.util.Arrays;

import java.util.HashMap;

import java.util.Map;

public class rowData2iceberg {

private static final StringHADOOP_CATALOG ="iceberg_hadoop_catalog";

    //定义iceberg schema

    private static final SchemaSCHEMA =

new Schema(

Types.NestedField.optional(1, "id", Types.IntegerType.get()),

                    Types.NestedField.optional(2, "image", Types.BinaryType.get())

//Types.NestedField.optional(3, "age", Types.IntegerType.get()),

//Types.NestedField.optional(4, "address", Types.StringType.get()),

//Types.NestedField.optional(5, "score1", Types.IntegerType.get()));

// Types.NestedField.optional(6, "school", Types.StringType.get());

//  Types.NestedField.optional(7, "class", Types.StringType.get())

                    );

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

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        System.setProperty("HADOOP_USER_NAME","daizhihao");

        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        checkpointConfig.setCheckpointInterval(60 *1000L);

        checkpointConfig.setMinPauseBetweenCheckpoints(60 *1000L);

        checkpointConfig.setTolerableCheckpointFailureNumber(10);

        checkpointConfig.setCheckpointTimeout(12 *1000L);

        checkpointConfig.enableExternalizedCheckpoints(

CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //定义mysql的监控字段

        TableSchema schema =

TableSchema.builder()

.add(TableColumn.of("id", DataTypes.INT()))

.add(TableColumn.of("image", DataTypes.BINARY(2)))

//                        .add(TableColumn.of("age", DataTypes.INT()))

//                        .add(TableColumn.of("address", DataTypes.STRING()))

//                        .add(TableColumn.of("score1", DataTypes.INT()))

//.add(TableColumn.of("school", DataTypes.STRING()))

//.add(TableColumn.of("class", DataTypes.STRING()))

                        .build();

        RowType rowType = (RowType) schema.toRowDataType().getLogicalType();

        DebeziumDeserializationSchema deserialer =

new RowDataDebeziumDeserializeSchema(

rowType,

                        createTypeInfo(schema.toRowDataType()),

                        (rowData, rowKind) -> {},

                        ZoneId.of("Asia/Shanghai"));

        SourceFunction sourceFunction = MySqlSource.builder()

.hostname("localhost")

.serverTimeZone("UTC")

.port(3306)

.databaseList("demo")// monitor all tables under inventory database

                .tableList("demo.picture")

.username("root")

.password("root")

//                .hostname("Tgz3-eip-gzjfzxgputest1")

//                .port(23308)

//                .databaseList("eip_cs")

                .deserializer(deserialer)

// .deserializer(new StringDebeziumDeserializationSchema()) // converts SourceRecord to String

                .build();

        DataStreamSource src = env.addSource(sourceFunction);

//设置Checkpoint的模式：精准一次

        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

        icebergSink_hadoop(src);

        env

//.addSource(sourceFunction)

//.print()

                .setParallelism(1); // use parallelism 1 for sink to keep message ordering

            env.execute();

    }

private static TypeInformationcreateTypeInfo(DataType producedDataType) {

final DataType internalDataType =

DataTypeUtils.transform(producedDataType, TypeTransformations.TO_INTERNAL_CLASS);

        return (TypeInformation)

TypeInfoDataTypeConverter.fromDataTypeToTypeInfo(internalDataType);

    }

private static void icebergSink_hadoop(DataStream src) {

Map properties =new HashMap<>();

        properties.put("type", "iceberg");

        properties.put("catalog-type", "hadoop");

        properties.put("property-version", "1");

        properties.put("warehouse", "hdfs://192.168.163.101:9000/user/hive/warehouse");

        CatalogLoader catalogLoader =

CatalogLoader.hadoop(HADOOP_CATALOG, new Configuration(), properties);

        icebergSink(src, catalogLoader);

    }

private static void icebergSink(DataStream input,  CatalogLoader loader) {

Catalog catalog = loader.loadCatalog();

        //iceberg 命名

        TableIdentifier identifier =

TableIdentifier.of(Namespace.of("iceberg_db"), "image5");

        Table table;

        if (catalog.tableExists(identifier)) {

table = catalog.loadTable(identifier);

        }else {

table =

catalog.buildTable(identifier, SCHEMA)

.withPartitionSpec(PartitionSpec.unpartitioned())

.create();

        }

// need to upgrade version to 2,otherwise 'java.lang.IllegalArgumentException: Cannot write

// delete files in a v1 table'

        TableOperations operations = ((BaseTable) table).operations();

        TableMetadata metadata = operations.current();

        operations.commit(metadata, metadata.upgradeToFormatVersion(2));

        TableLoader tableLoader = TableLoader.fromCatalog(loader, identifier);

        FlinkSink.forRowData(input)

.table(table)

.tableLoader(tableLoader)

.equalityFieldColumns(Arrays.asList("id"))

.writeParallelism(1)

.build();

    }

}

5.1代码主要修改点：

1.


该段代码为iceberg建表字段（顺序和mysql的一模一样，名称可以不一致，因为入库时，iceberg不会查看字段名称，只会按照顺序入库）

2.

该段代码是监控mysql字段，名称类型需一模一样

3.

数据库信息

4.

该处为iceberg配置信息

5.

iceberg_db 为数据库，image5为表名

6.flink11.1客户端查看入库信息。

6.1启动flink 集群 ：./start-cluster.sh。同时需自行启动hdfs集群。


6.2 查看当前进程，若出现StandaloneSessionClusterEntrypoint，TaskManagerRunner即代表成功。


6.3 进去客户端后，创建hadoop_catalog（注意：本人当使用hive_catalog时，插入数据时会报错。尚不清楚什么原因）

CREATE CATALOG hadoop_catalog WITH (

  'type'='iceberg',

  'catalog-type'='hadoop',

  'warehouse'='hdfs://192.168.163.101:9000/user/hive/warehouse',

  'property-version'='1'

);


6.4 使用hadoop_catalog及使用iceberg_db数据库（若不存在，个人创建一个 create database iceberg_db）


6.5 查看表数据，若发现表中有数据，及监控日志成功。

select * from image5.


本人参考以上链接，综合个人版本修改，得出以上结果

https://www.136.la/jingpin/show-126501.html

最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 205,236评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,867评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,715评论 0赞 340
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,899评论 1赞 278
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,895评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,733评论 1赞 283
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,085评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,722评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,025评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,696评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,816评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,447评论 4赞 322
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,057评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,009评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,254评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,204评论 2赞 352
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,561评论 2赞 343

推荐阅读更多精彩内容