我将在processBroadcastElement()函数中执行一些状态管理。 final val actvTagsMapValue = new MapStateDescriptor[String, List[String]]("actvTagsMapValue", classOf[String], classOf[List[String]])
override def processBroadcastElement(...): Unit {
val actvTagMap = getRuntimeContext.getMapState(actvTagsMapVal
我是Flink的新手,我正在使用apache flink进行模式匹配,其中模式列表处于广播状态,并迭代通过processElements函数中的模式来找到匹配的模式,我正在从数据库中读取此模式,这是一个准时的活动。下面是我的代码
MapState描述符和端输出流如下
public static final MapStateDescriptor<String, String> ruleDescriptor=
new MapStateDescriptor<String, String>("RuleSet", BasicTypeInfo.STR
让我把它说出来,我是一个非常初学者的Flink,并试图抓住尽可能多的概念。
比方说,我有一个flink集群,其中包含10个任务管理器。我每个人都有一个flink作业在运行。作业也使用广播状态。该广播状态是通过每10分钟读取5个S3文件,进行一些处理,并创建播放的int to list of strings地图来创建的。
问题:在哪里读取文件,是否在JobManager读取和处理文件,并将处理过的内容发送给任务管理人员。
或
是任务管理人员负责所有的读取和处理。如果是这种情况,那么flink如何确保如果任务管理器无法从S3读取,那么所有任务管理器的广播状态都是相同的。
编辑
因此,任务管理器读取
我正在尝试将一个自定义的动态分区程序从Flink 1.7迁移到Flink 1.9。原始分区程序在StreamPartitioner接口中实现了selectChannels方法,如下所示: // Original: working for Flink 1.7
//@Override
public int[] selectChannels(SerializationDelegate<StreamRecord<T>> streamRecordSerializationDelegate,
我正在写一个流Flink程序,为我们的离线训练模型做特征提取,并想知道程序的设计。我希望每个特征提取逻辑在其类中维护自己的状态,以便添加一个新的特征提取将等同于添加一个新的类。 粗略的高层设计如下: #data is the stream of relative paths to the feature extraction logic in our code e.g. com.xxx.FeatureExtraction1
val data:DataStream[String] = ...
#based on the relative path, use reflection to ini
我有一个包含表T1(id,name,age)和T2(id,subject)的数据库。Flink以事件流的形式接收来自数据库的所有更新,使用类似于的内容。这些表是相互关联的,所需数据可以通过连接 T1和T2 on id来提取。目前数据库的全部状态存储在Flink MapState中,其中id作为密钥。现在的问题是,我需要基于名称从T1中选择行,而无需使用id。我似乎需要在T1(name)上建立一个索引,以使它更快。是否有任何方法可以自动索引它,而不必手动为每个表创建索引。这样做的推荐方法是什么?我知道,但我需要对表的更新提供支持。顺便说一下,我在Scala中使用了Flink。如有任何建议或建议,