Flink之DataStream数据源

灰太狼 2022-11-07 04:12 56阅读 0赞

### 目录 ###

*   *  （1）Flink Source之文件数据源
     *  （2）Flink Source之Socket数据源
     *  （3）Flink Source之集合数据源
     *  （4）Flink Source之外部数据源

## （1）Flink Source之文件数据源 ##

Flink系统支持将文件内容读取到系统中，并转换成分布式数据集DataStream进行数据处理。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]  
在 DataStream API中，可以在read File方法中指定文件读取类型（Watch Type）检测文件变换时间间隔  
（interva）、文件路径过滤条件（File Filter）等参数，其中Watch Type共分为两种模式：  
`PROCESS CONTINUOUSLY`和 `PROCESS ONCE`模式在 `PROCESS CONTINUOUSLY`模式下，一且检测到文件内容发生变化，Fink会将该文件全部内容加载到Fink系统中进行处理。而在`PROCESS ONCE`模式下，当文件内容发生变化时，只会将变化的数据读取至Fink中，在这种情况下数据只会被读取和处理一次。

可以看出，在`PROCESS CONTINUOUSLY`模式下是无法实现`Excatly Once`级别数据一致性保障的，而在`PROCESSONCE`模式，可以保证数据Excatly Once级别的一致性保证。但是需要注意的是，如果使用文件作为数据源，当某个节点异常停止的时候，这种情况下 Checkpoints不会更新，如果数据一直不断地在生成将导致该节点数据形成积压，可能需要耗费非常长的时间从最新的checkpoint中恢复应用。

示例代码：

package com.aikfk.flink.datastream.source;
    
    import org.apache.flink.api.java.io.TextInputFormat;
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.core.fs.Path;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/10 9:29 下午 */
    public class FileSource { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<String> stream = env.readFile(
                    new TextInputFormat(new Path("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv"))
                    , "/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv");
    
            stream.print();
            env.execute("stream");
        }
    }

## （2）Flink Source之Socket数据源 ##

Flink支持从Socket端口中接入数据，在`StreamExecutionEnvironment`调用`socketTextStream`方法。该方法参数分别为Ip地址和端口，也可以同时传人字符串切割符delimiter和最大尝试次数maxRetry，其中delimiter负责将数据切割成Records数据格式；maxRetry在端口异常的情况，通过指定次数进行重连，如果设定为0，则Flink程序直接停止，不再尝试和端口进行重连。如下代码是使用`socketTextStream`方法实现了将数据从本地9999ロ中接入数据并转换成 `DataStream`数据集的操作：

DataStream<String> stream = env.socketTextStream("localhost",9999);

在linux系统下执行

nc -lk 9999

示例代码：

package com.aikfk.flink.datastream.source;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/10 9:29 下午 */
    public class SocketSource { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<String> stream = env.socketTextStream("bigdata-pro-m07",9999);
    
            stream.print();
            env.execute("stream");
        }
    }

## （3）Flink Source之集合数据源 ##

Fink可以直接将Java或Scaa程序中集合类（Collection）转换成 DataStream数据集，本质上是将本地集合中的数据分发到远端并行执行的节点中。目前Fink支持从Java util。 Collection和 Javautil。 Iterator序列中转换成DataStream数据集。这种方式非常适合调式Flink本地程序，但需要注意的是，集合内容的数据结构类型必须要一致，否则可能会出现数据转换异常。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]  
示例代码：

package com.aikfk.flink.datastream.source;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/10 9:29 下午 */
    public class CollectionSource { 
        public static void main(String[] args) throws Exception { 
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<Tuple2<String,Integer>> stream = env.fromElements(
                    new Tuple2<>("spark",2),
                    new Tuple2<>("hbase",3)
            );
    
            stream.print();
            env.execute("stream");
        }
    }

## （4）Flink Source之外部数据源 ##

前面提到的数据源类型都是一些基本的数据接入方式，例如从文件、Socket端口中接入数据，其实质是实现了不同的 SourceFunction， Fink将其封装成高级API，减少了用户的使用成本。对于流式计算类型的应用，数据大部分都是从外部第三方系统中获取，为此Flink通过实现 Source Function定义了非常丰富的第三方数据连接器，基本覆盖了大部分的高性能存储介质以及中间件等，其中部分连接器是仅支持读取数据，例如 TwitterStreaming API、Nety等；另外一部分仅支持数据输出（smhk，不支持数据输入（s）例如 Apache Cassandra、 Elasticsearch、 Hadoop FileSystem等。还有一部分是既支持数据输入，也支持数据输出，例如 Apache Kafka、Amazon Kinesis、 RabbitMQ等连接器。

![在这里插入图片描述][20210310220347741.png_pic_center]  
示例代码：

package com.aikfk.flink.datastream.source;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.datastream.DataStream;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/10 9:38 下午 */
    public class OutSource { 
        public static void main(String[] args) throws Exception { 
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
            DataStream<Tuple2<Integer,Integer>> stream = env.addSource(new OutSourceFunction());
    
            stream.print();
            env.execute("stream");
        }
    }

package com.aikfk.flink.datastream.source;
    
    import org.apache.flink.api.java.tuple.Tuple2;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    
    /** * @author ：caizhengjie * @description：TODO * @date ：2021/3/10 9:39 下午 */
    public class OutSourceFunction implements SourceFunction<Tuple2<Integer,Integer>> { 
    
        private int count = 0;
        @Override
        public void run(SourceContext sourceContext) throws Exception { 
    
            while (count < 1000){ 
                int first = (int) (Math.random() * 10);
                sourceContext.collect(new Tuple2<>(first,first));
                count++;
                Thread.sleep(100L);
            }
    
        }
    
        @Override
        public void cancel() { 
    
        }
    }

关于flink中DataStream数据源可以见官网：

*  [https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/datastream\_api.html\#data-sources][https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_datastream_api.html_data-sources]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/a367792834ae44219180beda8d261fd5.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221023/e9bdcecbba8f453b9c04edf322eb952b.png
[20210310220347741.png_pic_center]: /images/20221023/ee358cca58d641fd8158cf92ab31c56e.png
[https_ci.apache.org_projects_flink_flink-docs-release-1.12_zh_dev_datastream_api.html_data-sources]: https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/datastream_api.html#data-sources