devbox
从前慢现在也慢
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

手把手构建基于 GBase8s 的 Flink connector_org.apache.flink.connector.jdbc.internal.options.j

作者：从前慢现在也慢 | 2024-07-26 16:57:45

赞

踩

org.apache.flink.connector.jdbc.internal.options.jdbcoptions

​

简介

本篇文章，首先会向大家阐述什么是 Flink connector 和 CDC ， 然后会通过手把手的方式和大家一起构建一个简单的GBase8s的Flink connector,并完成实践项目，即通过Mysql CDC实时通过connector同步数据到GBase8s中。

什么是 Flink connector

Flink内置了一些基本数据源和接收器，这些数据源和接收器始终可用。该预定义的数据源包括文件、Mysql、RabbitMq、Kafka、ES等，同时也支持数据输出到文件、Mysql、RabbitMq、Kafka、ES等。

简单的说：flink连接器就是将某些数据源加载与数据输出做了封装（连接器），我们只要引入对应的连接器依赖，即可快速的完成对数据源的加载以及数据的输出。

什么是CDC（Change Data Capture）

首先什么是CDC ？它是Change Data Capture的缩写,即变更数据捕捉的简称，使用CDC我们可以从数据库中获取已提交的更改并将这些更改发送到下游，供下游使用。这些变更可以包括INSERT,DELETE,UPDATE等操作。

其主要的应用场景：

  • 异构数据库之间的数据同步或备份 / 建立数据分析计算平台

  • 微服务之间共享数据状态

  • 更新缓存 / CQRS 的 Query 视图更新

CDC 它是一个比较广义的概念，只要能捕获变更的数据，我们都可以称为 CDC 。业界主要有基于查询的 CDC 和基于日志的 CDC ，可以从下面表格对比他们功能和差异点。

基于查询的 CDC基于日志的 CDC
概念每次捕获变更发起 Select 查询进行全表扫描，过滤出查询之间变更的数据读取数据存储系统的 log ，例如 MySQL 里面的 binlog持续监控
开源产品Sqoop, Kafka JDBC SourceCanal, Maxwell, Debezium
执行模式BatchStreaming
捕获所有数据的变化❌✅
低延迟，不增加数据库负载❌✅
不侵入业务（LastUpdated字段）❌✅
捕获删除事件和旧记录的状态❌✅
捕获旧记录的状态❌✅

flink-connector-gbasedbt

我们其实是可以自己手写Sink将CDC的数据直接汇入我们的目标数据库的。这样是不是不够优雅？我们是不是可以通过Flink SQL的方式将数据汇入到GBase8s呢？答案是肯定的，接下来我们就来实现一个简单的GBase8s的Flink connector

  1. 构建 行转换器（RowConverter）

  2. 构建 方言（Dialect）

  3. 注册动态表工厂（DynamicTableFactory），以及相关Sink程序

经过上面三步，就可以实现一个简单的connector了。接下来我们就来看，如何实现:

构建 行转换器（RowConverter）

  1. package wang.datahub.converter;
  2.  
  3. import org.apache.flink.connector.jdbc.internal.converter.AbstractJdbcRowConverter;
  4. import org.apache.flink.table.types.logical.RowType;
  5.  
  6. /**
  7. * @author lijiaqi
  8. */
  9. public class GBasedbtRowConverter extends AbstractJdbcRowConverter {
  10.  
  11.    public GBasedbtRowConverter(RowType rowType) {
  12.        super(rowType);
  13.   }
  14.  
  15.    private static final long serialVersionUID = 1L;
  16.  
  17.    @Override
  18.    public String converterName() {
  19.        return "gbasedbt";
  20.   }
  21.  
  22. }

构建 方言（Dialect）

  1. package wang.datahub.dialect;
  2.  
  3. import org.apache.flink.connector.jdbc.dialect.JdbcDialect;
  4. import org.apache.flink.connector.jdbc.internal.converter.JdbcRowConverter;
  5. import org.apache.flink.table.api.TableSchema;
  6. import org.apache.flink.table.api.ValidationException;
  7. import org.apache.flink.table.types.logical.RowType;
  8. import wang.datahub.converter.GBasedbtRowConverter;
  9.  
  10. import java.util.Optional;
  11.  
  12. /**
  13. *
  14. * @author lijiaqi
  15. */
  16. public class GBasedbtDialect implements JdbcDialect {
  17.  
  18.    private static final long serialVersionUID = 1L;
  19.  
  20.    @Override
  21.    public String dialectName() {
  22.        return "gbasedbt";
  23.   }
  24.  
  25.    @Override
  26.    public boolean canHandle(String url) {
  27.        return url.startsWith("jdbc:gbasedbt-sqli:");
  28.   }
  29.  
  30.    @Override
  31.    public JdbcRowConverter getRowConverter(RowType rowType) {
  32.        return new GBasedbtRowConverter(rowType);
  33.   }
  34.  
  35.    @Override
  36.    public String getLimitClause(long l) {
  37.        return null;
  38.   }
  39.  
  40.    @Override
  41.    public void validate(TableSchema schema) throws ValidationException {
  42.        JdbcDialect.super.validate(schema);
  43.   }
  44.  
  45.    @Override
  46.    public Optional<String> defaultDriverName() {
  47.        return Optional.of("com.gbasedbt.jdbc.Driver");
  48.   }
  49.  
  50.    @Override
  51.    public String quoteIdentifier(String identifier) {
  52.        return "'" + identifier + "'";
  53.   }
  54.  
  55.    @Override
  56.    public Optional<String> getUpsertStatement(String tableName, String[] fieldNames, String[] uniqueKeyFields) {
  57.        return JdbcDialect.super.getUpsertStatement(tableName, fieldNames, uniqueKeyFields);
  58.   }
  59.  
  60.    @Override
  61.    public String getRowExistsStatement(String tableName, String[] conditionFields) {
  62.        return JdbcDialect.super.getRowExistsStatement(tableName, conditionFields);
  63.   }
  64.  
  65.    @Override
  66.    public String getInsertIntoStatement(String tableName, String[] fieldNames) {
  67.        return JdbcDialect.super.getInsertIntoStatement(tableName, fieldNames);
  68.   }
  69.  
  70.    @Override
  71.    public String getUpdateStatement(String tableName, String[] fieldNames, String[] conditionFields) {
  72.        return JdbcDialect.super.getUpdateStatement(tableName, fieldNames, conditionFields);
  73.   }
  74.  
  75.    @Override
  76.    public String getDeleteStatement(String tableName, String[] conditionFields) {
  77.        return JdbcDialect.super.getDeleteStatement(tableName, conditionFields);
  78.   }
  79.  
  80.    @Override
  81.    public String getSelectFromStatement(String tableName, String[] selectFields, String[] conditionFields) {
  82.        return JdbcDialect.super.getSelectFromStatement(tableName, selectFields, conditionFields);
  83.   }
  84.  
  85. }

注册动态表工厂（DynamicTableFactory），以及相关Sink程序

首先创建 GBasedbtSinkFunction 用于接受RowData数据输入，并将其Sink到配置的数据库中

  1. package wang.datahub.table;
  2.  
  3. import org.apache.flink.api.common.serialization.SerializationSchema;
  4. import org.apache.flink.configuration.Configuration;
  5. import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
  6. import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
  7. import org.apache.flink.table.data.RowData;
  8. import org.apache.flink.table.types.DataType;
  9.  
  10. import java.sql.Connection;
  11. import java.sql.DriverManager;
  12. import java.sql.Statement;
  13.  
  14. /**
  15. * @author lijiaqi
  16. */
  17. public class GBasedbtSinkFunction extends RichSinkFunction<RowData> {
  18.  
  19.    private static final long serialVersionUID = 1L;
  20.  
  21.    private final JdbcOptions jdbcOptions;
  22.    private final SerializationSchema<RowData> serializationSchema = null;
  23.    private DataType dateType;
  24.  
  25.    private Connection conn;
  26.    private Statement stmt;
  27.  
  28.    public GBasedbtSinkFunction(JdbcOptions jdbcOptions) {
  29.        this.jdbcOptions = jdbcOptions;
  30.   }
  31.  
  32.    public GBasedbtSinkFunction(JdbcOptions jdbcOptions, DataType dataType) {
  33.        this.jdbcOptions = jdbcOptions;
  34.        this.dateType = dataType;
  35.   }
  36.  
  37.    @Override
  38.    public void open(Configuration parameters) {
  39.        System.out.println("open connection !!!!!");
  40.        try {
  41.            if (null == conn) {
  42.                Class.forName(jdbcOptions.getDriverName());
  43.                conn = DriverManager.getConnection(jdbcOptions.getDbURL(),jdbcOptions.getUsername().orElse(null),jdbcOptions.getPassword().orElse(null));
  44.           }
  45.       } catch (Exception e) {
  46.            e.printStackTrace();
  47.       }
  48.   }
  49.  
  50.    @Override
  51.    public void invoke(RowData value, Context context) throws Exception {
  52.  
  53.        try {
  54.            stmt = conn.createStatement();
  55.            String sql = "insert into " + this.jdbcOptions.getTableName() + " values ( ";
  56.            for (int i = 0; i < value.getArity(); i++) {
  57.                //这里需要根据事情类型进行匹配
  58.                if(dateType.getChildren().get(i).getConversionClass().equals(Integer.class)){
  59.                    sql += +value.getInt(i)+ " ,";
  60.               }else {
  61.                    sql += "'"+value.getString(i) + "' ,";
  62.               }
  63.           }
  64.            sql = sql.substring(0, sql.length() - 1);
  65.            sql += " ); ";
  66.  
  67.            System.out.println("sql ==>" + sql);
  68.  
  69.            stmt.execute(sql);
  70.       }catch(Exception e){
  71.            e.printStackTrace();
  72.       }
  73.   }
  74.  
  75.    @Override
  76.    public void close() throws Exception {
  77.        if (stmt != null) {
  78.            stmt.close();
  79.       }
  80.        if (conn != null) {
  81.            conn.close();
  82.       }
  83.   }
  84.  
  85. }

构建 GBasedbtDynamicTableSink  

  1. package wang.datahub.table;
  2.  
  3. import org.apache.flink.api.common.serialization.SerializationSchema;
  4. import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
  5. import org.apache.flink.table.connector.ChangelogMode;
  6. import org.apache.flink.table.connector.format.EncodingFormat;
  7. import org.apache.flink.table.connector.sink.DynamicTableSink;
  8. import org.apache.flink.table.connector.sink.SinkFunctionProvider;
  9. import org.apache.flink.table.data.RowData;
  10. import org.apache.flink.table.types.DataType;
  11.  
  12. /**
  13. * @author lijiaqi
  14. */
  15. public class GBasedbtDynamicTableSink implements DynamicTableSink {
  16.  
  17.    private final JdbcOptions jdbcOptions;
  18.    private final EncodingFormat<SerializationSchema<RowData>> encodingFormat;
  19.    private final DataType dataType;
  20.  
  21.    public GBasedbtDynamicTableSink(JdbcOptions jdbcOptions, EncodingFormat<SerializationSchema<RowData>> encodingFormat, DataType dataType) {
  22.        this.jdbcOptions = jdbcOptions;
  23.        this.encodingFormat = encodingFormat;
  24.        this.dataType = dataType;
  25.   }
  26.  
  27.    @Override
  28.    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
  29.        return requestedMode;
  30.   }
  31.  
  32.    @Override
  33.    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
  34.        System.out.println("SinkRuntimeProvider");
  35.        System.out.println(dataType);
  36.        GBasedbtSinkFunction gbasedbtSinkFunction = new GBasedbtSinkFunction(jdbcOptions,dataType);
  37.        return SinkFunctionProvider.of(gbasedbtSinkFunction);
  38.   }
  39.  
  40.    @Override
  41.    public DynamicTableSink copy() {
  42.        return new GBasedbtDynamicTableSink(jdbcOptions, encodingFormat, dataType);
  43.   }
  44.  
  45.    @Override
  46.    public String asSummaryString() {
  47.        return "gbasedbt Table Sink";
  48.   }
  49.  
  50. }

构建GBasedbtDynamicTableFactory

  1. package wang.datahub.table;
  2.  
  3.  
  4. import org.apache.flink.configuration.ConfigOption;
  5. import org.apache.flink.configuration.ConfigOptions;
  6. import org.apache.flink.configuration.ReadableConfig;
  7. import org.apache.flink.connector.jdbc.internal.options.JdbcOptions;
  8. import org.apache.flink.table.api.TableSchema;
  9. import org.apache.flink.table.connector.sink.DynamicTableSink;
  10. import org.apache.flink.table.connector.source.DynamicTableSource;
  11. import org.apache.flink.table.factories.DynamicTableSinkFactory;
  12. import org.apache.flink.table.factories.DynamicTableSourceFactory;
  13. import org.apache.flink.table.factories.FactoryUtil;
  14. import org.apache.flink.table.types.DataType;
  15. import org.apache.flink.table.utils.TableSchemaUtils;
  16. import wang.datahub.dialect.GBasedbtDialect;
  17.  
  18. import java.util.HashSet;
  19. import java.util.Set;
  20.  
  21. /**
  22. * @author lijiaqi
  23. */
  24. public class GBasedbtDynamicTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {
  25.  
  26.    public static final String IDENTIFIER = "gbasedbt";
  27.  
  28.    private static final String DRIVER_NAME = "com.gbasedbt.jdbc.Driver";
  29.  
  30.    public static final ConfigOption<String> URL = ConfigOptions
  31.           .key("url")
  32.           .stringType()
  33.           .noDefaultValue()
  34.           .withDescription("the jdbc database url.");
  35.  
  36.    public static final ConfigOption<String> DRIVER = ConfigOptions
  37.           .key("driver")
  38.           .stringType()
  39.           .defaultValue(DRIVER_NAME)
  40.           .withDescription("the jdbc driver.");
  41.  
  42.    public static final ConfigOption<String> TABLE_NAME = ConfigOptions
  43.           .key("table-name")
  44.           .stringType()
  45.           .noDefaultValue()
  46.           .withDescription("the jdbc table name.");
  47.  
  48.    public static final ConfigOption<String> USERNAME = ConfigOptions
  49.           .key("username")
  50.           .stringType()
  51.           .noDefaultValue()
  52.           .withDescription("the jdbc user name.");
  53.  
  54.    public static final ConfigOption<String> PASSWORD = ConfigOptions
  55.           .key("password")
  56.           .stringType()
  57.           .noDefaultValue()
  58.           .withDescription("the jdbc password.");
  59.  
  60. //   public static final ConfigOption<String> FORMAT = ConfigOptions
  61. //           .key("format")
  62. //           .stringType()
  63. //           .noDefaultValue()
  64. //           .withDescription("the format.");
  65.  
  66.    @Override
  67.    public String factoryIdentifier() {
  68.        return IDENTIFIER;
  69.   }
  70.  
  71.    @Override
  72.    public Set<ConfigOption<?>> requiredOptions() {
  73.        Set<ConfigOption<?>> requiredOptions = new HashSet<>();
  74.        requiredOptions.add(URL);
  75.        requiredOptions.add(TABLE_NAME);
  76.        requiredOptions.add(USERNAME);
  77.        requiredOptions.add(PASSWORD);
  78. //       requiredOptions.add(FORMAT);
  79.        return requiredOptions;
  80.   }
  81.  
  82.    @Override
  83.    public Set<ConfigOption<?>> optionalOptions() {
  84.        return new HashSet<>();
  85.   }
  86.  
  87.    @Override
  88.    public DynamicTableSource createDynamicTableSource(Context context) {
  89.        
  90.        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
  91.  
  92.        final ReadableConfig config = helper.getOptions();
  93.        
  94.        helper.validate();
  95.  
  96.        JdbcOptions jdbcOptions = getJdbcOptions(config);
  97.        
  98.        TableSchema physicalSchema = TableSchemaUtils.getPhysicalSchema(context.getCatalogTable().getSchema());
  99.  
  100.        return new GBasedbtDynamicTableSource(jdbcOptions, physicalSchema);
  101.  
  102.   }
  103.  
  104.    @Override
  105.    public DynamicTableSink createDynamicTableSink(Context context) {
  106.        
  107.        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
  108.  
  109. //       final EncodingFormat<SerializationSchema<RowData>> encodingFormat = helper.discoverEncodingFormat(
  110. //               SerializationFormatFactory.class,
  111. //               FactoryUtil.FORMAT);
  112.  
  113.        final ReadableConfig config = helper.getOptions();
  114.  
  115.        helper.validate();
  116.  
  117.        JdbcOptions jdbcOptions = getJdbcOptions(config);
  118.  
  119.        final DataType dataType = context.getCatalogTable().getSchema().toPhysicalRowDataType();
  120.  
  121.        return new GBasedbtDynamicTableSink(jdbcOptions, null, dataType);
  122.   }
  123.  
  124.    private JdbcOptions getJdbcOptions(ReadableConfig readableConfig) {
  125.        final String url = readableConfig.get(URL);
  126.        final JdbcOptions.Builder builder = JdbcOptions.builder()
  127.               .setDriverName(DRIVER_NAME)
  128.               .setDBUrl(url)
  129.               .setTableName(readableConfig.get(TABLE_NAME))
  130.               .setDialect(new GBasedbtDialect());
  131.  
  132.        readableConfig.getOptional(USERNAME).ifPresent(builder::setUsername);
  133.        readableConfig.getOptional(PASSWORD).ifPresent(builder::setPassword);
  134.        return builder.build();
  135.   }
  136.  
  137. }

接下来通过SPI注册动态表：创建文件resources\META-INF\services\org.apache.flink.table.factories.Factory内容注册为wang.datahub.table.GBasedbtDynamicTableFactory

至此，我们的Flink connector 就构建完成，接下来，我们要使用其，来完成一个真正的项目。

实战项目

下面是项目的整体架构图，我们通过flink cdc 从mysql获取变更数据，然后通过 flink sql 将数据 sink 到 gbase8s 里

接下来，我们看一下如何通过Flink SQL实现CDC ，只需3条SQL语句即可。

创建数据源表

  1.        // 数据源表
  2.        String sourceDDL =
  3.                "CREATE TABLE mysql_binlog (\n" +
  4.                        " id INT NOT NULL,\n" +
  5.                        " name STRING,\n" +
  6.                        " description STRING\n" +
  7.                        ") WITH (\n" +
  8.                        " 'connector' = 'mysql-cdc',\n" +
  9.                        " 'hostname' = 'localhost',\n" +
  10.                        " 'port' = '3306',\n" +
  11.                        " 'username' = 'flinkcdc',\n" +
  12.                        " 'password' = '123456',\n" +
  13.                        " 'database-name' = 'test',\n" +
  14.                        " 'table-name' = 'test_cdc'\n" +
  15.                        ")";

创建输出表，输出到GBase8s ，这里 connector设置成gbasedbt

  1.        String url = "jdbc:gbasedbt-sqli://172.31.95.133:9088/t1:GBASEDBTSERVER=myserver;NEWCODESET=UTF8,zh_cn.UTF8,57372;DATABASE=mydb;DB_LOCALE=en_US.819;";
  2.        String userName = "gbasedbt";
  3.        String password = "123456";
  4.        String gbasedbtSinkTable = "ta";
  5.        // 输出目标表
  6.        String sinkDDL =
  7.                "CREATE TABLE test_cdc_sink (\n" +
  8.                        " id INT NOT NULL,\n" +
  9.                        " name STRING,\n" +
  10.                        " description STRING,\n" +
  11.                        " PRIMARY KEY (id) NOT ENFORCED \n " +
  12.                        ") WITH (\n" +
  13.                        " 'connector' = 'gbasedbt',\n" +
  14. //                       " 'driver' = 'com.gbasedbt.jdbc.Driver',\n" +
  15.                        " 'url' = '" + url + "',\n" +
  16.                        " 'username' = '" + userName + "',\n" +
  17.                        " 'password' = '" + password + "',\n" +
  18.                        " 'table-name' = '" + gbasedbtSinkTable + "' \n" +
  19.                        ")";

这里我们直接将数据汇入

  1.        String transformSQL =
  2.                "insert into test_cdc_sink select * from mysql_binlog";

完整参考代码

  1. package wang.datahub.cdc;
  2.  
  3. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
  4. import org.apache.flink.table.api.EnvironmentSettings;
  5. import org.apache.flink.table.api.SqlDialect;
  6. import org.apache.flink.table.api.TableResult;
  7. import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
  8.  
  9. public class MysqlToGBasedbtlMain {
  10.    public static void main(String[] args) throws Exception {
  11.        EnvironmentSettings fsSettings = EnvironmentSettings.newInstance()
  12.               .useBlinkPlanner()
  13.               .inStreamingMode()
  14.               .build();
  15.        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  16.        env.setParallelism(1);
  17.        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, fsSettings);
  18.  
  19.  
  20.  
  21.        tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);
  22.  
  23.  
  24.        // 数据源表
  25.        String sourceDDL =
  26.                "CREATE TABLE mysql_binlog (\n" +
  27.                        " id INT NOT NULL,\n" +
  28.                        " name STRING,\n" +
  29.                        " description STRING\n" +
  30.                        ") WITH (\n" +
  31.                        " 'connector' = 'mysql-cdc',\n" +
  32.                        " 'hostname' = 'localhost',\n" +
  33.                        " 'port' = '3306',\n" +
  34.                        " 'username' = 'flinkcdc',\n" +
  35.                        " 'password' = '123456',\n" +
  36.                        " 'database-name' = 'test',\n" +
  37.                        " 'table-name' = 'test_cdc'\n" +
  38.                        ")";
  39.  
  40.  
  41.        String url = "jdbc:gbasedbt-sqli://172.31.95.133:9088/t1:GBASEDBTSERVER=myserver;NEWCODESET=UTF8,zh_cn.UTF8,57372;DATABASE=mydb;DB_LOCALE=en_US.819;";
  42.        String userName = "gbasedbt";
  43.        String password = "123456";
  44.        String gbasedbtSinkTable = "ta";
  45.        // 输出目标表
  46.        String sinkDDL =
  47.                "CREATE TABLE test_cdc_sink (\n" +
  48.                        " id INT NOT NULL,\n" +
  49.                        " name STRING,\n" +
  50.                        " description STRING,\n" +
  51.                        " PRIMARY KEY (id) NOT ENFORCED \n " +
  52.                        ") WITH (\n" +
  53.                        " 'connector' = 'gbasedbt',\n" +
  54. //                       " 'driver' = 'com.gbasedbt.jdbc.Driver',\n" +
  55.                        " 'url' = '" + url + "',\n" +
  56.                        " 'username' = '" + userName + "',\n" +
  57.                        " 'password' = '" + password + "',\n" +
  58.                        " 'table-name' = '" + gbasedbtSinkTable + "' \n" +
  59.                        ")";
  60.  
  61.        String transformSQL =
  62.                "insert into test_cdc_sink select * from mysql_binlog";
  63.  
  64.        tableEnv.executeSql(sourceDDL);
  65.        tableEnv.executeSql(sinkDDL);
  66.        TableResult result = tableEnv.executeSql(transformSQL);
  67.  
  68.        result.print();
  69.        env.execute("sync-flink-cdc");
  70.   }
  71.  
  72. }

运行结果

查看数据，已经录入进数据库里

参考链接：

https://blog.csdn.net/zhangjun5965/article/details/107605396

https://cloud.tencent.com/developer/article/1745233?from=article.detail.1747773

https://segmentfault.com/a/1190000039662261

https://www.cnblogs.com/weijiqian/p/13994870.html

https://blog.csdn.net/dafei1288/article/details/118192917

本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号