如何从Kafka源中模拟Apache光束读取中的事件延迟
从Kafka源中模拟Apache Beam读取中的事件延迟，可以通过以下步骤实现：
- 确保已经安装并配置了Kafka和Apache Beam的相关环境。
- 创建一个Kafka主题(topic)，用于模拟事件数据的产生和传输。
- 编写一个生产者程序，使用Kafka的Producer API将模拟的事件数据发送到Kafka主题中。可以使用任何编程语言来实现，例如Java、Python等。在生产者程序中，可以设置事件的延迟时间，通过控制发送事件的频率来模拟延迟。
- 编写一个消费者程序，使用Apache Beam的KafkaIO读取Kafka主题中的事件数据。同样，可以使用Java、Python等编程语言来实现。在消费者程序中，可以通过记录事件的到达时间和处理时间来计算事件的延迟。
- 运行生产者程序，开始模拟产生事件数据并发送到Kafka主题。
- 运行消费者程序，使用Apache Beam读取Kafka主题中的事件数据，并计算事件的延迟。
通过以上步骤，可以模拟从Kafka源中读取事件数据时的延迟情况。可以根据实际需求调整生产者程序中的延迟时间和事件频率，以及消费者程序中的处理逻辑，来进一步了解和优化事件延迟的情况。
腾讯云相关产品推荐：
- 腾讯云消息队列 CKafka：提供高可用、高可靠、高吞吐量的消息队列服务，适用于大规模数据流转和实时计算场景。详情请参考：CKafka产品介绍
- 腾讯云流计算 Flink：基于Apache Flink的流式计算服务，支持实时数据处理和分析。详情请参考：腾讯云流计算 Flink
- 腾讯云云原生容器服务 TKE：提供高度可扩展的容器化应用管理平台，可用于部署和管理Apache Beam等容器化应用。详情请参考：腾讯云云原生容器服务 TKE
相关·内容
我正在尝试调整我的流式光束管道中的窗口参数。我正在修改的参数有withAllowedLateness、触发器、间隔、窗格触发等。然而，我不知道如何在我的Kafka消费管道中触发延迟来测试更改。有人能建议如何创建事件延迟吗？ 谢谢
浏览 14提问于2021-10-25得票数 0
我在GCS或其他受支持的文件系统上有一个目录，外部进程正在向其中写入新文件。
我想写一个Apache光束流管道，它持续监视这个目录中的新文件，并在新文件到达时读取和处理每个新文件。这个是可能的吗？
浏览 0提问于2017-12-20得票数 6
1回答
我正在尝试使用apache光束将来自kafka消费者的消息流式传输到30秒窗口。使用beam_nuggets.io从卡夫卡主题中读取。
你可以在下面看到我的代码：
with beam.Pipeline(options=PipelineOptions()) as p:
consumer_message = (p | "Reading messages from Kafka" >> kafkaio.KafkaConsume(consumer_config=consumer_config)
| 'window
浏览 8提问于2021-10-20得票数 0
1回答
如何使用Apache Kafka生成/消费延迟消息？似乎标准的Kafka (和Java kafka客户端)功能没有这个功能。我知道我可以用标准的等待/通知机制自己实现它，但它似乎不是很可靠，所以任何建议和好的实践都是值得感谢的。
找到了，但没用。正如我所看到的: Kafka基于从文件系统的顺序读取，并且只能用于读取主题，简单明了地保持消息顺序。我说的对吗？
浏览 3提问于2017-12-27得票数 10
回答已采纳
我需要延迟处理或发布文件名(文件)。我在找最好的选择。
目前，我有两个阿帕奇光束数据流和PubSub之间。第一个数据流从源文件读取文件名，并将这些文件名推送到PubSub主题。另一个数据流读取并处理它们。但是，我的用例是在源代码中创建实际文件后至少1小时开始处理/读取这些文件。
所以我有两个选择：
1)延迟发布消息，以便立即处理它，但要在好的/预期的时刻进行处理
2)延迟处理检索到的文件
就像上面提到的，我正在寻找最好的解决方案。我不确定是否应该在Apache光束中使用芭乐重试机制？还有其他想法吗？
浏览 0提问于2018-12-21得票数 0
1回答
我在Apache光束上实现了一个流作业，它从Apache Kafka读取消息，处理它们，并将它们输出到BigTable中。 我想要获得此作业内部入口/出口的吞吐量指标，即该作业正在读取多少msg/sec以及它正在写入多少msg/sec。 看一下图形可视化，我看到有吞吐量指标，例如，看看下面的示例性图片以进行演示 ? 然而，从文档来看，它在Stackdriver上是不可用的。 是否有任何现有的解决方案来获取此指标？
浏览 14提问于2019-01-04得票数 0
我使用过Debezium作为Mysql的-> Elasticsearch。现在的问题是，当我从MySQL中删除数据时，它仍然会在Elasticsearch中重新出现，即使数据不再存在于MySQL DB中。UPDATE和INSERT工作正常，但DELETE不是。另外，我做了以下工作：
Delete在MySQL中的数据
Delete Elasticsearch索引和ES Kafka Sink
Create -- Kakfa中ES的新连接器
现在，奇怪的是，所有我删除的数据收割机在这里，以及！当我在步骤(3)之前检查ES数据时，数据不存在。但之后，这种行为会被观察到。
请帮我解决这个问题！
M
浏览 7提问于2021-03-14得票数 0
4回答
同步来自多个数据源的数据
、、、、
我们的团队正在尝试构建一个预测维护系统，其任务是查看一组事件，并预测这些事件是否描述了一组已知的异常。
我们正处于设计阶段，目前的系统设计如下：
这些事件可能发生在IoT系统的多个源(如云平台、边缘设备或任何中间平台)上。
事件由数据源推入消息队列系统(目前我们选择了Apache )。
每个数据源都有自己的队列(Kafka主题)。
从队列中，数据由多个推理机(实际上是神经网络)消耗。
根据特征集的不同，推理引擎将订阅多个Kafka主题和来自这些主题的流数据，以持续输出推理。
总体架构遵循单一责任原则，这意味着每个组件将彼此分离，并在单独的Docker容器中运行。
浏览 0提问于2019-05-27得票数 8
回答已采纳
我是新手，不知道如何完成这项任务。我想用Java编写一个使用Apache光束的分区拼图文件。数据是从Kafka读取的，我希望文件每小时有一个新的分区。时间戳列存在于数据中。
浏览 16提问于2021-10-11得票数 0
我已经实现了一个来自Kafka的光束管道读取，基于这里的文档：
管道本身对于有界源工作得很好，并且我有测试用例，它可以从文件中读取数据而不会出现问题。
从Kafka中读取代码非常简单，与示例基本相同：
PCollection<String> input = p.apply(KafkaIO.<Long, String>read()
.withBootstrapServers(KAFKA_BROKER)
.w
浏览 1提问于2017-04-04得票数 3
1回答
Apache通过从检查点恢复作业来保证在故障和恢复时处理一次，检查点是分布式数据流和操作符状态的一致快照(Chandy-Lamport算法用于分布式快照)。此保证在故障转移时恰好是一次。
在正常的集群操作中，Flink如何保证只处理一次，例如给定从外部源读取的Flink源(例如Kafka)，Flink如何保证事件从源读取一次？事件源和Flink源之间是否存在某种类型的应用程序级别？另外，Flink如何保证事件准确地从上游运营商到下游运营商之间只发生一次？对于接收到的事件，这也需要任何形式的附加条件吗？
浏览 3提问于2020-09-30得票数 3
回答已采纳
我正在努力建立一个数据摄取管道使用Apache光束"go“SDK。 我的流程是使用Kafka队列中的数据，并将数据持久化到Google Cloud Bigtable (和/或另一个Kafka主题)。 到目前为止，我还没有找到用"go“编写的Kafka IO连接器(也称为Apache I/O转换)(但是我能够找到java版本)。 下面是支持的Apache Beam内置I/O转换的链接：https://beam.apache.org/documentation/io/built-in/ 我正在寻找与以下Java代码等效的"go“： pipeline.apply
浏览 9提问于2020-01-25得票数 1
是否有人使用Apache Kafka在每个服务都有自己的数据库的微服务体系结构中维护数据完整性？我一直在搜索，有一些帖子提到了使用Kafka，但我正在寻找更多的细节，比如Kafka是如何使用的。您是否必须为生产者和消费者编写代码(例如，对于作为生产者的客户数据库和作为消费者的订单数据库，以便如果在客户数据库中删除了一个客户，那么订单数据库需要以某种方式知道这一点，这样它也会删除该客户的所有订单)。
浏览 1提问于2019-12-06得票数 1
我有一个从Kafka源读取的光束管道(AppendLogTime不可用)，时间戳嵌入到Kafka流的数据对象中。 我想为我的管道使用事件时间，在谷歌了一下之后，我发现一些解决方案在从KafkaIO读取数据时使用了CustomeFieldTimePolicy (提取数据对象的时间戳并使用它来设置水印)。 但是，我看到了另一种解决方案，即使用WithTimestamps.of()将时间戳分配给元素。 我的问题是，这两种方法有什么不同？因为在我看来，他们做的工作非常相似。 谢谢。
浏览 17提问于2021-01-06得票数 0
回答已采纳
我使用的版本是confluent kafka 5.1.14
我试图运行Kafka连接器Elasticsearch将数据从Kafka发送到Elasticsearch，我已经在独立模式下测试了这个配置，一切正常。以下是elasticsearch接收器的独立配置
name=elasticsearch-sink-standalone
connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
tasks.max=1
topics.regex=^[a-zA-Z0-9]((?!-raw$)[a-zA-Z0-9-
浏览 0提问于2020-08-04得票数 0
1回答
我有一个用例，我想使用Apache Beam和Spark runner以批处理模式从Kafka读取数据。 使用KafkaIO类的withMaxNumRecords(long)方法，可以从UnboundedReader生成BoundedReader。但我发现，在批处理模式中，数据首先从每个分区读取，放入内存，然后传递给下一个操作(map、filter等)。 我在每个分区中都有大量数据，在批处理模式下读取这些数据时，我得到了OOM错误。我试着增加执行器的内存。但对于每次运行，我不能将此参数配置为所需的值。另一件事是，我能够以流模式读取相同的数据。 我认为之所以会发生这种情况，是因为在批处理模式下
浏览 17提问于2019-12-11得票数 0
我正在从批处理管道中的有界源csv文件读取数据，并希望根据存储为csv文件中列的数据为元素分配一个时间戳。我如何在Apache光束管道中做到这一点？
浏览 0提问于2019-02-01得票数 2
我在使用flink- connector -kafka-0.9连接器运行Flink时出现以下错误。
Exception in thread "main" java.lang.ClassCastException: org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer cannot be cast to org.apache.flink.streaming.api.functions.source.SourceFunction
at com.dataartisans.ReadFromKafka.main
浏览 2提问于2016-06-29得票数 1
我愿意使用Apache光束从URL而不是文件中获取数据输入。我找不到它的一些内置方法。有什么办法可以做到吗？
浏览 12提问于2020-03-04得票数 0
回答已采纳
我正在做一些关于光束SQL的实验。我从转换SampleSource中获得一个PCollection<Row>，并将其输出传递给一个SqlTransform。
String sql1 = "select c1, c2, c3 from PCOLLECTION where c1 > 1";
下面的代码运行时没有任何错误。
POutput it = p.apply(new SampleSource()).apply(SqlTransform.query(sql1));
p.run().waitUntilFinish();
但是，当我尝试以下几行代码时，我得到了一个运
浏览 7提问于2019-07-02得票数 0
我有一个Kafka Streams应用程序，它正在从一个Store Changelog中读取，它偶尔会抛出这个错误：
org.apache.kafka.clients.consumer.OffsetOutOfRangeException: Offsets out of range with no configured reset policy for partitions: {topic-partition=offset}
at org.apache.kafka.clients.consumer.internals.Fetcher.parseCompletedFetch(Fetcher
浏览 1提问于2018-06-26得票数 1
回答已采纳
我是Apache flume 的新手。对于其中一个用例，我需要将数据从一个集群上的Kafka主题(引导: bootstrap1，主题: topic1)移到不同集群中具有不同名称的主题(引导: bootstrap2，主题: topic2)。有另一个用例，在同一个项目中，最适合水槽，我需要使用相同的水槽管道为这个用例-虽然有其他的选择从卡夫卡复制到卡夫卡。
我尝试了下面的吐露和结果，如在每个选项中提到的。
#1: telnet到kafka接收器(bootstrap2，topic2) ->工作完美。吐露：
a1.sources = r1
a1.sinks = k1
a1.channels =
浏览 5提问于2021-03-13得票数 0
我希望将多个数据库的更改聚合到一个数据库中，因此我想在每个数据库旁边运行一个Debezium连接器和一个Kafka服务器/broker，并使用Kafka接收器连接器从所有的Kafka中使用，将其写入一个数据库。
问题是，我能否同时使用Kafka接收器连接器的一个实例，使用多个独立的Kafka代理(而不是集群)。
浏览 0提问于2021-08-29得票数 0
回答已采纳
假设我们正在KStream和KTable之间进行内部连接，如下所示：
StreamsBuilder sb = new StreamsBuilder();
JsonSerde<SensorMetaData> sensorMetaDataJsonSerde = new JsonSerde<>(SensorMetaData.class);
KTable<String, String> kTable = sb.stream("sensorMetadata",
Consum
浏览 2提问于2020-06-06得票数 0
回答已采纳
我有一个弹簧引导项目，我是spring连接到底层的kafka事件中心。
在同一个消费类中，我必须听两个不同的主题。我有两种方法可以这样做。
一个是有两个卡夫卡听众这样的：
@KafkaListener(topics = "topic1")
public void consumeTopic1(String message) throws Exception {
//do something
}
@KafkaListener(topics = "topic2")
public void consumeTopic2(String message)
浏览 1提问于2019-05-08得票数 0
我目前正在构建一个应用程序，我想使用微服务作为模式和GraphQl进行通信。我正在考虑使用卡夫卡/狂犬病+ authZ + auth0 +阿波罗+普里斯马。所有这些都是在码头上运行的。我发现了许多关于事件来源、优势/前瞻的资源，我被困在它如何在现实世界中工作的问题上。就目前而言，我将这样做：
阿波罗引擎监测请求/响应。
用于认证管理的Auth0
AuthZ授权
graphql网关。遗憾的是，我没有找到一个可靠的解决方案，我想我必须使用阿波罗+graphql工具来合并模式。
理想的情况是：
Prisma为bill's MS的读边
用于bill的M
浏览 1提问于2018-05-31得票数 0
回答已采纳
2回答
我们正在开放源码Apache连接器之上与mongodb kafka connetor合作，以获取从Mongo到HDFS的json数据。我们有kafka使用者，它读取kafka中的数据更改并将它们写入hdfs文件。
我们希望在特定时间安排不同时间的源连接器。
我们需要根据预定日期触发卡夫卡消息。
浏览 9提问于2022-09-23得票数 0
回答已采纳
2回答
我有两个流： topicA -> processingA -> topicB -> processingB -> topicC 登录我的应用程序时，我注意到在将processingA的输出发送到topicB和从topicB为processingB挑选消息之间，每次花费的时间超过100ms (相当于150ms)。它可能不是很多，但它会积累起来，最后相当简单的级联处理几乎只需要几秒钟。 我可以调优kafka使这些延迟尽可能接近于零吗？哪些配置参数对这些延迟有影响？ 我有大部分的默认配置。是不是commit.interval.ms导致了延迟？我已经将其从更高的默认值更改为.
浏览 14提问于2019-06-11得票数 5
1回答
应用服务器日志进程
、、、、
我的客户端需要为部署在集群上的基于后端REST的App服务器处理应用程序(Tomcat)服务器日志文件。Clint希望从这些具有不同参数的数据中生成“访问”和“频率”报告。
我最初的计划是使用kafka从应用服务器日志中获取这些数据-->使用kafka推送到Spark Streaming并处理这些数据-->将这些数据存储到HIVE -->使用zeppelin来获取那些经过处理和集中的日志数据，并根据客户需求生成报告。
但是据我所知，Kafka没有任何功能可以读取日志文件中的数据，并将它们发布在Kafka broker中，在这种情况下，我们已经编写了一个调度程序作业进程，它将读
浏览 31提问于2017-07-24得票数 0
2回答
我真的很喜欢活动采购的想法。我的主要优势是：
如果您构建的是微服务，而不是使用事件源，那么就很容易进行交流。您的组件是解耦的，他们所需要做的就是知道事件存储在哪里。
你知道最简单的事件商店是什么？我只想存储在我的应用程序中发生的事件，并让其他组件来查看这些新事件。
我在用scala
我有过使用apache的经验，有很多库可以阅读kafka的主题。卡夫卡河)
Apache是一个集群系统。很难部署，设置，这对我来说是最困难的部分。我希望构建应用程序，并使用服务逻辑，而不是设置kafka集群。我听说了vertx，它是事件总线，但我还没试过
浏览 1提问于2018-11-10得票数 1
1回答
我想按照它们的时间戳顺序处理多个事件，这些事件通过MQ、S3、KAFKA等多个源系统进入系统。应该采取什么办法来解决这个问题？
浏览 5提问于2022-09-14得票数 1
2回答
卡夫卡流如何用于事件源？
、、、、
我读到了如何通过使用Apache作为事件代理来实现事件源。()
如果我们看一下这张图片，它会显示事件是如何被写入Kafka的，然后Kafka流被用来在数据库中创建视图。
我在这里的问题是，我们如何才能使用卡夫卡流实现这一目标？如果我正确的话，它是一个客户端库，所以我们需要使用它的东西，比如一个名为“聚合服务”的微服务。这是实现这种设计的正确方法吗？它的规模会很好吗？
浏览 8提问于2022-08-01得票数 0
2回答
我想要创建一个可伸缩的分布式应用程序，它使用物化视图而不是传统的数据库。
您能告诉我物化get的视图与选择的按id相比有多高的效率吗？我担心使用REST的不同实例之间的这种“跳转”会大大减缓它们的速度。
org.apache.kafka.streams.state.HostInfo hostInfo = interactiveQueryService.getHostInfo("store-name",
key, keySerializer);
if (interactiveQueryService.getCurrentHostI
浏览 14提问于2022-06-09得票数 0
回答已采纳
我需要将MySQL数据库复制到PostgreSQL数据库。我选择了：
Debezium连接
Avro格式
合流模式注册表
卡夫卡
数据正在被复制，但是，我正在丢失一些模式信息。例如，在Postgres中将具有mysql格式的datetime格式的列复制为bigint，没有创建外键，也没有保留列的顺序(这很好)，等等。
PostgreSQL接收器连接器：
{
"name": "jdbc-sink-dbt",
"config": {
"connector.class": "
浏览 1提问于2022-01-14得票数 0
回答已采纳
我使用官方的Kafka RabbitMQ连接器从RabbitMQ实例获取JSOn消息。 目前，此连接器的配置如下 name=RabbitMQConsentConsumer
connector.class=io.confluent.connect.rabbitmq.RabbitMQSourceConnector
rabbitmq.host=HOST
rabbitmq.port=PORT
rabbitmq.username=USER
rabbitmq.password=PASSWORD
rabbitmq.virtual.host=vhost
tasks.max=1
kafka.topic=TOP
浏览 28提问于2019-04-25得票数 2
1回答
了解flink保存点和检查点
、、、、
考虑具有如下管道的Apache流应用程序：
Kafka-Source -> flatMap 1 -> flatMap 2 -> flatMap 3 -> Kafka-Sink
其中，每个flatMap函数都是一个无状态运算符(例如，Datastream的正常.flatMap函数)。
当传入消息在flatMap 3上挂起时，检查点/保存点是如何工作的？消息是从flatMap 1开始重新启动后重新处理，还是跳到flatMap 3
我有点困惑，因为似乎将应用程序状态称为我可以在有状态运算符中使用的东西，但我的应用程序中没有有状态运算符。是否保存和恢复了“处理过程”，还是在失败
浏览 2提问于2019-03-04得票数 0
2回答
在Apache光束中使用已修复的窗口。水印由事件时间设置。
某些数据可能会乱序到达，并导致窗口关闭。
在Java中，触发器如何定义为在看到最后一个数据后2分钟发生？
浏览 1提问于2019-07-05得票数 0
我想将JSON数据从Azure blob存储加载到apache Kafka中，然后再加载到另一个数据库中。我试着使用没有编写任何python代码的，但我认为在读取源连接器之前，必须使用接收器连接器写入blob，如果我没有弄错的话。将json数据加载到kafka的最好方法是什么？我的要求是，kafka将持续侦听上传到azure blob stored的文件，并将其自动存储在Kafka主题中。
浏览 6提问于2021-11-24得票数 0
我使用星火2.1.0和Kafka 0.10.2.1。
我编写了一个Spark应用程序，它从Kafka主题读取数据集。
守则如下：
package com.example;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.Dataset;
import
浏览 8提问于2017-06-29得票数 1
回答已采纳
1回答
我们有一个Flink应用程序，可以按键对2个Kafka主题执行基于窗口的连接。加入配置如下： window-type: TumblingWindow
window-duration: 10s
allowed-lateness: 10s 因此，当我们将流设置为从earliest偏移量开始时，问题就发生了。似乎窗口边界仍然是基于系统时钟设置的，因此拒绝最早的事件，因为它们依赖于Kafka retention-period可能像14 days一样古老。 有没有一个建议的方法来解决这个问题，或者我的理解上有差距。
浏览 10提问于2020-08-19得票数 0
2回答
我们正在构建一个使用火花流和Kafka的容错系统，并且正在测试检查点火花流，如果火花作业因任何原因崩溃，我们可以选择重新启动它。下面是我们的火花过程：
火花流每5秒运行一次(幻灯片间隔)，从kafka读取数据。
卡夫卡每秒收到大约80条信息。
我们想要实现的是一个设置，在这里我们可以降低火花流作业(以模拟故障)，然后重新启动它，并仍然确保我们处理来自Kafka的每一条消息。这似乎很好，但是，下面是我所看到的，我不知道该做什么：
在我们重新启动星火作业之后，将为所有丢失的时间创建一个批处理。因此，例如，如果我们在一分钟后恢复并重新启动，将创建12批(每5秒一批)。请看下面的图
浏览 1提问于2016-10-06得票数 2
1回答
我已经启动了zookeeper，kafka服务器，kafka生产者和kafka消费者，我已经把从confluent下载的jdbc sql连接器jar放在路径中，我已经在connect-standalone properties.and中提到了plugin.path我已经运行了connect-Standalone.BAT....\config\connect-Standalone.properties...\config\sink quickstart-MYSQL.properties没有任何错误，但是它有很多警告并且没有开始，但是我的数据没有反映在tables.what中我错过了，你能帮我吗
浏览 21提问于2020-10-28得票数 0
我配置了一个Kafka JDBC Source连接器，以便推送来自PostgreSQL数据库的记录更改(插入或更新)的Kafka主题。我使用"timestamp+incrementing“模式。似乎工作得很好。我没有配置JDBC接收器连接器，因为我使用的是一个监听主题的Kafka Consumer。
关于这个主题的消息是一个JSON。这是一个示例：
{
"schema": {
"type": "struct",
"fields": [
{
"type": &#
浏览 1提问于2020-10-01得票数 0
我目前正在使用将卡夫卡消息复制到S3，并且在处理后期数据时出现了OutOfMemory错误。
我知道这看起来是个很长的问题，但我尽力让它清晰易懂。我非常感谢你的帮助。
高层次信息
连接器对Kafka消息做一个简单的字节到字节的复制，并在字节数组的开头添加消息的长度(用于解压缩目的)。
这是CustomByteArrayFormat类的角色(参见下面的说明)
根据Record时间戳对数据进行分区和存储。
CustomTimeBasedPartitioner扩展了io.confluent.connect.storage.partitioner.TimeBasedPa
浏览 1提问于2018-06-21得票数 21
回答已采纳
我创建了一个简单的Apache光束流管道，它从Kafka读取数据，进行一些处理，并通过调用一些外部服务的API来持久化结果。我希望确保在管道重启或失败期间不会丢失数据，因此我希望在特定doFun执行结束时成功调用接口后，手动将记录偏移量提交给Kafka。 在我之前的Kafka经验中，我知道通过使用Kafka Consumer的below接口，我可以手动将记录偏移量提交给Kafka。 consumer.commitSync(currentOffsets); 在KafkaIO设置中有关闭自动提交的设置，但是我没有找到任何有效的解决方案来手动提交偏移量，因为似乎没有办法在doFun中访问消费者。
浏览 10提问于2021-09-21得票数 1
回答已采纳
1回答
我是用卡夫卡连接码头形象。
我的线人
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{"name": "redissource","config": {"connector.class": "com.redis.kafka.connect.RedisSourceConnector", "t
浏览 19提问于2022-09-02得票数 -1
1回答
触发异步更新。
、、
在书中红色，在行动中火花如下：
“但是，Spark不适合对共享数据进行异步更新(例如在线事务处理)，因为它的创建考虑了批处理。(星火流仅仅是批处理应用于时间窗口中的数据。)专门针对这些用例的工具仍然是必要的。”
有人能解释一下它的意思吗？
我感兴趣的是使用火花来执行一些ETL过程。作为旁注，我打算在中间使用卡夫卡。虽然我不明白这个问题。因为从Kafka获取数据并将其写入数据库在某种程度上是同样的问题。这将以平行方式进行。
浏览 1提问于2016-11-15得票数 1
回答已采纳
我想知道，当流道是一个KafkaIO时，是否有人有过在Beam SparkRunner组件中限制速率的经验。我使用的版本是:Beam 2.29，Spark3.2.0和Kafka client 2.5.0？
我把光束参数maxRecordsPerBatch设置为一个很大的数，100000000。但是，即使管道停止45分钟，这个值也不会被击中。但当数据高于正常水平时，卡夫卡的滞后性就会增加，直到最终赶上。在SparkUI中，我看到参数batchIntervalMillis=300000 (5分钟)没有到达，批处理最多需要3分钟。看起来KafkaIO在某个时候停止了阅读，即使滞后很大。我的Kafka
浏览 7提问于2022-03-14得票数 0
回答已采纳
1回答
流在Hadoop中的应用
、、、、
我正在尝试学习的例子。在这个例子中，我确实看到他们创建了一个单独的应用程序(类似于流应用程序)，该应用程序生成和消费数据，并在数据上应用模式匹配。他们现在还没有在.Till(如卡夫卡)之间放置一个流层，单应用程序就足以满足这一目的，这使得它非常优化。现在，我知道，如果我使用Kafka，那么我需要两个应用程序；一个用于将数据摄取到Kafka主题中，另一个用于消费Kafka主题的数据。我有几个问题我没有得到回答：
为什么在这个例子中他们没有使用任何流层(比如Kafka) ?？
什么时候?在哪里需要流？？
参考Flink CEP的例子，我想知道流层(如Kafka/Kinesis)将如何发
浏览 2提问于2016-06-13得票数 0
我正在尝试使用Samza Runner从here运行单词计数演示。这是我的build.gradle plugins {
id 'eclipse'
id 'java'
id 'application'
// 'shadow' allows us to embed all the dependencies into a fat jar.
id 'com.github.johnrengelman.shadow' version '4.0.3'
}
mainClassName =
浏览 23提问于2020-07-12得票数 3
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
