Apache Flink是一个用于可扩展批处理和流数据处理的开源平台。 Flink在一个系统中支持批量和流分析。分析程序可以用Java和Scala中简洁优雅的API编写。
每次都不能从processElement函数中使用BroadCastState
我想第一次使用BroadCastState。我按照文档中的一个小例子测试了它。我使用KeyedBroadcastProcessFunction并从...更新Map状态
我使用Flink 1.7仪表板并选择流媒体作业。这应该向我展示一些指标，但它仍然需要加载。我在Flink 1.5集群中部署了相同的工作，我可以查看指标。 Flink是......
在Flink中有没有办法使用keyby（key1或key2）（ORing键）我不是指keyBy（0,1）我需要将带有key1的元素或带有key2的元素转发到同一节点。
Apache Flink：如何实现SourceFunction？
我已经实现了一个从URL中获取数据（String）的SourceFunction。然后我正在执行该数据的keyBy（）并应用一个10分钟的窗口。现在SourceFunction只被调用一次而且......
连接ResourceManager for Flink FLIP-6启动新的“会话作业”
我想用会话作业创建一个新的k8s部署;并在flink-conf.yaml中使用这样的配置部署了一个taskmanager：jobmanager.rpc.address：analytics-job jobmanager.rpc ....
如何根据json中的特定键将一个数据流中的接收器添加到不同的路径？
我有jsons喜欢，{“姓名”：“某人”，“工作”：“医生”，“等”：“等等”}在每个json中，对于“工作”有不同的价值，如医生，飞行员，司机，守望者等我想分开每个json ......
Flink Job：获取InvalidClassException：com.fasterxml.jackson.databind.ObjectMapper
我们的flink工作在当地运行得非常好，但在尝试将其部署在纱线集群中时，它会低于错误。 Flink版本：1.5.0 Jackson 2.6.7引起：java.io ....
Flink - 从Cassandra获取数据作为通用ResultSet并将其转换为DataSet
我有来自kafka简单cql选择查询的StreamExecutionEnvironment作业。我尝试使用以下代码异步处理此查询：public class GenericCassandraReader ...
我已经在批处理中使用了apache flink一段时间，但现在我们想将这个批处理作业转换为流式作业。我遇到的问题是如何运行端到端测试。它是如何在批处理中工作的......
如何以byte []格式从Kafka读取数据？我有一个实现，使用SimpleStringSchema（）读取事件为String但我找不到一个架构来读取数据作为byte []。这是我的代码：...
我有一个flink UI仪表板，在概述选项卡下，我看到一个包含以下文本的框。只想了解它的含义。来源：自定义源 - >时间戳/水印 - > FlatMap - > ParDo（...
Apache Flink：如何处理Kafka接收器中的异常？
我有一个Flink工作，将数据写入Kafka。 Kafka主题的最大消息大小设置为5 MB，因此如果我尝试写任何大于5 MB的记录，它会抛出以下异常并带来...
Scala - Flink Monitoring API（上传作业）
美好的一天，我有一个问题使用Scala All将请求上传到Flink API获取请求似乎工作导入scalaj.http._ val url：String =“http://127.0.0.1:8081”val response：HttpResponse [String] = ...
用于保持访问配置数据<10 GB的最佳分布式缓存，并从Flink流应用程序访问每条记录？
我的数据不会超过10 GB，我需要将它放在分布式缓存中并为每条记录访问它以便从我的Flink流应用程序进行验证。哪一个最适合......
我试图以流式和批量方式使用flink，将大量数据添加到Accumulo（每分钟几百万）。我想在将记录发送到Accumulo之前批量记录。我摄取数据......
我需要为flink流式拓扑编写单元测试。它基本上是一个CoFlatMapFunction，它有2个输入。我尝试从这个页面获得一些灵感：https：//ci.apache.org/projects/flink / ...
从Flink文档中我看到有两个不同的窗口对象：timeWindow（Time.seconds（5））和window（TumblingWindow / SlidingWindow）....，我对它们之间的区别感到困惑，...
我创建了一个滑动窗口，并希望递归打包所有元素进入该窗口期间，这是代码的一部分.map（x =>（（x.pickup.get.latitude，x.pickup.get.longitude），（ x.dropoff ....
我想在一个表中的Flink SQL中进行一个简单的查询，其中包含一个group by语句。但是在结果中，group by语句中指定的列存在重复的行。就是它 ...
Flink是否提供了一些API来获取Session Windows中的活动数据？
我创建了一个Flink的会话窗口来跟踪我站点中访问者的URL。并且它从不活动开始5分钟后结束（会话间隙）Flink是否支持一些API来读取活动窗口中的数据...