【flink】自定义flink-socket-connector_flink socket 字节

用户自定义 Sources & Sinks

概述

实心箭头展示了在转换过程中对象如何从一个阶段到下一个阶段转换为其他对象。

元数据

Table API 和 SQL 都是声明式 API。这包括表的声明。因此，执行 CREATE TABLE 语句会导致目标 catalog 中的元数据更新。

对于大多数 catalog 实现，外部系统中的物理数据不会针对此类操作进行修改。特定于连接器的依赖项不必存在于类路径中。在 WITH 子句中声明的选项既不被验证也不被解释。

动态表的元数据（ 通过 DDL 创建或由 catalog 提供 ）表示为 CatalogTable 的实例。必要时，表名将在内部解析为 CatalogTable。

解析器

在解析和优化以 table 编写的程序时，需要将 CatalogTable 解析为 DynamicTableSource（ 用于在 SELECT 查询中读取 ）和 DynamicTableSink（ 用于在 INSERT INTO 语句中写入 ）。

DynamicTableSourceFactory 和 DynamicTableSinkFactory 提供连接器特定的逻辑，用于将 CatalogTable 的元数据转换为 DynamicTableSource 和 DynamicTableSink 的实例。在大多数情况下，以工厂模式设计的目的是验证选项（例如示例中的 ‘port’ = ‘5022’ ），配置编码解码格式（ 如果需要 ），并创建表连接器的参数化实例。

默认情况下，DynamicTableSourceFactory 和 DynamicTableSinkFactory 的实例是使用 Java的 [Service Provider Interfaces (SPI)] (https://docs.oracle.com/javase/tutorial/sound/SPI-intro.html) 发现的。 connector 选项（例如示例中的 ‘connector’ = ‘custom’）必须对应于有效的工厂标识符。

尽管在类命名中可能不明显，但 DynamicTableSource 和 DynamicTableSink 也可以被视为有状态的工厂，它们最终会产生具体的运行时实现来读写实际数据。

规划器使用 source 和 sink 实例来执行连接器特定的双向通信，直到找到最佳逻辑规划。取决于声明可选的接口（ 例如 SupportsProjectionPushDown 或 SupportsOverwrite），规划器可能会将更改应用于实例并且改变产生的运行时实现。

运行时的实现

一旦逻辑规划完成，规划器将从表连接器获取 runtime implementation。运行时逻辑在 Flink 的核心连接器接口中实现，例如 InputFormat 或 SourceFunction。

这些接口按另一个抽象级别被分组为 ScanRuntimeProvider、LookupRuntimeProvider 和 SinkRuntimeProvider 的子类。

例如，OutputFormatProvider（ 提供 org.apache.flink.api.common.io.OutputFormat ）和 SinkFunctionProvider（ 提供org.apache.flink.streaming.api.functions.sink.SinkFunction）都是规划器可以处理的 SinkRuntimeProvider 具体实例。

动态表的工厂类

在根据 catalog 与 Flink 运行时上下文信息，为某个外部存储系统配置动态表连接器时，需要用到动态表的工厂类。

比如，通过实现 org.apache.flink.table.factories.DynamicTableSourceFactory 接口完成一个工厂类，来生产 DynamicTableSource 类。

通过实现 org.apache.flink.table.factories.DynamicTableSinkFactory 接口完成一个工厂类，来生产 DynamicTableSink 类。

默认情况下，Java 的 SPI 机制会自动识别这些工厂类，同时将 connector 配置项作为工厂类的”标识符“。

在 JAR 文件中，需要将实现的工厂类路径放入到下面这个配置文件：

META-INF/services/org.apache.flink.table.factories.Factory

Flink 会对工厂类逐个进行检查，确保其“标识符”是全局唯一的，并且按照要求实现了上面提到的接口 (比如 DynamicTableSourceFactory)。

如果必要的话，也可以在实现 catalog 时绕过上述 SPI 机制识别工厂类的过程。即在实现 catalog 接口时，在org.apache.flink.table.catalog.Catalog#getFactory 方法中直接返回工厂类的实例。

SocketDynamicTableFactory

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.connector.format.DecodingFormat;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.DeserializationFormatFactory;
import org.apache.flink.table.factories.DynamicTableSourceFactory;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;

import java.util.HashSet;
import java.util.Set;

/**
 * The {@link SocketDynamicTableFactory} translates the catalog table to a table source.
 *
 * <p>Because the table source requires a decoding format, we are discovering the format using the
 * provided {@link FactoryUtil} for convenience.
 */
public final class SocketDynamicTableFactory implements DynamicTableSourceFactory {
   

    // define all options statically
    public static final ConfigOption<String> HOSTNAME =
            ConfigOptions.key("hostname").stringType().noDefaultValue();

    public static final ConfigOption<Integer> PORT =
            ConfigOptions.key("port").intType().noDefaultValue();

    public static final ConfigOption<Integer> BYTE_DELIMITER =
            ConfigOptions.key("byte-delimiter").intType().defaultValue(10); // corresponds to '\n'

    @Override
    public String factoryIdentifier() {
   
        return "socket"; // used for matching to `connector = '...'`
    }

    @Override
    public Set<ConfigOption<?>> requiredOptions() {
   
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(HOSTNAME);
        options.add(PORT);
        options.add(FactoryUtil.FORMAT); // use pre-defined option for format
        return options;
    }

    @Override
    public Set<ConfigOption<?>> optionalOptions() {
   
        final Set<ConfigOption<?>> options = new HashSet<>();
        options.add(BYTE_DELIMITER);
        return options;
    }

    @Override
    public DynamicTableSource createDynamicTableSource(Context context) {
   
        // either implement your custom validation logic here ...
        // or use the provided helper utility
        final FactoryUtil.TableFactoryHelper helper =
                FactoryUtil.createTableFactoryHelper(this, context);

        // discover a suitable decoding format
        final DecodingFormat<DeserializationSchema<RowData>> decodingFormat =
                helper.discoverDecodingFormat(
                        DeserializationFormatFactory.class, FactoryUtil.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69