使用嵌入式 Kafka 测试 Flink
我有一个简单的 Flink 应用程序，它总结了最后一分钟内具有相同 id 和时间戳的事件: DataStream输入 = 环境.addSource(consumerProps).uid("应用程序");数据流events = input.map(record -> mapper.readValue(record, Event.class));像素.assignTimestampsAndWater
..
apache-flink相关内容
使用 Kafka 作为 EventStore 时在 Flink 中恢复状态一致性
问题 我正在将微服务作为事件溯源聚合实现，而该聚合又作为 Flink FlatMapFunction 实现.在基本设置中，聚合从两个 kafka 主题中读取事件和命令.然后，它将新事件写入第一个主题，并在第三个主题中处理结果.因此，Kafka 充当事件存储.希望这张图有帮助: RPC 请求 RPC 结果||~~~~>命令-||--->结果~~~~~~||--> 聚合--|~>输入 evs
..
基于流的应用程序中的受控/手动错误/恢复处理
我正在开发一个基于 Apache Flink 的应用程序，它使用 Apache Kafka 进行输入和输出.可能这个应用程序将被移植到 Apache Spark，所以我也将它添加为标签，问题保持不变. 我要求所有通过 kafka 接收的传入消息必须按顺序处理，并安全地存储在持久层(数据库)中，并且不能丢失任何消息. 这个应用程序中的流部分相当琐碎/小，因为主要逻辑将归结为: envi
..
Kafka - 无法建立到节点 -1 的连接
我正在尝试使用 apache flink 流来使用 kafka 主题.但我遇到了这个问题. 2018-04-10 02:55:59,856|- ProducerConfig 值:确认数 = 1批量大小 = 16384bootstrap.servers = [本地主机:9092]缓冲区.内存 = 33554432客户端.id =压缩类型 = 无连接数.max.idle.ms = 540000ena
..
尝试升级到 flink 1.3.1 时出现异常
我尝试将集群中的 flink 版本升级到 1.3.1(以及 1.3.2)，但在任务管理器中出现以下异常: 2018-02-28 12:57:27,120 错误 org.apache.flink.streaming.runtime.tasks.StreamTask - 处理流操作符时出错.org.apache.kafka.common.KafkaException: java.lang.Inter
..
ClassNotFoundException: org.apache.flink.streaming.api.checkpoint.CheckpointNotifier 在使用 kafka 主题时
我使用的是最新的 Flink-1.1.2-Hadoop-27 和 flink-connector-kafka-0.10.2-hadoop1 jar. Flink 消费者如下: StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();如果(属性 == 空){属性 = 新属性()；I
..
如何在 Flink (Scala) 中反序列化来自 Kafka 的 Avro 消息?
我正在将来自 Kafka 的消息读入 Flink Shell (Scala)，如下: scala>val stream = senv.addSource(new FlinkKafkaConsumer011[String]("topic", new SimpleStringSchema(), properties)).print()警告:有一个弃用警告；使用 -deprecation 重新运行以获
..
在 Scala 中反序列化 Avro 数据时遇到问题
我正在 Scala 中构建一个 Apache Flink 应用程序，它从 Kafka 总线读取流数据，然后对其执行汇总操作.来自 Kafka 的数据是 Avro 格式，需要一个特殊的反序列化类.我找到了这个 Scala 类 AvroDeserializationScehema (http://codegists.com/snippet/scala/avrodeserializationschema
..
如何在窗口流媒体etl中显示中间结果?
我们目前在事件存储中实时聚合数据.这个想法是可视化多个时间范围(每月、每周、每天、每小时)和多个名义键的交易数据.我们经常有迟到的数据，所以我们需要考虑到这一点.此外，要求显示“运行"结果，即当前窗口完成之前的值. 目前我们正在使用 Kafka 和 Apache Storm(特别是 Trident，即微批处理)来做到这一点.我们的架构大致如下: (为我丑陋的照片道歉).我们使用 Mon
..
Flink:将文件与 kafka 流连接
我有一个我真的无法弄清楚的问题.所以我有一个 kafka 流，其中包含一些这样的数据: {"adId":"9001", "eventAction":"start", "eventType":"track", "eventValue":"", "timestamp":"1498118549550"} 我想用另一个值“bookingId"替换“adId".这个值位于一个 csv 文件中，但我真的不
..
Flink Kafka - 如何让应用程序并行运行?
我正在 Flink 中创建一个应用程序 阅读来自某个主题的消息 对其进行一些简单的处理 将结果写入不同的主题 我的代码可以运行，但是它不能并行运行 我该怎么做? 我的代码似乎只在一个线程/块上运行? 在 Flink 网络仪表板上: 应用进入运行状态 但是，概览子任务中只显示了一个块 接收/发送的字节数，接收/发送的记录数始终为零(无更新) 这是我的代码，请
..
当我重新运行 Flink 消费者时，Kafka 再次消费最新消息
我用 Scala 编写的 Apache Flink API 创建了一个 Kafka 消费者.每当我从某个主题传递一些消息时，它都会及时接收它们.但是，当我重新启动使用者时，它不会接收新的或未使用的消息，而是使用发送到该主题的最新消息. 这是我在做什么: 运行生产者: $ bin/kafka-console-producer.sh --broker-list localhost:909
..
如何使用 Avro 和 Flink 解码 Kafka 消息
我正在尝试使用 Flink 1.0.3 从 Kafka 主题中读取 AVRO 数据. 我只知道这个特定的 Kafka 主题有 AVRO 编码的消息，而我有 AVRO 架构文件. 我的 Flink 代码: public static void main(String[] args) 抛出异常 {StreamExecutionEnvironment env = StreamExecuti
..
任何人都可以在 Scala 中分享 Flink Kafka 示例吗?
谁能分享一个 Scala 中 Flink Kafka 的工作示例(主要是接收来自 Kafka 的消息)?我知道有一个
..
卡夫卡&重启时 Flink 重复消息
首先，这和 当我重新运行 Flink 消费者时，Kafka 再次消费了最新的消息，但它不一样.该问题的答案似乎不能解决我的问题.如果我在该答案中遗漏了某些内容，请重新表述答案，因为我显然遗漏了一些内容. 问题是完全一样的——Flink(kafka 连接器)重新运行了它在关闭之前看到的最后 3-9 条消息. 我的版本 Flink 1.1.2卡夫卡 0.9.0.1斯卡拉 2.11.7Ja
..
flink kafka 消费者组 ID 不起作用
我正在使用带有 flink 的 kafka.在一个简单的程序中，我使用了flinks FlinkKafkaConsumer09，给它分配了组id. 根据 Kafka 的行为，当我使用相同的 group.Id 在同一个主题上运行 2 个消费者时，它应该像消息队列一样工作.我认为它应该像这样工作:如果向 Kafka 发送 2 条消息，则每个或其中一个 flink 程序将总共处理 2 条消息两次(
..
卡夫卡&重启时 Flink 重复消息
首先，这和 当我重新运行 Flink 消费者时，Kafka 再次消费了最新的消息，但它不一样.该问题的答案似乎不能解决我的问题.如果我在该答案中遗漏了某些内容，请重新表述答案，因为我显然遗漏了一些内容. 问题是完全一样的——Flink(kafka 连接器)重新运行了它在关闭之前看到的最后 3-9 条消息. 我的版本 Flink 1.1.2卡夫卡 0.9.0.1斯卡拉 2.11.7Ja
..
如何从程序中停止 flink 流作业
我正在尝试为 Flink 流作业创建 JUnit 测试，该作业将数据写入 kafka 主题并分别使用 FlinkKafkaProducer09 和 FlinkKafkaConsumer09 从同一 kafka 主题读取数据.我正在产品中传递测试数据: DataStream流 = env.fromElements("tom", "jerry", "bill"); 并检查来自消费者的数据是否与以下
..
Apache Flink:java.lang.NoClassDefFoundError
我正在尝试遵循此示例但是当我尝试编译它时，出现此错误: 错误:无法初始化主类 com.amazonaws.services.kinesisanalytics.aws引起:java.lang.NoClassDefFoundError: org/apache/flink/streaming/api/functions/source/SourceFunction 错误是由于此代码: 私有静态 D
..
集群中的 Apache Flink 流不会与工作人员拆分作业
我的目标是建立一个使用 Kafka 作为源的高吞吐量集群 &Flink 作为流处理引擎.这是我所做的. 我已经在主节点和工作节点上设置了一个 2 节点集群，以下配置. 掌握 flink-conf.yaml jobmanager.rpc.address: #本地主机jobmanager.rpc.port: 6123jobmanager.heap.mb: 256taskmanager.h
..