本文已参与「开源摘星计划」,欢迎正在阅读的你加入。活动链接:https://github.com/weopenprojects/WeOpen-Star

你好! 这是我第一次简单分享一下开源项目SeaTunnel(下文简称ST,用于数据集成的项目)的一小部分的源码解析,如有出错,请多指教并在评论区指正。

1、总览

分支源码的总览

2、ST封装flink连接数据库的源码分析

ST对Hive、MySql、Oracle、Postgres等进行了jdbc的连接支持:

  1. DefaultTypeInformationMap ,主要是创建一个HashMap存储默认的数据类型的转换;
private static final Map<String, TypeInformation<?>> INFORMATION_MAP = new HashMap<>();

    static {
        {
            INFORMATION_MAP.put("VARCHAR", STRING_TYPE_INFO);
            INFORMATION_MAP.put("BOOLEAN", BOOLEAN_TYPE_INFO);
            INFORMATION_MAP.put("TINYINT", BYTE_TYPE_INFO);
            INFORMATION_MAP.put("SMALLINT", SHORT_TYPE_INFO);
            INFORMATION_MAP.put("INTEGER", INT_TYPE_INFO);
            INFORMATION_MAP.put("MEDIUMINT", INT_TYPE_INFO);
            INFORMATION_MAP.put("INT", INT_TYPE_INFO);
            INFORMATION_MAP.put("BIGINT", LONG_TYPE_INFO);
            INFORMATION_MAP.put("FLOAT", FLOAT_TYPE_INFO);
            INFORMATION_MAP.put("DOUBLE", DOUBLE_TYPE_INFO);
            INFORMATION_MAP.put("CHAR", STRING_TYPE_INFO);
            INFORMATION_MAP.put("TEXT", STRING_TYPE_INFO);
            INFORMATION_MAP.put("LONGTEXT", STRING_TYPE_INFO);
            INFORMATION_MAP.put("DATE", SqlTimeTypeInfo.DATE);
            INFORMATION_MAP.put("TIME", SqlTimeTypeInfo.TIME);
            INFORMATION_MAP.put("DATETIME", SqlTimeTypeInfo.TIMESTAMP);
            INFORMATION_MAP.put("TIMESTAMP", SqlTimeTypeInfo.TIMESTAMP);
            INFORMATION_MAP.put("DECIMAL", BIG_DEC_TYPE_INFO);
            INFORMATION_MAP.put("BINARY", BYTE_PRIMITIVE_ARRAY_TYPE_INFO);
        }
  1. 以此类推,PostgresTypeInformationMapMysqlTypeInformationMapOracleTypeInformationMapHiveTypeInformationMap也是创建一个HashMap存储相应的数据库字段类型的转换,因此还可以拓展更多的关系型数据库。
  2. JdbcSink 有这些核心的方法:设置配置setConfig(…)、获取配置getConfig()、检查配置checkConfig()、准备环境prepare(…)、流输出outputStream(…)、批输出outputBatch(…)、执行sql语句executeSql(…)、sink前执行sql语句executePreSql() 、sink后执行sql语句executePostSql()、关闭close();
  3. JdbcSource 有这些核心的方法:设置配置setConfig(…)、获取配置getConfig()、检查配置checkConfig()、准备环境prepare(…)、初始化分区initPartition(…)、获取数据集getData(…)、获取数据库字段对应的数据类型转换集合getRowInfo(…)、初始化表字段的数据类型initTableField(…)等等;
  4. Config 记录配置信息的一些常量:驱动名称、用户名、密码、链接等等。
    注:SeaTunnel开源项目地址
Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐