我使用的是lyft flink operator部署的Flink 1.14 我正在尝试使用Table API使tumble窗口聚合，从transactions表源读取，并将窗口的聚合结果放入新的kafkascan.startup.mode' = 'earliest-offset',\n"+
" 'format' = 'debezium-json'\n&quo
浏览 126提问于2021-11-14得票数 0
回答已采纳
我看不到太多关于kafka flink连接器python的信息。我知道它是beta version.can，我使用kafkaflink连接器API，使用python.If是的，给我推荐一个这个连接器的程序。我也提到了Apache Flink: Kafka connector in Python streaming API, "Cannot load user class"，.Also，我可以使用pythonAPI的滑动窗口的概念吗？
浏览 31提问于2020-01-23得票数 0
目前，我有一个Flink集群，它希望用一个模式来使用Kafka主题，通过这种方式，我们不需要维护一个硬代码Kafka主题列表。import java.util.regex.Pattern;import org.apache.flink.streaming.connectors.kafka.FlinkKafkaCo
浏览 1提问于2019-07-30得票数 1
回答已采纳
我想设置Flink，这样它就可以将数据流从Apache转换并重定向到MongoDB。出于测试目的，我构建在flink流-Connectors.kafka示例()之上。我发现的关于MongoDB集成的唯一例子是来自github的flink-mongodb测试。不幸的是，它使用的是静态数据源(数据库)，而不是数据流。我是否需要编写自定义接收器函数，还是可能遗漏了什么？也许应该用不同的方式来做？下面是一个例子，我作为输入得到了什么，作为输出我需要存储什么。Apa
浏览 3提问于2016-02-02得票数 11
回答已采纳
例如，用户点击按钮'A‘，点击按钮'B’后的一段时间，我想统计在这段时间内有多少事件(来自其他主题)已经到达。我在我的应用程序中面临的一般概念是，我的事件具有持续时间，它们不是在给定时间独立发生的单个事件。在这个例子中，点击按钮'A‘将是事件的开始，点击按钮'B’将是结束。我的问题是，kafka提供的窗口过程(翻滚、跳跃、滑动、会话)不适合我的场景。有没有其他方法可以在Kafka</em
浏览 29提问于2020-11-09得票数 0
1回答
例如，我想测试卡夫卡/ Flink与FlinkKafkaConsumer011和FlinkKafkaProducer011的集成。这一进程将是：
对于一个字符串示例，它将是，从输入主题读取字符串，转换为大写，写入一个新主题。
浏览 0提问于2018-04-16得票数 3
据我所知，处理Kafka流的状态在内存、磁盘或Kafka主题中都是本地的，因为所有的输入数据都来自一个分区，其中所有的消息都是由一个定义的值键控的。大多数情况下，计算可以在不知道其他处理器的状态的情况下完成。如果是这样的话，您有另一个Streams实例来计算结果。如图所示：Flink到底在哪里存储它的状态？Flink是否也可以在本地存储状态，或者总是将它们发布到所有实例(任务)？是否可以将Flink配置为将状态存储在<em
浏览 0提问于2019-02-10得票数 3
我使用Flink 1.11 (通过Python和Anaconda虚拟环境)，并将Kafka作为我的源和接收器。我将我的Flink作业提交到一个集群中。所有的一切都是在码头(当地)进行的。由于我对它还不熟悉，现在我已经设置了它，因为它实际上是作为传递--使用一些窗口，然后慢慢地构建起来的。从这里开始，这很好，在新的Kafka主题中出现了一些信息：
table.insert_into("s
浏览 3提问于2020-08-28得票数 0
我对Flink相当陌生，在技术用例方面需要一些帮助。我有一个在GKE上运行的flink应用程序，并使用自定义接收器将记录(从Kafka源)写入BigQuery。目前，记录被一个接一个地写入接收器中，因此每个kafka消息都会得到自己对BigQuery的插入api调用，这并不理想，因为我们需要执行批量插入，并且单独插入每个记录将非常昂贵。新要求：
在将记录写入BigQu
浏览 5提问于2022-10-18得票数 2
回答已采纳
我正在专门评估Flink的流式窗口支持，以支持可能的警报生成。我关心的是内存的使用，所以如果有人能在这方面提供帮助，我将不胜感激。producer -> kafka -> flinkkafkaconsumer -> table.window(Tumble.over("
浏览 0提问于2017-11-04得票数 2
我有以下简单的代码，执行基于处理时间的翻滚窗口，与表api，但异常抛出时，我运行它。我不知道它在说什么，有人能帮我看看吗?谢谢！import org.apache.flink.streaming.api.scala._import org.apache.flink.<
浏览 85提问于2021-01-05得票数 0
回答已采纳
1回答
我的数据位于100个kafka分区中，在将它们发送到自定义HTTP接收器之前，我需要通过map操作符传递事件。 出于性能原因，需要将事件批处理成n个消息组，理想情况下，批处理应该在几秒钟后刷新。如果事件计数满足我的最大批处理大小，那么使用带有清除触发器的翻滚处理时间窗口似乎是一个很好的解决方案。 问题是我需要将这个任务扩展到每秒处理100k个事件。为了获得有窗口的批处理，Flink希望我使用key by语句。我尝试过使用一些不同的键，但遇到了平衡问题，
浏览 9提问于2021-11-12得票数 0
我有一个带有以下Avro IDL的Kafka主题，并注册到Schema Registry。现在，当我使用STRING作为类型时，我在Flink-1.13.1中得到了以下消息： *Caused by: java.io.IOException: Failed to deserialize(DeserializationSchema.java:82) org.apache.flink.streaming.connectors.kafka
浏览 190提问于2021-10-12得票数 1
回答已采纳
我有apache访问日志文件，我想在cassandra表中存储每个页面的访问计数(总数/每天/每小时)。
我正在尝试使用kafka连接从日志文件流到kafka主题。为了在Cassandra中增加指标计数器，我可以再次使用Kafka Connect吗？否则，这里应该使用哪些其他工具，例如kafka streams，spark，flink，kafka connect等？
浏览 0提问于2017-10-11得票数 0
我正在使用Flink TableApi和sql来计算每个翻滚窗口中的不同用户，但结果仅在早于水印的时段内可见。有没有可能，在仍然可以改变的时间段内获得“部分”结果？我的代码(在Flink SQL中运行) CREATE TABLE KafkaTable `user_id` BIGINT,
`event_ts` TIMESTAMP(3),, window_end; 在kafka中，
浏览 20提问于2021-11-09得票数 0
2回答
我正在构建一个flink应用程序，它从kafka主题读取数据，应用一些转换并写入Iceberg表。我从kafka主题(在json中)读取数据，并使用circe将其解码为scala类，其中包含scala选项值。对datastream的所有转换都工作得很好。table.printSchema()co
浏览 17提问于2022-02-20得票数 1
回答已采纳
我有一个Flink作业，它在本地运行得很好，但当我尝试在集群上运行flink作业时失败。当尝试通过'connector‘= ' Kafka’从kafka加载数据时发生错误。我使用Flink-Table API和confluent-avro格式从Kafka读取数据。所以基本上我创建了一个从kafka主题中读取数据的表： val inputTableSQL =
s
浏览 299提问于2021-10-22得票数 0
回答已采纳
1回答
我的Flink代码的结构是:使用kafka获取数据(topic_1_in)、->反序列化消息、->地图、->、操作数据->、获取POJO ->序列化消息->、用kafka发送数据(topic_1_out我在Flink网站上找到了以下例子：
但是，我仍然有点困惑如何将我的POJO转换成一个字符串来填充Kafka接收器。这个类非常简单，所
浏览 3提问于2021-01-10得票数 0
回答已采纳
在Flink1.11中，我尝试了debezium-format，下面的代码应该可以工作，对吧？我在试着跟着文档1走 "CREATE TABLE products (\n"name || products |1 row in set
Exception in thread "main&qu
浏览 2提问于2021-03-31得票数 0
云服务器
ICP备案
云直播
即时通信 IM
实时音视频
腾讯云开发者
