目录
1.3 Flink 中的数据类型(TypeInformation)
Flink 版本 : 1.13
开发语言 : Scala 2.12
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]
...
Flink 中的数据源分为 非并行数据源 和 并行数据源 两大类
并行数据源 : 可以将数据源拆分成多个 子任务，并行执行( 并行度允许大于1)
非并行数据源 : 不可以将数据源拆分，只能有单独的任务处理数据 (并行度必须1)
我们可以通过 StreamExecutionEnvironment.addSource(SourceFunction) 将一个 source对象 关 联到编写的 flink应用程序中
Flink API中 自带了许多 SourceFunction的实现类
我们也可以 通过实现 SourceFunction接口 来编写 自定义的非并行的source对象
通过实现 ParallelSourceFunction 接口
继承 RichParallelSourceFunction 类 来编写 自定义的并行source对象
Flink 会将外部的数据加载到 DataStreamSource 对象中，加载过程中会将外部的数据的类型转换为 Flink 定义的数据类型
为了方便 数据序列化和反序列化，Flink定义了自己的数据类型系统
非并行数据源
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()
}思考: 读取文件时可以设置哪些规则呢？
1. 文件的格式(txt、csv、二进制...)
2. 文件的分隔符(按\n 分割)
3. 是否需要监控文件变化(一次读取、持续读取)
基于以上规则，Flink为我们提供了非常灵活的 读取文件的方法
语法说明:
定义:
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()
}语法说明:
定义:
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()
}
语法说明:
语法:
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()
}语法说明:
语法:
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()
}
}
代码示例:
/*
* 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()
}
执行结果:
代码示例:
/*
* 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()
}
执行结果:
我正在寻找执行以下操作的正确语法(在Perl、Shell或Ruby中):#variabletoaccessthedatalinesappendedasafileEND_OF_SCRIPT_MARKERrawdatastartshereanditcontinues. 最佳答案 Perl用__DATA__做这个:#!/usr/bin/perlusestrict;usewarnings;while(){print;}__DATA__Texttoprintgoeshere 关于ruby-如何将脚
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
使用带有Rails插件的vim，您可以创建一个迁移文件，然后一次性打开该文件吗？textmate也可以这样吗？ 最佳答案 你可以使用rails.vim然后做类似的事情::Rgeneratemigratonadd_foo_to_bar插件将打开迁移生成的文件，这正是您想要的。我不能代表textmate。 关于ruby-使用VimRails，您可以创建一个新的迁移文件并一次性打开它吗？，我们在StackOverflow上找到一个类似的问题： https://sta
我需要从一个View访问多个模型。以前，我的links_controller仅用于提供以不同方式排序的链接资源。现在我想包括一个部分(我假设)显示按分数排序的顶级用户(@users=User.all.sort_by(&:score))我知道我可以将此代码插入每个链接操作并从View访问它，但这似乎不是“ruby方式”，我将需要在不久的将来访问更多模型。这可能会变得很脏，是否有针对这种情况的任何技术？注意事项:我认为我的应用程序正朝着单一格式和动态页面内容的方向发展，本质上是一个典型的网络应用程序。我知道before_filter但考虑到我希望应用程序进入的方向，这似乎很麻烦。最终从任何
我想要做的是有2个不同的Controller，client和test_client。客户端Controller已经构建，我想创建一个test_clientController，我可以使用它来玩弄客户端的UI并根据需要进行调整。我主要是想绕过我在客户端中内置的验证及其对加载数据的管理Controller的依赖。所以我希望test_clientController加载示例数据集，然后呈现客户端Controller的索引View，以便我可以调整客户端UI。就是这样。我在test_clients索引方法中试过这个:classTestClientdefindexrender:template=>
好的，所以我的目标是轻松地将一些数据保存到磁盘以备后用。您如何简单地写入然后读取一个对象？所以如果我有一个简单的类classCattr_accessor:a,:bdefinitialize(a,b)@a,@b=a,bendend所以如果我从中非常快地制作一个objobj=C.new("foo","bar")#justgaveitsomerandomvalues然后我可以把它变成一个kindaidstring=obj.to_s#whichreturns""我终于可以将此字符串打印到文件或其他内容中。我的问题是，我该如何再次将这个id变回一个对象？我知道我可以自己挑选信息并制作一个接受该信
如果您尝试在Ruby中的nil对象上调用方法，则会出现NoMethodError异常并显示消息:"undefinedmethod‘...’fornil:NilClass"然而，有一个tryRails中的方法，如果它被发送到一个nil对象，它只返回nil:require'rubygems'require'active_support/all'nil.try(:nonexisting_method)#noNoMethodErrorexceptionanymore那么try如何在内部工作以防止该异常？ 最佳答案 像Ruby中的所有其他对象
关闭。这个问题需要detailsorclarity.它目前不接受答案。想改进这个问题吗？通过editingthispost添加细节并澄清问题.关闭8年前。Improvethisquestion为什么SecureRandom.uuid创建一个唯一的字符串？SecureRandom.uuid#=>"35cb4e30-54e1-49f9-b5ce-4134799eb2c0"SecureRandom.uuid方法创建的字符串从不重复？
我有一个正在构建的应用程序，我需要一个模型来创建另一个模型的实例。我希望每辆车都有4个轮胎。汽车模型classCar轮胎模型classTire但是，在make_tires内部有一个错误，如果我为Tire尝试它，则没有用于创建或新建的activerecord方法。当我检查轮胎时，它没有这些方法。我该如何补救？错误是这样的:未定义的方法'create'forActiveRecord::AttributeMethods::Serialization::Tire::Module我测试了两个环境:测试和开发，它们都因相同的错误而失败。 最佳答案
有时我需要处理键/值数据。我不喜欢使用数组，因为它们在大小上没有限制(很容易不小心添加超过2个项目，而且您最终需要稍后验证大小)。此外，0和1的索引变成了魔数(MagicNumber)，并且在传达含义方面做得很差(“当我说0时，我的意思是head...”)。散列也不合适，因为可能会不小心添加额外的条目。我写了下面的类来解决这个问题:classPairattr_accessor:head,:taildefinitialize(h,t)@head,@tail=h,tendend它工作得很好并且解决了问题，但我很想知道:Ruby标准库是否已经带有这样一个类？ 最佳