体验项目介绍

本体验项目使用 GaussDB for Mysql 作为数据源， 使用 Flink和Flink CDC 实时采集数据源的数据并最终写入到GaussDB 数据库。

Flink CDC 是一个基于流的数据集成工具，旨在为用户提供一套功能更加全面的编程接口（API）。Flink CDC 深度集成并由 Apache Flink 驱动，提供以下核心功能：

✅ 端到端的数据集成框架
✅ 为数据集成的用户提供了易于构建作业的 API
✅ 支持在 Source 和 Sink 中处理多个表
✅ 整库同步
✅具备表结构变更自动同步的能力（Schema Evolution）。

Flink 是一个框架和分布式处理引擎，用于对无界和有界数据流进行有状态计算。Flink 被设计为在所有常见的集群环境中运行，能以内存级速度在任意规模(数据集)下执行计算任务。

GaussDB(for MySQL)是最新一代企业级高扩展高性能云原生数据库，完全兼容MySQL。

华为云数据库 GaussDB 是华为自主创新研发的分布式关系型数据库。该产品具备企业级复杂事务混合负载能力，同时支持分布式事务，同城跨AZ部署，数据0丢失，支持1000+的扩展能力，PB级海量存储。同时拥有云上高可用，高可靠，高安全，弹性伸缩，一键部署，快速备份恢复，监控告警等关键能力。

关于 openGauss 和 GaussDB 驱动的说明

获取GaussDB驱动包-GaussDB

获取GaussDB驱动包-GaussDB(DWS)

两者在使用上的区别：GaussDB 驱动包为了兼容 ORM 框架，在创建数据库连接时注册的驱动名称兼容“postgres”和“postgresql”，而 openGauss 在连接数据库时注册的驱动名称是“opengauss”。兼容性参考

实现代码示例

//代码讲解：
      
package com.open;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.PreparedStatement;
import java.sql.SQLException;


public class FlinkCDC_DataStream {

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

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> mysqlDS = env.fromSource(FlinkSourceUtil.getMySqlSource("db_name","table_name"), WatermarkStrategy.noWatermarks(), "mysql-source");

        SingleOutputStreamOperator<Tuple2<String,Integer>> mapDS = mysqlDS.map(new MapFunction<String, Tuple2<String,Integer>>() {
            @Override
            public Tuple2<String,Integer> map(String value) throws Exception {
                JSONObject mysqlJson = JSONObject.parseObject(value);
                System.out.println(value);
                JSONObject after = mysqlJson.getJSONObject("after");
                String name = after.getString("name"); // 字段1 name
                int age = after.getIntValue("age"); // 字段2 age
                return new Tuple2<String,Integer>(name,age);
            }
        });

        SinkFunction<Tuple2<String,Integer>> jdbcSink = JdbcSink.sink(

                "INSERT INTO flink_sink_t (name, age) VALUES (?,?)",
                new JdbcStatementBuilder<Tuple2<String,Integer>>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, Tuple2<String,Integer> s) throws SQLException {
                        preparedStatement.setString(1,s.f0);
                        preparedStatement.setInt(2,s.f1);
                    }
                }
    ,
                JdbcExecutionOptions.builder()
                        .withMaxRetries(3) // 重试次数
                        .withBatchSize(2) // 批次的大小：条数
                        .withBatchIntervalMs(300) // 批次的时间
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        //.withUrl("jdbc:postgresql://ip:8000/db?currentSchema=schema&useUnicode=true&characterEncoding=UTF-8") 
                        .withUrl("jdbc:gaussdb://ip:8000/db?currentSchema=schema&useUnicode=true&characterEncoding=UTF-8")
                        .withUsername("username")
                        .withPassword("password")
                        .withConnectionCheckTimeoutSeconds(60) // 重试的超时时间
                        .build()
        );

        mapDS.addSink(jdbcSink);

        //6.启动
        env.execute();

    }
}