Flink-Source的使用

Data Sources 是什么呢？就字面意思其实就可以知道：数据来源。

Flink 做为一款流式计算框架，它可用来做批处理，也可以用来做流处理，这个 Data Sources 就是数据的来源地。

flink在批/流处理中常见的source主要有两大类。

预定义Source

基于本地集合的source（Collection-based-source）

基于文件的source（File-based-source）

基于网络套接字（socketTextStream）

自定义Source

预定义Source演示 

Collection [测试]--本地集合Source

在flink最常见的创建DataStream方式有四种：

l 使用env.fromElements()，这种方式也支持Tuple，自定义对象等复合形式。

注意：类型要一致，不一致可以用Object接收，但是使用会报错，比如：env.fromElements("haha", 1);

源码注释中有写：

|使用env.fromCollection(),这种方式支持多种Collection的具体类型，如List，Set，Queue

l 使用env.generateSequence()方法创建基于Sequence的DataStream --已经废弃了

l 使用env.fromSequence()方法创建基于开始和结束的DataStream

一般用于学习测试时编造数据时使用

1.env.fromElements(可变参数);

2.env.fromColletion(各种集合);

3.env.fromSequence(开始,结束);

package com.bigdata.source;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;public class _01YuDingYiSource {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 各种获取数据的SourceDataStreamSource<String> dataStreamSource = env.fromElements("hello world txt", "hello nihao kongniqiwa");dataStreamSource.print();// 演示一个错误的//DataStreamSource<Object> dataStreamSource2 = env.fromElements("hello", 1,3.0f);//dataStreamSource2.print();DataStreamSource<Tuple2<String, Integer>> elements = env.fromElements(Tuple2.of("张三", 18),Tuple2.of("lisi", 18),Tuple2.of("wangwu", 18));elements.print();// 有一个方法，可以直接将数组变为集合  复习一下数组和集合以及一些非常常见的APIString[] arr = {"hello","world"};System.out.println(arr.length);System.out.println(Arrays.toString(arr));List<String> list = Arrays.asList(arr);System.out.println(list);env.fromElements(Arrays.asList(arr),Arrays.asList(arr),Arrays.asList(arr)).print();// 第二种加载数据的方式// Collection 的子接口只有 Set 和 ListArrayList<String> list1 = new ArrayList<>();list1.add("python");list1.add("scala");list1.add("java");DataStreamSource<String> ds1 = env.fromCollection(list1);DataStreamSource<String> ds2 = env.fromCollection(Arrays.asList(arr));// 第三种DataStreamSource<Long> ds3 = env.fromSequence(1, 100);ds3.print();// execute 下面的代码不运行，所以，这句话要放在最后。env.execute("获取预定义的Source");}
}

本地文件的案例：

package com.bigdata.source;import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;public class _02YuDingYiSource {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 获取并行度System.out.println(env.getParallelism());// 讲第二种Source File类型的// 给了一个相对路径，说路径不对，老闫非要写，我咋办？// 相对路径，转绝对路径File file = new File("datas/wc.txt");File file2 = new File("./");System.out.println(file.getAbsoluteFile());System.out.println(file2.getAbsoluteFile());DataStreamSource<String> ds1 = env.readTextFile("datas/wc.txt");ds1.print();// 还可以获取hdfs路径上的数据DataStreamSource<String> ds2 = env.readTextFile("hdfs://bigdata01:9820/home/a.txt");ds2.print();// execute 下面的代码不运行，所以，这句话要放在最后。env.execute("获取预定义的Source");}
}

Socket [测试] 

socketTextStream(String hostname, int port) 方法是一个非并行的Source，该方法需要传入两个参数，第一个是指定的IP地址或主机名，第二个是端口号，即从指定的Socket读取数据创建DataStream。该方法还有多个重载的方法，其中一个是socketTextStream(String hostname, int port, String delimiter, long maxRetry)，这个重载的方法可以指定行分隔符和最大重新连接次数。这两个参数，默认行分隔符是”\n”，最大重新连接次数为0。

提示：

如果使用socketTextStream读取数据，在启动Flink程序之前，必须先启动一个Socket服务，为了方便，Mac或Linux用户可以在命令行终端输入nc -lk 8888启动一个Socket服务并在命令行中向该Socket服务发送数据。Windows用户可以在百度中搜索windows安装netcat命令。

使用nc 进行数据的发送

yum install -y nc   
nc -lk 8888   --向8888端口发送消息，这个命令先运行，如果先运行java程序，会报错！

如果是windows平台：nc -lp 8888 

代码演示：

//socketTextStream创建的DataStream，不论怎样，并行度永远是1
public class StreamSocketSource {public static void main(String[] args) throws Exception {//local模式默认的并行度是当前机器的逻辑核的数量StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();int parallelism0 = env.getParallelism();System.out.println("执行环境默认的并行度：" + parallelism0);DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);//获取DataStream的并行度int parallelism = lines.getParallelism();System.out.println("SocketSource的并行度：" + parallelism);SingleOutputStreamOperator<String> words = lines.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String line, Collector<String> collector) throws Exception {String[] words = line.split(" ");for (String word : words) {collector.collect(word);}}});int parallelism2 = words.getParallelism();System.out.println("调用完FlatMap后DataStream的并行度：" + parallelism2);words.print();env.execute();}
}

以下用于演示：统计socket中的 单词数量，体会流式计算的魅力！

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;public class SourceDemo02_Socket {public static void main(String[] args) throws Exception {//TODO 1.env-准备环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);//TODO 2.source-加载数据DataStream<String> socketDS = env.socketTextStream("bigdata01", 8889);//TODO 3.transformation-数据转换处理//3.1对每一行数据进行分割并压扁DataStream<String> wordsDS = socketDS.flatMap(new FlatMapFunction<String, String>() {@Overridepublic void flatMap(String value, Collector<String> out) throws Exception {String[] words = value.split(" ");for (String word : words) {out.collect(word);}}});//3.2每个单词记为<单词,1>DataStream<Tuple2<String, Integer>> wordAndOneDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {@Overridepublic Tuple2<String, Integer> map(String value) throws Exception {return Tuple2.of(value, 1);}});//3.3分组KeyedStream<Tuple2<String, Integer>, String> keyedDS = wordAndOneDS.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {@Overridepublic String getKey(Tuple2<String, Integer> value) throws Exception {return value.f0;}});//3.4聚合SingleOutputStreamOperator<Tuple2<String, Integer>> result = keyedDS.sum(1);//TODO 4.sink-数据输出result.print();//TODO 5.execute-执行env.execute();}
}

自定义数据源

SourceFunction:非并行数据源(并行度只能=1) --接口

RichSourceFunction:多功能非并行数据源(并行度只能=1) --类

ParallelSourceFunction:并行数据源(并行度能够>=1) --接口

RichParallelSourceFunction:多功能并行数据源(并行度能够>=1) --类 【建议使用的】

package com.bigdata.day02;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;import java.util.Random;
import java.util.UUID;/*** 需求: 每隔1秒随机生成一条订单信息(订单ID、用户ID、订单金额、时间戳)* 要求:* - 随机生成订单ID(UUID)* - 随机生成用户ID(0-2)* - 随机生成订单金额(0-100)* - 时间戳为当前系统时间*/@Data  // set get toString
@AllArgsConstructor
@NoArgsConstructor
class OrderInfo{private String orderId;private int uid;private int money;private long timeStamp;
}
// class MySource extends RichSourceFunction<OrderInfo> {
//class MySource extends RichParallelSourceFunction<OrderInfo> {
class MySource implements SourceFunction<OrderInfo> {boolean flag = true;@Overridepublic void run(SourceContext ctx) throws Exception {// 源源不断的产生数据Random random = new Random();while(flag){OrderInfo orderInfo = new OrderInfo();orderInfo.setOrderId(UUID.randomUUID().toString());orderInfo.setUid(random.nextInt(3));orderInfo.setMoney(random.nextInt(101));orderInfo.setTimeStamp(System.currentTimeMillis());ctx.collect(orderInfo);Thread.sleep(1000);// 间隔1s}}// source 停止之前需要干点啥@Overridepublic void cancel() {flag = false;}
}
public class CustomSource {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();env.setParallelism(2);// 将自定义的数据源放入到env中DataStreamSource dataStreamSource = env.addSource(new MySource())/*.setParallelism(1)*/;System.out.println(dataStreamSource.getParallelism());dataStreamSource.print();env.execute();}}

 通过ParallelSourceFunction创建可并行Source

/*** 自定义多并行度Source*/
public class CustomerSourceWithParallelDemo {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();DataStreamSource<String> mySource = env.addSource(new MySource()).setParallelism(6);mySource.print();env.execute();}public static class MySource implements ParallelSourceFunction<String> {@Overridepublic void run(SourceContext<String> ctx) throws Exception {ctx.collect(UUID.randomUUID().toString());/*如果不设置无限循环可以看出，设置了多少并行度就打印出多少条数据*/}@Overridepublic void cancel() {}}
}

如果代码换成ParallelSourceFunction，每次生成12个数据，假如是12核数的话。

总结：Rich富函数总结 ctrl + o

 Rich 类型的Source可以比非Rich的多出有：
    - open方法，实例化的时候会执行一次，多个并行度会执行多次的哦（因为是多个实例了）
    - close方法，销毁实例的时候会执行一次，多个并行度会执行多次的哦
    - getRuntimeContext 方法可以获得当前的Runtime对象（底层API）

Kafka Source --从kafka中读取数据

https://nightlies.apache.org/flink/flink-docs-release-1.13/zh/docs/connectors/datastream/kafka/

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>${flink.version}</version>
</dependency>

创建一个topic1 这个主题：

cd /opt/installs/kafka3/bin/kafka-topics.sh --bootstrap-server bigdata01:9092 --create --partitions 1 --replication-factor 3 --topic topic1通过控制台向topic1发送消息：
bin/kafka-console-producer.sh  --bootstrap-server bigdata01:9092 --topic topic1
package com.bigdata.day02;import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;import java.util.Properties;public class KafkaSource {public static void main(String[] args) throws Exception {StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();Properties properties = new Properties();properties.setProperty("bootstrap.servers", "bigdata01:9092");properties.setProperty("group.id", "g1");FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>("topic1",new SimpleStringSchema(),properties);DataStreamSource<String> dataStreamSource = env.addSource(kafkaSource);// 以下代码跟flink消费kakfa数据没关系，仅仅是将需求搞的复杂一点而已// 返回true 的数据就保留下来，返回false 直接丢弃dataStreamSource.filter(new FilterFunction<String>() {@Overridepublic boolean filter(String word) throws Exception {// 查看单词中是否包含success 字样return word.contains("success");}}).print();env.execute();}
}

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/477094.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

Linux线程_线程控制_线程库

Linux线程_线程控制_线程库

一.线程控制 在Linux操作系统的视角&#xff0c;Linux下没有真正意义上的线程&#xff0c;而是用进程模拟的线程&#xff08;LWP&#xff09;。所以&#xff0c;Linux不会提供直接创建线程的系统调用&#xff0c;而是提供创建轻量级进程的接口。但是由于用户只认线程&#xff0…
阅读更多...
计算机网络：运输层 —— TCP 的超时重传机制

计算机网络：运输层 —— TCP 的超时重传机制

文章目录 TCP 的超时重传超时重传时间的选择重传策略与拥塞控制的关联 TCP 的超时重传 TCP 的超时重传是保证数据可靠传输的重要机制之一 保证数据可靠性&#xff1a;通过超时重传机制&#xff0c;即使在网络状况不佳&#xff0c;出现数据包丢失等情况时&#xff0c;也能够确保…
阅读更多...
C嘎嘎探索篇：和stack，queue的相遇

C嘎嘎探索篇：和stack，queue的相遇

C嘎嘎探索篇&#xff1a;和stack&#xff0c;queue的再次相遇 前言&#xff1a; 小编在前几日刚完成了关于list容器的介绍&#xff0c;中间由于我牙齿出现了问题所以断更了不少天&#xff0c;如今我牙齿已经恢复&#xff0c;我也要开始继续新内容的讲解了&#xff0c;各位读者…
阅读更多...
GPTZero：高效识别AI生成文本，保障学术诚信与内容原创性

GPTZero：高效识别AI生成文本，保障学术诚信与内容原创性

产品描述 GPTZero 是一款先进的AI文本检测工具&#xff0c;专为识别由大型语言模型&#xff08;如ChatGPT、GPT-4、Bard等&#xff09;生成的文本而设计。它通过分析文本的复杂性和一致性&#xff0c;判断文本是否可能由人类编写。GPTZero 已经得到了超过100家媒体机构的报道&…
阅读更多...
MyBatis Plus 项目的创建和使用

MyBatis Plus 项目的创建和使用

1. 快速上手 1.1. 项目的创建和配置 首先&#xff0c;创建一个 Spring Boot 工程&#xff0c;添加 MyBatis Plus 和 MySQL 对应的依赖&#xff0c;然后&#xff0c;和 MyBatis 一样&#xff0c;需要在 yml 文件中配置数据库连接信息 <dependency><groupId>com.b…
阅读更多...
IDEA 2024.3 版本更新主要功能介绍

IDEA 2024.3 版本更新主要功能介绍

IDEA 2024.3 版本提供的新特性 IntelliJ IDEA 2024.3 的主要新特性&#xff1a; AI Assistant 增强 改进的代码补全和建议更智能的代码分析和重构建议Java 支持改进 支持 Java 21 的所有新特性改进的模式匹配和记录模式支持更好的虚拟线程调试体验开发工具改进 更新的 UI/UX 设…
阅读更多...
Java编程，配置mongoUri连接mongodb时，需对特殊字符进行转义

Java编程，配置mongoUri连接mongodb时，需对特殊字符进行转义

一、背景 java程序连接mongo有两种方式&#xff1a; 用户名和密码方式uri方式 1、用户名和密码 以用户数据库为例&#xff0c;注意看它的密码 spring:data:mongodb:host: 192.168.10.17database: db_user_serviceport: 3717username: user_servicepassword: user_service3…
阅读更多...
学习笔记｜MaxKB对接本地大模型时，选择Ollma还是vLLM？

学习笔记｜MaxKB对接本地大模型时，选择Ollma还是vLLM？

在使用MaxKB开源知识库问答系统的过程中&#xff0c;除了对接在线大模型&#xff0c;一些用户出于资源配置、长期使用成本、安全性等多方面考虑&#xff0c;还在积极尝试通过Ollama、vLLM等模型推理框架对接本地离线大模型。而在用户实践的过程中&#xff0c;经常会对候选的模型…
阅读更多...
Python 快速入门（上篇）❖ Python基础知识

Python 快速入门（上篇）❖ Python基础知识

Python 基础知识 Python安装**运行第一个程序：基本数据类型算术运算符变量赋值操作符转义符获取用户输入综合案例：简单计算器实现Python安装** Linux安装： yum install python36 -y或者编译安装指定版本：https://www.python.org/downloads/source/ wget https://www.pyt…
阅读更多...
【1.2 Getting Started---＞Installation Guide】

【1.2 Getting Started---＞Installation Guide】

NVIDIA TensorRT DOCS 此 NVIDIA TensorRT 10.6.0 安装指南提供安装要求、TensorRT 包中包含的内容列表以及安装 TensorRT 的分步说明。 安装指南 摘要&#xff1a; 本 NVIDIA TensorRT 10.3.0 安装指南提供了安装要求、TensorRT 软件包中包含的内容列表以及安装 TensorRT 的…
阅读更多...
RT_Thread内核源码分析(三)——线程

RT_Thread内核源码分析(三)——线程

目录 1. 线程结构 2. 线程创建 2.1 静态线程创建 2.2 动态线程创建 2.3 源码分析 2.4 线程内存结构 3. 线程状态 3.1 线程状态分类 3.2 就绪状态和运行态 3.3 阻塞/挂起状态 3.3.1 阻塞工况 3.4 关闭状态 3.4.1 线程关闭接口 3.4.2 静态线程关闭 3.4.3 动态线程关…
阅读更多...
Unity图形学之CubeMap立方体贴图

Unity图形学之CubeMap立方体贴图

1.CubeMap&#xff1a;有六个面的贴图组成 2. 假反射&#xff1a;反射天空盒子 &#xff08;1&#xff09;正常UV采样&#xff1a; &#xff08;2&#xff09;Cube的采样&#xff1a;利用反射角采样&#xff0c;反射角X和Cube的交点采样 Shader "Custom/TestReflect"…
阅读更多...
C语言基础学习：抽象数据类型（ADT）

C语言基础学习：抽象数据类型（ADT）

基础概念 抽象数据类型&#xff08;ADT&#xff09;是一种数据类型&#xff0c;它定义了一组数据以及可以在这组数据上执行的操作&#xff0c;但隐藏了数据的具体存储方式和实现细节。在C语言中&#xff0c;抽象数据类型&#xff08;ADT&#xff09;是一种非常重要的概念&…
阅读更多...
Qt-多元素控件

Qt-多元素控件

Qt中的多元素控件 Qt提供的多元素控件有&#xff1a; 这里的多元素控件都是两两一对的。 xxWidget和xxView的一个比较简单的理解就是&#xff1a; xxView是更底层的实现&#xff0c; xxWidget是基于xxView封装来的。 可以说&#xff0c;xxView使用起来比较麻烦&#xff0c;但…
阅读更多...
2023年9月GESPC++一级真题解析

2023年9月GESPC++一级真题解析

一、单选题&#xff08;每题2分&#xff0c;共30分&#xff09; 题号 123456789101112131415 答案 CDBCDBACACBBDDA 1. 我们通常说的 “ 内存 ” 属于计算机中的&#xff08;&#xff09;。 A. 输出设备 B. 输 ⼊ 设备 C. 存储设备 D. 打印设备 【答案】 C 【考纲知识点】…
阅读更多...
wend看源码-APISJON

wend看源码-APISJON

项目地址 腾讯APIJSON官方网站 定义 APIJSON 可以定义为一个面向HTTP 协议的JSON 规范&#xff0c;一个面向数据访问层的ORM 框架。其主要工作流程包括&#xff1a;前端按照既定格式组装 JSON 请求报文&#xff0c;通过 APIJSON-ORM 将这些报文直接转换为 SQL 语句&#xff0c…
阅读更多...
VMware虚拟机Ubuntu桥接模式突然连接不上网络解决办法

VMware虚拟机Ubuntu桥接模式突然连接不上网络解决办法

在Linux环境进行开发时突然发现虚拟机中的Ubuntu突然连接不上网络&#xff0c;图形化界面也找不到有线连接选项。在此记录解决办法。 解决办法 1. 在终端命令行输入以下命令&#xff1a; sudo service network-manager stop2. 然后编辑以下文件将其中NetworkingEnable fals…
阅读更多...
丹摩征文活动｜摩智算平台深度解析：Faster R-CNN模型的训练与测试实战

丹摩征文活动｜摩智算平台深度解析：Faster R-CNN模型的训练与测试实战

目录 文章前言Faster R-CNN的简介Faster RCNN的训练与测试提前准备1.1 mobaxterm&#xff08;远程连接服务器&#xff09;1.2 本文的源码下载 目标检测模型 Faster-Rcnn2.1云服务器平台 数据上传内置JupyterLab的使用本地连接使用DAMODEL实例获取实例的SSH访问信息通过SSH连接通…
阅读更多...
【数据结构】归并排序 —— 递归及非递归解决归并排序

【数据结构】归并排序 —— 递归及非递归解决归并排序

归并排序 一、归并排序1、归并排序的思想2、归并排序代码实现&#xff08;递归&#xff09;<1> 归并排序的递归区间<2> 归并排序的稳定性<3> 拷贝 3、归并排序代码实现&#xff08;非递归&#xff09;<1> 循环区间溢出问题 二、总结 一、归并排序 1、…
阅读更多...
调大Vscode资源管理器字体

调大Vscode资源管理器字体

对于调整资源管理器字体大小&#xff08;也就是下图红框&#xff09;&#xff0c;查找了网上很多方法。要么介绍的方法是调整了代码字体&#xff0c;要么是调节了终端字体&#xff0c;要么是通过整体放缩实现的调整&#xff0c;总之都不合适。 唯一的调整方法是在几篇CSDN里看到…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023