示例：自定义 Redis Connector

public class RedisOutputFormat<T> extends RichOutputFormat<T> {
    //TODO 通过重写open 方法来初始化所需要的数据源相关连接
    @Override
    public void open(int taskNumber, int numTasks) throws IOException {
        JedisPoolConfig jedisPoolConfig = RedisConnectionUtils.getJedisPoolConfig();
        jedisPool = RedisConnectionUtils.getJedisPool(jedisPoolConfig, redisIp, redisPort, redisPassword);
        LOG.info("Initialize redis success.");
    }

    //TODO 通过重新writeRecord将flink的中的数据转为所需要的数据类型，并对相关数据源进行具体的写入操作
    @Override
    public void writeRecord(T record) throws IOException {
        Tuple2<Boolean, Row> tuple;
        RowData rowData;
        if (record instanceof Tuple2) {
            tuple = (Tuple2<Boolean, Row>) record;
            writeRedis(tuple.f1);
        } else if (record instanceof RowData) {
            rowData = (RowData) record;
            writeRedis(rowData);
        } else {
            throw new FlinkRuntimeException("can not handle record: " + record);
        }
    }

    // TODO 释放资源具体逻辑
    @Override
    public void close() throws IOException {
        RedisConnectionUtils.closeJedisPool(jedisPool);
    }
}

public class RedisSinkFunction<T> extends RichSinkFunction<T> implements CheckpointedFunction {
    private final RedisOutputFormat<T> redisOutputFormat;

    public RedisSinkFunction(RedisOutputFormat<T> redisOutputFormat) {
        this.redisOutputFormat = redisOutputFormat;
    }

    // TODO 定义相关数据源具体的初始化逻辑
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        final RuntimeContext ctx = getRuntimeContext();
        redisOutputFormat.setRuntimeContext(ctx);
        redisOutputFormat.open(ctx.getIndexOfThisSubtask(), ctx.getNumberOfParallelSubtasks());
    }

    // TODO 定义快照相关逻辑
    @Override
    public void snapshotState(FunctionSnapshotContext context) throws Exception {

    }

    // TODO 定义从快照恢复任务的相关逻辑
    @Override
    public void initializeState(FunctionInitializationContext context) throws Exception {

    }

    // TODO 定义数据具体的写入相关逻辑
    @Override
    public void invoke(T value, Context context) throws Exception {
        redisOutputFormat.writeRecord(value);
    }

    // TODO 定义释放资源的相关逻辑
    @Override
    public void close() throws Exception {
        redisOutputFormat.close();
        super.close();
    }
}

public class RedisLookupFunction extends TableFunction<Row> {
    // TODO 初始化数据源相关逻辑
    @Override
    public void open(FunctionContext context) throws Exception {
        JedisPoolConfig config = RedisConnectionUtils.getJedisPoolConfig();
        jedisPool = RedisConnectionUtils.getJedisPool(config, ip, port, redisPassword);
        if (openCache) {
            if (cacheMaxSize <= 0 || cacheMaxSize > DEFAULT_MAX_SIZE) {
                cacheMaxSize = DEFAULT_MAX_SIZE;
            }
            cache = new LRUMap(this.cacheMaxSize);
        }
    }

    // TODO 返回数据类型结果集
    @Override
    public TypeInformation<Row> getResultType() {
        DataType dataType = TableSchema.builder().fields(fieldNames, fieldTypes).build().toRowDataType();
        return (TypeInformation<Row>) TypeConversions.fromDataTypeToLegacyInfo(dataType);
    }

    // TODO 释放资源相关逻辑
    @Override
    public void close() throws Exception {
        RedisConnectionUtils.closeJedisPool(jedisPool);
    }

    // TODO 具体lookup 相关逻辑，通过collect方法收集读取数据
    public void eval(Object... params) {
        Pair<Boolean, Object> pair = null;
        try {
            final String joinKeyVal = String.valueOf(params[0]);
            String redisKey = joinKeyVal;
            String[] values = null;

            if (writeOpen) {
                values = StringUtils.split(joinKeyVal, keyAppendCharacter);
                redisKey = values[values.length - 1];
            }

            pair = getObjectFromRedis(redisKey);

            if (writeOpen && (redisWriteStrategy == RedisWriteStrategy.ALWAYS || redisWriteStrategy == RedisWriteStrategy.WRITE_IF_NOT_EXIST && !pair.getLeft())) {
                setKVToRedis(values);
            }
            collect(Row.of(redisKey, pair.getRight()));
        } catch (Exception e) {
            LOG.error(String.format("redis lookup join failed, params [%s], pair [%s]", Arrays.toString(params),
                    (null == pair ? "null" : pair.getLeft() + ":" + pair.getRight())), e);
        }
    }
}

public class RedisDynamicTableSource implements LookupTableSource, ScanTableSource {

    // TODO 返回封装后的lookupfunction的类
    @Override
    public LookupRuntimeProvider getLookupRuntimeProvider(LookupContext context) {
        String[] keyNames = new String[context.getKeys().length];
        for (int i = 0; i < keyNames.length; i++) {
            final int[] innerKeyArr = context.getKeys()[i];
            Preconditions.checkArgument(innerKeyArr.length == 1, "redis only support non-nested look up keys");
            keyNames[i] = fieldNames[innerKeyArr[0]];
        }
        RedisLookupFunction lookupFunction =
                RedisLookupFunction.builder()
                        .setIp(ip)
                        .setPort(port)
                        .setRedisPassword(redisPassword)
                        .setRedisValueType(redisValueType)
                        .setFieldNames(fieldNames)
                        .setFieldTypes(fieldTypes)
                        .setRedisJoinKeys(keyNames)
                        .setCacheMaxSize(cacheMaxSize)
                        .setOpenCache(openCache)
                        .setKeyAppendCharacter(keyAppendCharacter)
                        .setWriteOpen(writeOpen)
                        .setRedisWriteStrategy(redisWriteStrategy)
                        .build();
        return TableFunctionProvider.of(lookupFunction);
    }

    // TODO 定义changelog所支持的操作类型
    @Override
    public ChangelogMode getChangelogMode() {
        return ChangelogMode.newBuilder()
                .addContainedKind(RowKind.INSERT)
                .addContainedKind(RowKind.UPDATE_BEFORE)
                .addContainedKind(RowKind.UPDATE_AFTER)
                .build();
    }

    // TODO 返回读取数据相关的类
    @Override
    public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeProviderContext) {
        return null;
    }

    @Override
    public DynamicTableSource copy() {
        return new RedisDynamicTableSource(
                fieldNames,
                fieldTypes,
                ip,
                port,
                redisPassword,
                redisValueType,
                openCache,
                cacheMaxSize,
                keyAppendCharacter,
                writeOpen,
                redisWriteStrategy);
    }

    @Override
    public String asSummaryString() {
        return "redis source";
    }
}

public class RedisDynamicTableSink implements DynamicTableSink {
    // TODO 选择changelog支持的操作类型
    @Override
    public ChangelogMode getChangelogMode(ChangelogMode requestedMode) {
        return ChangelogMode.newBuilder()
                .addContainedKind(RowKind.INSERT)
                .addContainedKind(RowKind.UPDATE_BEFORE)
                .addContainedKind(RowKind.UPDATE_AFTER)
                .addContainedKind(RowKind.DELETE)
                .build();
    }

    // TODO 返回写入数据相关的类
    @Override
    public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
        SinkFunction sinkFunction = new RedisSinkFunction<>(newFormat());
        return SinkFunctionProvider.of(sinkFunction);
    }

    @Override
    public DynamicTableSink copy() {
        return new RedisDynamicTableSink(
                redisIp, redisPort, redisPassword, redisValueType, tableSchema, ttl);
    }

    @Override
    public String asSummaryString() {
        return "redis sink";
    }
}

public class RedisDynamicTableFactory implements DynamicTableSourceFactory, DynamicTableSinkFactory {

    // TODO 定义数据结果类的初始化逻辑
    @Override
    public DynamicTableSink createDynamicTableSink(Context context) {
        final FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        final ReadableConfig tableOptions = helper.getOptions();
        helper.validateExcept("No");
        return RedisDynamicTableSink.builder()
                .setRedisIp(tableOptions.get(HOSTNAME))
                .setRedisPort(tableOptions.get(PORT))
                .setRedisPassword(tableOptions.get(PASSWORD))
                .setRedisValueType(tableOptions.get(TYPE))
                .setTtl(tableOptions.get(TTL))
                .setTableSchema(context.getCatalogTable().getSchema())
                .build();
    }

    // TODO 定义数据源类的初始化逻辑
    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
        FactoryUtil.TableFactoryHelper helper = FactoryUtil.createTableFactoryHelper(this, context);
        final ReadableConfig readableConfig = helper.getOptions();
        helper.validateExcept("No");
        return RedisDynamicTableSource.builder()
                .setIp(readableConfig.get(HOSTNAME))
                .setPort(readableConfig.get(PORT))
                .setRedisPassword(readableConfig.get(PASSWORD))
                .setRedisValueType(readableConfig.get(TYPE))
                .setWriteOpen(readableConfig.get(WRITE_OPEN))
                .setRedisWriteStrategy(
                        RedisWriteStrategy.get(readableConfig.get(STORE_KEY_STRATEGY)))
                .setOpenCache(readableConfig.get(OPEN_CACHE))
                .setKeyAppendCharacter(readableConfig.get(KEY_APPEND_CHARACTER))
                .setCacheMaxSize(readableConfig.get(CACHE_SIZE))
                .setFieldNames(context.getCatalogTable().getSchema().getFieldNames())
                .setFieldTypes(context.getCatalogTable().getSchema().getFieldDataTypes())
                .build();
    }

    // TODO 定义 connector 参数对应的名称 'connector' = 'redis'
    @Override
    public String factoryIdentifier() {
        return "redis";
    }

    // TODO 定义必填相关参数
    @Override
    public Set<ConfigOption<?>> requiredOptions() {
        Set<ConfigOption<?>> options = new HashSet<>();
        options.add(HOSTNAME);
        options.add(PORT);
        options.add(PASSWORD);
        options.add(TYPE);
        return options;
    }

    // TODO 定义选填相关参数
    @Override
    public Set<ConfigOption<?>> optionalOptions() {
        Set<ConfigOption<?>> options = new HashSet<>();
        options.add(WRITE_OPEN);
        options.add(STORE_KEY_STRATEGY);
        options.add(KEY_APPEND_CHARACTER);
        options.add(OPEN_CACHE);
        options.add(CACHE_SIZE);
        options.add(TTL);
        return options;
    }
}