Flink 1.12.2 源码浅析 : DataSource如何获取数据并返回DataStream

小咪咪 2022-11-18 13:50 300阅读 0赞

### . ###

*  一 .前言
 *  二 .代码分析
 *   *  2.1. 入口
     *  2.2. SocketTextStreamFunction
     *   *  2.2.1. SourceFunction接口
         *  2.2.2. SourceContext
         *  2.2.3. 属性
         *  2.2.4. 构造方法
         *  2.2.5. run(SourceContext ctx)
         *  2.2.6. cancle ()
         *  2.2.7. 官方提供的SourceFunction示例
     *  2.3. addSource
 *  三 .StreamSource
 *  四 .DataStreamSource
 *  五 .ParallelSourceFunction
 *  六 .RichParallelSourceFunction

# 一 .前言 #

本文主要分析DataSource数据源如何获取数据,并且返回DataStream .  
为后续分析DataStream做铺垫.

# 二 .代码分析 #

本文从socketTextStream为例进行分析.

## 2.1. 入口 ##

`DataStream<String> text = env.socketTextStream(hostname, port, "\n");` 是读取数据&转化为`DataStream`的入口.  
所以我们从StreamExecutionEnvironment\#socketTextStream 方法开始入手.

/** * Creates a new data stream that contains the strings received infinitely from a socket. Received strings are * decoded by the system's default character set. On the termination of the socket server connection retries can be * initiated. * * <p>Let us note that the socket itself does not report on abort and as a consequence retries are only initiated when * the socket was gracefully terminated. * * @param hostname * The host name which a server socket binds * @param port * The port number which a server socket binds. A port number of 0 means that the port number is automatically * allocated. * @param delimiter * A string which splits received strings into records * @param maxRetry * The maximal retry interval in seconds while the program waits for a socket that is temporarily down. * Reconnection is initiated every second. A number of 0 means that the reader is immediately terminated, * while * a negative value ensures retrying forever. * @return A data stream containing the strings received from the socket */
    	@PublicEvolving
    	public DataStreamSource<String> socketTextStream(String hostname, int port, String delimiter, long maxRetry) { 
    		return addSource(new SocketTextStreamFunction(hostname, port, delimiter, maxRetry),
    				"Socket Stream");
    	}

## 2.2. SocketTextStreamFunction ##

StreamExecutionEnvironment\#addSource方法中,传入了一个SocketTextStreamFunction对象. 并返回`DataStreamSource`  
我们看一下SocketTextStreamFunction 的代码 .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]

### 2.2.1. SourceFunction接口 ###

> Base interface for all stream data sources in Flink.

SocketTextStreamFunction 实现了SourceFunction接口. SourceFunction接口继承了Function和 Serializable类.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]

一. Stream Source 的约定如下:
    
    1. 当Source开始发射元素时，{ @link#run}方法被调用，其中{ @link SourceContext}可用于发射元素。
    2. run方法可以根据需要运行任意长的时间。
    3. 可以调用{ @link #cancel()} 方法退出run 方法
    
    二.  CheckpointedFunction Sources
         Sources可以实现 { @link  org.apache.flink.streaming.api.checkpoint.CheckpointedFunction} 接口必须保证  state checkpointing , 内部状态更新 和 元素发送不会同时进行.
         可以在synchronized代码块汇总 使用提供的 checkpointing lock object 锁 来保护 状态更新和 发送数据.
    
         这是实现  checkpointed Source 时应遵循的基本模式：
    
    三. Timestamps and watermarks
    
     Sources可以为数据分配timestamps，并且可以手动发出watermarks 。
         但是，只有当 streaming program 在{ @link TimeCharacteristic#EventTime}上运行时，才会有效。
     在({ @link TimeCharacteristic#IngestionTime} and { @link  TimeCharacteristic#ProcessingTime})模式watermarks就会失效.

*  我们看到SourceFunction一共就两个方法

run : 执行方法, 在里面写读取数据的实际操作.  
cancel : 取消函数, 用于取消/关闭连接使用.

### 2.2.2. SourceContext ###

SourceContext 接口用于 source function 发送数据或者水印相关的操作. 这里的方法返回值都是void .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]

<table> 
 <thead> 
  <tr> 
   <th>方法</th> 
   <th>描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>void collect(T element);</td> 
   <td>从source 发送 一条没有附加timestamp的数据, 在大多数情况下,这是默认的数据获取方式<br>element的时间戳取决于streaming program的 time characteristic</td> 
  </tr> 
  <tr> 
   <td>void collectWithTimestamp(T element, long timestamp);</td> 
   <td>从source发送一条给定timestamp的数据.</td> 
  </tr> 
  <tr> 
   <td>void emitWatermark(Watermark mark);</td> 
   <td>发送一个Watermark</td> 
  </tr> 
  <tr> 
   <td>void markAsTemporarilyIdle();</td> 
   <td>标记当前source 为暂时空闲.</td> 
  </tr> 
  <tr> 
   <td>Object getCheckpointLock();</td> 
   <td>返回一个checkpoint lock .</td> 
  </tr> 
  <tr> 
   <td>void close();</td> 
   <td>关闭</td> 
  </tr> 
 </tbody> 
</table>

### 2.2.3. 属性 ###

SocketTextStreamFunction 类里面有一些属性是为了建立请求用的信息.

// 主机地址
        private final String hostname;
        // 端口
        private final int port;
        // 分隔符
        private final String delimiter;
        // 最大重试次数
        private final long maxNumRetries;
        // 重试间隔
        private final long delayBetweenRetries;
    
        // 构建的socket
        private transient Socket currentSocket;
    
        // 标识符,是否正在运行
        private volatile boolean isRunning = true;

### 2.2.4. 构造方法 ###

构造方法主要就是为了赋值.

public SocketTextStreamFunction(
                String hostname, int port, String delimiter, long maxNumRetries) { 
            this(hostname, port, delimiter, maxNumRetries, DEFAULT_CONNECTION_RETRY_SLEEP);
        }
    
        public SocketTextStreamFunction(
                String hostname,
                int port,
                String delimiter,
                long maxNumRetries,
                long delayBetweenRetries) { 
            checkArgument(isValidClientPort(port), "port is out of range");
            checkArgument(
                    maxNumRetries >= -1,
                    "maxNumRetries must be zero or larger (num retries), or -1 (infinite retries)");
            checkArgument(delayBetweenRetries >= 0, "delayBetweenRetries must be zero or positive");
    
            this.hostname = checkNotNull(hostname, "hostname must not be null");
            this.port = port;
            this.delimiter = delimiter;
            this.maxNumRetries = maxNumRetries;
            this.delayBetweenRetries = delayBetweenRetries;
        }

### 2.2.5. run(SourceContext ctx) ###

run方法主要用于建立连接&获取数据.

@Override
        public void run(SourceContext<String> ctx) throws Exception { 
    
            //构建buffer 缓存...
            final StringBuilder buffer = new StringBuilder();
    
            // 重试次数
            long attempt = 0;
    
            // 是否运行
            while (isRunning) { 
    
                // 构建socket
                try (Socket socket = new Socket()) { 
    
    
                    // 设置当前currentSocket
                    currentSocket = socket;
    
                    LOG.info("Connecting to server socket " + hostname + ':' + port);
                    // 设置连接信息和超时时间
                    socket.connect(new InetSocketAddress(hostname, port), CONNECTION_TIMEOUT_TIME);
    
                    // 读取socket数据...
                    try (BufferedReader reader =
                            new BufferedReader(new InputStreamReader(socket.getInputStream()))) { 
    
                        char[] cbuf = new char[8192];
                        int bytesRead;
                        while (isRunning && (bytesRead = reader.read(cbuf)) != -1) { 
                            buffer.append(cbuf, 0, bytesRead);
                            int delimPos;
    
                            // 读取数据...
                            while (buffer.length() >= delimiter.length()
                                    && (delimPos = buffer.indexOf(delimiter)) != -1) { 
    
                                // 获取数据记录...
                                String record = buffer.substring(0, delimPos);
                                // truncate trailing carriage return
                                if (delimiter.equals("\n") && record.endsWith("\r")) { 
                                    record = record.substring(0, record.length() - 1);
                                }
                                // 处理数据
                                ctx.collect(record);
                                // 清理掉buffer中的缓存数据...
                                buffer.delete(0, delimPos + delimiter.length());
                            }
                        }
                    }
                }
    
                // 如果退出当前循环则重试操作...
                // if we dropped out of this loop due to an EOF, sleep and retry
                if (isRunning) { 
                    // 重试次数相关处理.
                    attempt++;
                    if (maxNumRetries == -1 || attempt < maxNumRetries) { 
                        LOG.warn(
                                "Lost connection to server socket. Retrying in "
                                        + delayBetweenRetries
                                        + " msecs...");
                        Thread.sleep(delayBetweenRetries);
                    } else { 
                        // this should probably be here, but some examples expect simple exists of the
                        // stream source
                        // throw new EOFException("Reached end of stream and reconnects are not
                        // enabled.");
                        break;
                    }
                }
            }
    
            // 最后操作...
            // collect trailing data
            if (buffer.length() > 0) { 
                ctx.collect(buffer.toString());
            }
        }

### 2.2.6. cancle () ###

关闭操作

@Override
        public void cancel() { 
            isRunning = false;
    
            // we need to close the socket as well, because the Thread.interrupt() function will
            // not wake the thread in the socketStream.read() method when blocked.
            Socket theSocket = this.currentSocket;
            if (theSocket != null) { 
                IOUtils.closeSocket(theSocket);
            }
        }

### 2.2.7. 官方提供的SourceFunction示例 ###

public class ExampleCountSource implements SourceFunction<Long>, CheckpointedFunction { 
        private long count = 0L;
        private volatile boolean isRunning = true;
    
        private transient ListState<Long> checkpointedCount;
    
        public void run(SourceContext<T> ctx) { 
            while (isRunning && count < 1000) { 
                // this synchronized block ensures that state checkpointing,
                // internal state updates and emission of elements are an atomic operation
                synchronized (ctx.getCheckpointLock()) { 
                    ctx.collect(count);
                    count++;
                }
            }
        }
    
        public void cancel() { 
            isRunning = false;
        }
    
        public void initializeState(FunctionInitializationContext context) { 
            this.checkpointedCount = context
                .getOperatorStateStore()
                .getListState(new ListStateDescriptor<>("count", Long.class));
    
            if (context.isRestored()) { 
                for (Long count : this.checkpointedCount.get()) { 
                    this.count = count;
                }
            }
        }
    
        public void snapshotState(FunctionSnapshotContext context) { 
            this.checkpointedCount.clear();
            this.checkpointedCount.add(count);
        }
     }

## 2.3. addSource ##

添加具有自定义类型信息的数据源，从而打开\{@link DataStream\}。  
只有在非常特殊的情况下，用户才需要支持类型信息。  
否则使用 \{@link \#addSource(org.apache.flink.streaming.api.functions.source.SourceFunction)\}

// addSource方法用来添加一个数据源到计算任务中。
        // 默认情况下数据源是非并行的，
        // 用户需要实现ParallelSourceFunction接口或者继承RichParallelSourceFunction来实现可并行的数据源。
        //
        // addSource方法将一个StreamFunction封装为StreamSource，
        // 当数据源开始执行时调用SourceFunction#run(SourceContext<T> ctx)方法，
        // 持续地向SourceContext发送生成的数据。
        private <OUT> DataStreamSource<OUT> addSource(
                // SocketTextStreamFunction
                final SourceFunction<OUT> function,
                // Socket Stream
                final String sourceName,
                // null
                @Nullable final TypeInformation<OUT> typeInfo,
                // 无界: Boundedness.CONTINUOUS_UNBOUNDED
                final Boundedness boundedness) { 
            // Boundedness 是枚举类, 代表数据源的类型是有界数据[BOUNDED]或者是无界[CONTINUOUS_UNBOUNDED]数据.
            checkNotNull(function);
            checkNotNull(sourceName);
            checkNotNull(boundedness);
    
            // 获取数据的处理类型 : String
            TypeInformation<OUT> resolvedTypeInfo =
                    getTypeInfo(function, sourceName, SourceFunction.class, typeInfo);
    
            // 是否是并行的SourceFunction : false
            boolean isParallel = function instanceof ParallelSourceFunction;
    
            // 清理函数...
            clean(function);
    
            // 构建sourceOperator
            final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);
    
            // 构建 DataStreamSource
            return new DataStreamSource<>(
                    this, resolvedTypeInfo, sourceOperator, isParallel, sourceName, boundedness);
        }

这里有三个关键信息

1.  是否是支持并发读取数据. 是通过 `boolean isParallel = function instanceof ParallelSourceFunction;`验证的. 所以如果数据源需要支持并发读取的话,需要实现ParallelSourceFunction(SourceFunction的子类)接口.
2.  SourceFunction 会包装成 StreamSource 对象 .
3.  最终 new了一个 `DataStreamSource`对象返回.

# 三 .StreamSource #

在`addSource`方法中,是通过`final StreamSource<OUT, ?> sourceOperator = new StreamSource<>(function);` 包装成 StreamSource.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]

# 四 .DataStreamSource #

DataStreamSource表示DataStream的起点。  
DataStreamSource是SingleOutputStreamOperator的子类. 超级父类是 DataStream .  
所有的数据源都会包装成DataStreamSource .

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]

DataStreamSource代码里面除了构造方法之外, 没有任何方法, 只有一个属性 isParallel .  
从写了 `setParallelism`方法, 只有在 `isParallel=true` 的时候才可以设置并发度.

@Override
        public DataStreamSource<T> setParallelism(int parallelism) { 
            OperatorValidationUtils.validateParallelism(parallelism, isParallel);
            super.setParallelism(parallelism);
            return this;
        }

Flink 中你可以使用 StreamExecutionEnvironment.addSource(source) 来为你的程序添加数据来源。  
Flink 已经提供了若干实现好了的 source functions，当然你也可以通过实现 SourceFunction 来自定义非并行的  
source或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的source。

# 五 .ParallelSourceFunction #

ParallelSourceFunction是一个接口, 并行执行的stream data 数据源 .  
在执行时, runtime会根据source配置中的并行度去启动相同数量的function实例 .

`这个接口只是作为一个标记告诉系统这个源可以并行执行`。  
当需要不同的并行实例来执行不同的任务时，  
请使用\{@link RichParallelSourceFunction\}访问 runtime context ，  
该 runtime context 将显示诸如并行任务数 和 当前实例是哪个并行任务等信息。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]

# 六 .RichParallelSourceFunction #

RichParallelSourceFunction 是ParallelSourceFunction 的子类, 实现了ParallelSourceFunction接口 .

用于实现并行数据源的基类。  
在执行时, runtime会根据source配置中的并行度去启动相同数量的function实例 .  
这个数据源可以访问 context 信息 (比如数据源的并行实例的数量和当前的实例归属于哪个任务. )  
它还提供了附加的生命周期方法 \#open(org.apache.flink.configuration.Configuration)\} and \{@link \#close()\}.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]

@Public
    public abstract class RichParallelSourceFunction<OUT> extends AbstractRichFunction
            implements ParallelSourceFunction<OUT> { 
    
        private static final long serialVersionUID = 1L;
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70]: /images/20221022/ac55b71c9762404d97a66d7c02591535.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 1]: /images/20221022/f91416daf4e0460eb37da9e672757c6c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 2]: /images/20221022/600dcf3597f74279b883a7707b61051a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 3]: /images/20221022/979c35abb24740819d53ec7ad6cafb0d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 4]: /images/20221022/6ead20b67d124c56be34204665e8bfa0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 5]: /images/20221022/a0ca0f5ca326431699f6ef10a3cbd799.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70 6]: /images/20221022/7a2c82e35fb2492089309fed8f19bfcb.png