动态限制flink kafka源
我们正在使用多个Kafka主题，但希望优先考虑其中一些主题（~服务质量）。
根据我在网上找到的，共识是不要限制运算符，而是限制源，更具体地说是反序列化器[1]。
我们如何访问源中有关流媒体环境状态的信息（即主题落后于当前偏移量的程度）。
目前，我们计划将我们的整个设置转换为CoFlatMaps[2]，并拥有一个控制流，该控制流为所有主题发出当前偏移滞后-低优先级流运算符，然后根据高优先级流的滞后Hibernate。
你将如何解决这个问题？Tl；博士：有没有办法跨任务管理器的源/反序列化器共享信息？
[1]http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/Rate-limit-processing-td11174.html
[2]http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/newbie-Share-state-between-streams-td7368.html
共有1个答案
对于需要回答这个问题的人来说：我在反压闪存中遇到了类似的话题。我发现人们在源代码运算符和序列化部分做了他们的速率限制。
有一个来自flink github repo的例子：https://github.com/apache/flink/blob/master/flink-examples/flink-examples-streaming/src/main/java/org/apache/flink/streaming/examples/utils/ThrottledIterator.java
-
大家好，我正在努力将一个简单的avro模式与模式注册表一起序列化。 设置： 两个用java编写的Flink jobs（一个消费者，一个生产者） 目标：生产者应该发送一条用ConfluentRegistryAvroSerializationSchema序列化的消息，其中包括更新和验证模式。 然后，使用者应将消息反序列化为具有接收到的模式的对象。使用。 到目前为止还不错：如果我将架构注册表上的主题配置
-
有人能详细说明酒吧/子系统实施的动态费率限制吗？我在gcloud文档或faq页面上找不到任何细节。 下面是我的pubsub用法：我计划在我们的生产中使用pubsub。现在，我有1个主题、1个订阅和1个订阅者（Webhook HTTPS回调）。有时我的订阅者可能会抛出异常（非常少），在这种情况下，我的订阅者将返回一个400响应给pubsub，以便pubsub可以保留消息并重试。 如果pubsub从用
-
在应用开发中，经常会有对请求进行限速的需求。 通常意义上的限速，其实可以分为以下三种: limit_rate 限制响应速度 limit_conn 限制连接数 limit_req 限制请求数 接下来让我们看看，这三种限速在 OpenResty 中分别怎么实现。 限制响应速度 Nginx 有一个 $limit_rate，这个变量反映的是当前请求每秒能响应的字节数。该字节数默认为配置文件中 limit_
-
问题内容： 我想编写一个查询，在其中我可以动态地将各种限制传递给SQLA Core。例如，我希望能够在查询中指定或不动态指定。通过限制指令可能会很好，或者可能会有更标准的方法。 我已经阅读了教程和API文档，但我想知道： 对于SQLA Core中的动态限制，什么是惯用合同，它将如何实施？ 问题答案： 您可以毫无问题地动态构建查询。例如，您可以执行以下操作： 那里只是限制的字典，如。这将产生一个查询
-
问题内容： 我想使用Django REST Framework，限制可以在创作的相关字段中使用的值。 例如，考虑以下示例（基于http://django-rest-framework.org/api- guide/filtering.html 上的过滤示例，但已更改为ListCreateAPIView）： 在此示例中，如何确保购买者在创建时只能等于self.request.user，并且这是可浏览
-
基于上面的代码--我能期望发生什么？我在运行时会因为Horse没有实现add（Animal）而失败吗？在上述性质的调用中，被比较的值的编译时类型是动物的（至少，正如所写的那样），但它们的运行时类型更具体，它能准确地区分它们吗？如果我们使用var而不是val，这有什么改变吗？ 提前道谢。 编辑：修改的核心代码--我看到了第一个响应者强调的问题，我没有正确地思考。显然，我还没有真正编译这个，我仍然是在