用一个例子告诉你 怎样在Flink DataStream API 中读取数据源
Posted 广阔天地大有可为
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了用一个例子告诉你 怎样在Flink DataStream API 中读取数据源相关的知识，希望对你有一定的参考价值。
目录
1.3 Flink 中的数据类型(TypeInformation)
1. 前言
Flink 版本 : 1.13
开发语言 : Scala 2.12
1.1 加载数据源的方式
StreamExecutionEnvironment 对象提供了多种方法来加载 数据源对象
// 通用方法
def addSource[T: TypeInformation](function: SourceFunction[T]): DataStream[T]
// 预定义方法
def readTextFile(filePath: String): DataStream[String]
def fromCollection[T: TypeInformation](data: Seq[T]): DataStream[T]
def socketTextStream(hostname: String, port: Int, delimiter: Char = '\\n', maxRetry: Long = 0):DataStream[String]
...
1.2 数据源的类型
Flink 中的数据源分为 非并行数据源 和 并行数据源 两大类
并行数据源 : 可以将数据源拆分成多个 子任务，并行执行( 并行度允许大于1)
非并行数据源 : 不可以将数据源拆分，只能有单独的任务处理数据 (并行度必须1)
我们可以通过 StreamExecutionEnvironment.addSource(SourceFunction) 将一个 source对象 关 联到编写的 flink应用程序中
Flink API中 自带了许多 SourceFunction的实现类
我们也可以 通过实现 SourceFunction接口 来编写 自定义的非并行的source对象
通过实现 ParallelSourceFunction 接口
继承 RichParallelSourceFunction 类 来编写 自定义的并行source对象
1.3 Flink 中的数据类型(TypeInformation)
Flink 会将外部的数据加载到 DataStreamSource 对象中，加载过程中会将外部的数据的类型转换为 Flink 定义的数据类型
为了方便 数据序列化和反序列化，Flink定义了自己的数据类型系统
2. 从集合中读取数据
非并行数据源
def fromElements[T: TypeInformation](data: T*): DataStream[T]
def fromCollection[T: TypeInformation](data: Seq[T]): DataStream[T]
def fromCollection[T: TypeInformation] (data: Iterator[T]): DataStream[T]
并行数据源
def fromParallelCollection[T: TypeInformation] (data: SplittableIterator[T])
代码示例:
// --------------------------------------------------------------------------------------------
// TODO 从集合中读取数据
// --------------------------------------------------------------------------------------------
test("fromCollection 方法")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将本地集合作为数据源
val list = List("刘备", "张飞", "关羽", "赵云", "马超", "...")
val ds: DataStream[String] = env.fromCollection(list).setParallelism(1)
/*
* tips: 如果这里将 并行度设置为4，将报错
* java.lang.IllegalArgumentException: The parallelism of non parallel operator must be 1.
* */
println(s"并行度: $ds.parallelism")
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
test("fromParallelCollection 方法")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将本地集合作为数据源
val iterator: NumberSequenceIterator = new NumberSequenceIterator(1, 10)
val ds: DataStream[lang.Long] = env.fromParallelCollection(iterator).setParallelism(6)
println(s"并行度: $ds.parallelism")
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
3. 从文件中读取数据
思考: 读取文件时可以设置哪些规则呢？
1. 文件的格式(txt、csv、二进制...)
2. 文件的分隔符(按\\n 分割)
3. 是否需要监控文件变化(一次读取、持续读取)
基于以上规则，Flink为我们提供了非常灵活的 读取文件的方法
3.1 readTextFile
语法说明:
定义:
def readTextFile(filePath: String): DataStream[String]
def readTextFile(filePath: String, charsetName: String)
功能:
1.读取文本格式的文件
2.按行读取(\\n为分隔符)，每行数据被封装为 DataStream 的一个元素
3.可以指定字符集(默认为UDF-8)
4.文件只会读取一次
源码分析：
public DataStreamSource<String> readTextFile(String filePath, String charsetName)
// 初始化 TextInputFormat对象
TextInputFormat format = new TextInputFormat(new Path(filePath));
// 指定路径过滤器(使用默认过滤器)
format.setFilesFilter(FilePathFilter.createDefaultFilter());
// 指定Flink中的数据类型
TypeInformation<String> typeInfo = BasicTypeInfo.STRING_TYPE_INFO;
// 指定字符集
format.setCharsetName(charsetName);
// 调用 readFile 方法
return readFile(format, filePath, FileProcessingMode.PROCESS_ONCE, -1, typeInfo);
代码示例:
// --------------------------------------------------------------------------------------------
// TODO 从文件中读取数据
// --------------------------------------------------------------------------------------------
test("readTextFile 方法")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将文本文件作为数据源
val ds: DataStream[String] = env.readTextFile("src/main/resources/data/1.txt").setParallelism(4)
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
// --------------------------------------------------------------------------------------------
// TODO 从hdfs_文本文件中读取数据
// --------------------------------------------------------------------------------------------
test("从hdfs_文本文件中读取数据")
//System.setProperty("HADOOP_USER_NAME", "root")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将文本文件作为数据源
val ds: DataStream[String] = env.readTextFile("hdfs://worker01:8020/tmp/1.txt")
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
3.2 readFile
语法说明:
定义:
def readFile[T: TypeInformation](
inputFormat: FileInputFormat[T],
filePath: String,
watchType: FileProcessingMode,
interval: Long): DataStream[T] =
val typeInfo = implicitly[TypeInformation[T]] // 隐私转换(将java 数据类型 转换为 Flink数据类型)
asScalaStream(javaEnv.readFile(inputFormat, filePath, watchType, interval, typeInfo))
参数:
inputFormat : 指定 FileInputFormat 实现类(根据文件类型 选择相适应的实例)
filePath : 指定 文件路径
watchType : 指定 读取模式(提供了2个枚举值)
PROCESS_ONCE ：只读取一次
PROCESS_CONTINUOUSLY ：按照指定周期扫描文件
interval : 指定 扫描文件的周期(单位为毫秒)
功能:
按照 指定的 文件格式 和 读取方式 读取数据
代码示例:
test("readFile 方法")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将文本文件作为数据源
val filePath = "src/main/resources/data/1.txt"
// 初始化 TextInputFormat 对象
val format = new TextInputFormat(new Path(filePath))
format.setFilesFilter(FilePathFilter.createDefaultFilter) // 指定过滤器
format.setCharsetName("UTF-8") // 指定编码格式
val ds: DataStream[String] = env.readFile(
format
, filePath
, FileProcessingMode.PROCESS_CONTINUOUSLY // 周期性读取 指定文件
, 1000 // 1s读取一次
)
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
4. 从Socket中读取数据
语法说明:
语法:
def socketTextStream(hostname: String, port: Int, delimiter: Char = '\\n', maxRetry: Long = 0):
DataStream[String] =
asScalaStream(javaEnv.socketTextStream(hostname, port))
功能:
执行监控，socket中的文本流，按行读取数据(默认分隔符为 \\n)
参数:
hostname : socket服务ip
prot : socket服务端口号
Char : 行分隔符
maxRetry : 当 socket服务 停止时，flink程序 重试连接时间(单位为秒)
=0 时，表示 连接不到 socket服务后，立刻停止 flink程序
=-1 时，表示 永远保持重试连接
tips:
scala API 只提供了一种 socketTextStream方法的实现
如果想使用其他参数，需要使用java api代码示例:
/*
* TODO 从 Socket 文本流中读取数据
* 开启socket端口: nc -lk 9999
*
* */
test("从 Socket 文本流中读取数据")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将文本文件作为数据源
//val ds: DataStream[String] = env.socketTextStream("localhost", 9999)
val ds: DataStream[String] = env.socketTextStream("localhost", 9999,'#')
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
5. 从Kafka中读取数据
语法说明:
语法:
public FlinkKafkaConsumer(String topic, DeserializationSchema<T> valueDeserializer, Properties props)
public FlinkKafkaConsumer(List<String> topics, DeserializationSchema<T> deserializer, Properties props)
参数:
topic : 指定 topic(多个topic时，使用List)
valueDeserializer : 指定 value的序列化类型(kafka数据类型 to flink数据类型)
props : 指定 kafka集群的配置信息tips:
FlinkKafkaConsumer 是Flink 提供的kafka消费者的实现类
消费者可以在多个并行示例中运行，每个实例将从一个或多个Kafka分区中提取数据
代码示例:
object FlinkReadKafka
def main(args: Array[String]): Unit =
// 0. 创建配置对象 并添加消费者相关配置信息
val properties = new Properties
/*
* bootstrap.servers
* 指定broker连接信息 (为保证高可用,建议多指定几个节点)
* 示例: host1:port1,host2:port2
* */
properties.put("bootstrap.servers", "worker01:9092")
/*
* key.deserializer value.deserializer
* 指定 key、value反序列化类型(全类名)
* 示例: key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer
* 示例: value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
* */
properties.put("key.deserializer","org.apache.kafka.common.serialization.IntegerDeserializer")
properties.put("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer")
properties.put("auto.offset.reset", "latest")
/*
* group.id
* 指定 消费者组id(不存在时会因创建)
*
* */
properties.put("group.id", "FlinkConsumer")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将kafka作为数据源
val ds: DataStream[String] = env.addSource(
new FlinkKafkaConsumer(
"20230327", new SimpleStringSchema(), properties
)
)
// 3. 打印DataStream
ds.print()
// 4. 出发程序执行
env.execute()
6. 自定义数据源
6.1 自定义非并行数据源
代码示例:
/*
* TODO 自定义非并行数据源
* 实现步骤:
* 1.实现 SourceFunction接口
* 2.实现 run方法
* 调用 collect方法 发送数据
* 3.实现 cancel方法
* 注意事项:
* 1.接口的泛型为 数据源的数据类型
* */
class CustomNonParallelSource extends SourceFunction[String]
// 标志位，用来控制循环的退出
var isRunning = true
override def run(ctx: SourceFunction.SourceContext[String]): Unit =
val list = List("刘备1", "关羽2", "张飞3", "赵云4", "马超5")
while (isRunning)
// 调用 collect 方法向下游发送数据
list.foreach(
e =>
ctx.collect(e)
Thread.sleep(1000)
)
// 通过将 isRunning 设置为false，来终止消息的发送
override def cancel(): Unit = isRunning = false
/*
* TODO 从 自定义数据源中 读取数据
*
* */
test("从 自定义非并行数据源中 读取数据")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将 自定义数据源 作为数据源
val ds: DataStream[String] = env.addSource(new CustomNonParallelSource).setParallelism(1)
// 3. 打印DataStream
ds.print().setParallelism(1)
// 4. 出发程序执行
env.execute()
执行结果:
6.2 自定义并行数据源
代码示例:
/*
* TODO 自定义并行数据源
* 实现步骤:
* 1.实现 ParallelSourceFunction接口 或者 继承RichParallelSourceFunction
* 2.实现 run方法
* 调用 collect方法 发送数据
* 3.实现 cancel方法
* 注意事项:
* 1.接口的泛型为 数据源的数据类型
* */
class CustomParallelSource extends RichParallelSourceFunction[String]
// 标志位，用来控制循环的退出
var isRunning = true
override def run(ctx: SourceFunction.SourceContext[String]): Unit =
val list = List("刘备1", "关羽2", "张飞3", "赵云4", "马超5")
while (isRunning)
// 调用 collect 方法向下游发送数据
// 调用 collect 方法向下游发送数据
list.foreach(
e =>
ctx.collect(e)
Thread.sleep(1000)
)
// 通过将 isRunning 设置为false，来终止消息的发送
override def cancel(): Unit = isRunning = false
test("从 自定义并行数据源中 读取数据")
// 1. 获取流执行环境
val env = StreamExecutionEnvironment.getExecutionEnvironment
// 2. 将 自定义数据源 作为数据源
val ds: DataStream[String] = env.addSource(new CustomParallelSource).setParallelism(4)
// 3. 打印DataStream
ds.print().setParallelism(1)
// 4. 出发程序执行
env.execute()
执行结果:
用一个例子告诉你 怎样在spark中创建RDD
目录
1. 前言
众所周知，spark是一种计算引擎(用来计算数据)，但是数据从何而来呢？
spark获取数据主要有两种方式:
方式1:
分发驱动程序中scala集合中的数据
方式2:
分发外部存储系统中的数据(HDFS、HBase、其他共享文件系统)
spark将读来的数据，分发到了哪里去？
spark是一个分布式计算引擎，spark会将读取来的数据
按照指定的并行度，分发到不同的计算节点上去
2. 分发驱动中scala集合中的数据
spark提供了两个方法，用来将本地集合的数据(客户端JVM)切分成若干份
然后再分发到不同的计算节点中去
主要有两个参数:
seq: Seq[T] 本地集合
numSlices: Int 切片数(可选参数，不指定时使用默认切片数)
2.1 parallelize
test("parallelize")
// 初始化 spark配置实例
val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")
// 初始化 spark环境对象
val sc: SparkContext = new SparkContext(sparkconf)
val list = List("java", "scala", "c++", "c#")
// 指定切片数
val rdd1: RDD[String] = sc.parallelize(list, 2)
// 使用默认切片
val rdd2: RDD[String] = sc.parallelize(list)
sc.stop()
2.2 makeRDD
test("makeRDD")
/*
* TODO : 源码中 makeRDD 调用的还是 parallelize方法
*
* */
// 初始化 spark配置实例
val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")
// 初始化 spark环境对象
val sc: SparkContext = new SparkContext(sparkconf)
val list = List("java", "scala", "c++", "c#")
// 指定切片数
val rdd1: RDD[String] = sc.makeRDD(list, 2)
// 使用默认切片
val rdd2: RDD[String] = sc.makeRDD(list)
sc.stop()
2.3 range
def range(start: Long, end: Long,step: Long = 1,numSlices: Int = defaultParallelism): RDD[Long]
功能:
创建一个Long类型的RDD，元素内容为 start到end，公差为step 的等差数列
参数:
start: Long, 起始位置
end: Long, 结束位置,不包括该位置
step: Long = 1, 数列公差，默认为1
numSlices: Int = defaultParallelism 切片数，不指定时使用默认切片数
使用场景:
常用来造数据使用
test("range")
// 初始化 spark配置实例
val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")
// 初始化 spark环境对象
val sc: SparkContext = new SparkContext(sparkconf)
val rdd: RDD[Long] = sc.range(0, 10)
sc.stop()
3. 分发外部存储系统中的数据
spark提供了两个方法，用于将外部文件数据切片后，再分发到不同的计算节点上去
主要有两个参数:
path: String 指定文件系统URL
minPartitions: Int 指定切片数(不指定时，使用默认切片数)
使用限制:
对文件系统的要求:
读取的文件系统必须是 HDFS、本地文件系统、任何hadoop支持的文件系统
对读取文件的要求:
文件格式必须是 text格式且UTF-8
对url的要求:
支持单个文件 /my/directory/1.txt
支持多个文件 /my/directory/*.txt
支持目录 /my/directory (目录下的必须都是文件，不能有目录存在)
java.io.IOException: Path: /dir/dir2 is a directory, which is not supported by the record
只能读取目录下的文件，不会对子目录进行遍历读取
支持gz格式的压缩文件 /my/directory/*.txt
3.1 textFile
返回 RDD[String] 格式的rdd，每个元素内容为 读取到text文件的每行，rdd的长度为所有文件的行数
test("textFile")
// 初始化 spark配置实例
val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")
// 初始化 spark环境对象
val sc: SparkContext = new SparkContext(sparkconf)
// 读取目录下的所有文件
val rdd: RDD[String] = sc.textFile("src/main/resources/data/dir")
// 读取gz格式的压缩文件
//val rdd: RDD[String] = sc.textFile("src/main/resources/data/dir/1.txt.gz")
rdd.foreach(println(_))
sc.stop()
3.2 wholeTextFiles
返回 RDD[(String, String)] 格式的rdd，每个元素内容为 (文件名称,文件内容)，rdd的长度为读取到的文件个数
test("wholeTextFiles")
// 初始化 spark配置实例
val sparkconf: SparkConf = new SparkConf().setMaster("local[4]").setAppName("")
// 初始化 spark环境对象
val sc: SparkContext = new SparkContext(sparkconf)
/*
* TODO data/dir目录下 虽然存在目录不会报错，但是读取时会过滤掉目录,并不会递归读取子目录
* */
val rdd: RDD[(String, String)] = sc.wholeTextFiles("src/main/resources/data/dir")
rdd.foreach(e => println(s"fileName:$e._1 data:$e._2"))
sc.stop()
以上是关于用一个例子告诉你 怎样在Flink DataStream API 中读取数据源的主要内容，如果未能解决你的问题，请参考以下文章
数栈技术分享：用短平快的方式告诉你Flink-SQL的扩展实现
[2] Flink大数据流式处理利剑: 用Flink进行统计的一个简单例子