2回答
我们需要在第一个事务使用数据后将其保留在流中。我们该怎么做呢？第二个事务将再次消费并偏移流。需要为第二个DML的解决方案，以查看第一个DML完成后的流数据。两个DML都需要流中的相同数据。
浏览 13提问于2021-10-05得票数 2
我试图使用以下方法从我的星火流工作到Kafka进行抵消：正如我从这个问题中得到的：
这很好，补偿也在进行中然而，问题是，这是异步的，这意味着，即使在又发送了两次偏移提交之后，Kafka可能仍然保留之前的<
浏览 0提问于2019-04-24得票数 4
回答已采纳
我有一个火花结构化的流媒体工作，它需要读取kafka主题中的数据并进行一些聚合。作业需要每天重新启动，但是当它重新启动时，如果我设置了startingOffsets="latest"，就会丢失在重新启动之间的数据。如果我设置了startingOffsets="earliest"，那么作业将从主题读取所有数据，而不是从上一个流作业离开的位置读取数据。有人能帮助我如何配置，以设置偏移的权利，在最后一
浏览 1提问于2019-05-15得票数 1
回答已采纳
我正在使用谷歌数据流中的光束KafkaIO源运行作业，但无法找到一种简单的方法来在作业重新启动时保留偏移量(作业更新选项不够，我需要重新启动作业)
将Beam的KafkaIO与PubSubIO进行比较(或者更准确地说，将PubsubCheckpoint与KafkaCheckpointMark进行比较)，我可以看到检查点持久化并未在KafkaIO中实现(KafkaCheckpointMark.finalizeCheckpoint方法为空)，而它是在Pub
浏览 61提问于2018-01-24得票数 5
回答已采纳
使用正常的k流，kafka将每个应用程序的偏移量存储在其内部偏移主题上。在重新启动应用程序时，应用程序将根据auto.offset.reset策略重新处理主题。这确实是的解释。我正在使用kafka流的GlobalKTable在应用程序上复制数据。但是，我对应用程序的重新启动有点困惑，因为在重新启动后(由于部署或崩溃) id (**StreamsConfig.APPLICATION_ID_CONFIG**)不更改的应用程序中没有填充。此外，对于支持Glo
浏览 0提问于2018-03-24得票数 1
回答已采纳
我使用Python和Hadoop流，但我无法找到任何方法来配置Hadoop流以获得每一行的偏移量，正如所说，使用TextInputFormat。还有一个在StackOverflow上询问同样的事情，但是对于如何做到这一点没有任何反应。说要抛弃keys of LongWritable class，但我不确定这是否是偏移量。如何配置作业以获得偏移量？是否需要生成从TextInputFormat派生的新类而不</e
浏览 1提问于2016-02-03得票数 1
回答已采纳
我的火花流应用程序从Kafka获取数据并对其进行处理。我找到了"auto.reset.offest"选项，它在星火中的行为并没有什么不同。如果配置好了，它将删除存储在动物园管理员中的偏移量。然而，尽管它的
浏览 1提问于2014-08-26得票数 7
1回答
我有一个postgresql数据库，用于存储包含时区(utc偏移量)的日期时间。我使用了timestampz类型，因为它似乎是我所需要的，但是当我插入日期时，UTC偏移量被转换为UTC+00：
例如，如果我插入2022-10-20 00:00:00+01，实际存储的数据将变为2022是否有一种方法可以保留偏移量而不添加列来存储该信息(或时区)？
浏览 5提问于2022-11-03得票数 2
回答已采纳
在“带有Apache的星火流”一书中，作者提到在使用Kafka作为源进行测试时，应该将"failOnDataLoss (默认: true)“字段设置为false。他们说，
我想了解为什么在停止&
浏览 0提问于2019-06-16得票数 3
Requirement I希望并行处理来自多个使用者的事件数据(相同数据)我从文档中了解到如何处理相同的数据表单事件中心，我们需要创建多个使用者组。使用者组允许多个消费应用程序对每个事件流有一个单独的视图，并以自己的速度和自己的偏移量独立地读取该流。消息保留1，分区计数3问题:我不是从log1消费者组获得数据，而是在尝试$Default和log2消费者组时并
浏览 0提问于2018-04-30得票数 1
回答已采纳
我使用AudioFileStream在网上播放AAC音乐。当用户尝试查找时，我需要调用AudioFileStreamSeek来知道我需要查找的流文件的偏移量，因为AAC是VBR音乐格式，没有AudioFileStreamSeek我无法知道偏移量。如果我发现自己没有数据，但在调用AudioFileStreamSeek后仍需要查找，该怎么办？我唯一能做的就是继续解析字节而不进行查找。有时音乐会停止发声，我想这是因为我违背了在文件的正确偏移量中给目标
浏览 2提问于2013-05-10得票数 2
回答已采纳
我们使用的服务器数量与狮身人面像作为一个分离的层，提供搜索。索引器由cron独立运行，以减少数据流和搜索索引耦合。根据需要，我们应该提供实时搜索(没有任何延迟)，因此我们决定保留狮身人面像的结果，每40分钟重新索引一次，并与基于ActiveRecord updated_at列的新的实际数据进行混合，实际上不超过40这意味着我们有实际的数据而不使用增量功能(这增加了某种耦合)。
问题在于我们如何将Th
浏览 1提问于2012-02-28得票数 0
回答已采纳
我使用boost::iostreams::mapped_file_source将文本文件从特定位置读取到特定位置，并操作每一行(使用g++ -Wall -O3 -lboost_iostreams -o测试if (std_str.eof()) break; } 这段代码也运行良好，但我不知道如何像第一段代码那样设置起始点你能告诉我如何在第二个代码中设置这两个值吗？
浏览 4提问于2015-04-11得票数 4
从文档中可以清楚地看到，当以TRIM_HORIZON作为迭代器类型启动KCL应用程序时，记录将从流的开头读取。文档还提到，应用程序的状态是通过使用检查点在DynamoDB表中维护的。但是，我没有找到任何参考，KCL应用程序是如何使用这些DynamoDB表信息的。
具体而言，我的问题如下-我有流与168小时的保留期，这是大量的数据。假设我的KCL(从TRIM_HORIZON上的迭代器开始)与传入的数据同步，并且在流的末尾处理记录，并且定
浏览 5提问于2017-04-01得票数 3
回答已采纳
我向队列发送了一条消息，所有的消费者都使用了该消息，但是消息仍然保留在队列中(我等待了1分钟以上)，作为“就绪”。我担心的是在rabbitmq实例的HD中信息的积累。如果是的话，我如何才能清除(在这种情况下，最大年龄不起作用)这些消息从HD的狂犬病实例？
浏览 9提问于2022-03-17得票数 0
正如标题所述，对于流插入而言，BigQuery的一致性保证是什么？在使用流插入插入行之后，我能保证读取一致吗？
浏览 6提问于2022-03-25得票数 0
回答已采纳
3回答
但是现在我想把这些数据的定期转储保存在HDFS中，以便在以后进行分析。
因为这基本上只是我需要的一个垃圾场，我不确定我是否需要火花流。然而，所有的文档和示例都使用星火流来实现这一点。有没有一种方法可以在不运行流作业的情况下从Kafka主题填充DF/RDD？或者，在处理了最小到最大偏移量的设置窗口后，这里的范例是否会终止“流”作业？因此，将流作业视为批处理作业。
浏览 1提问于2018-04-11得票数 0
回答已采纳
我正试图使用Apache NiFi将Apache NiFi与卡夫卡消费数据集成在一起。下面是几个问题，在我的脑海中，然后继续进行这项工作。我知道Apache是进行这种处理的合适人选，但是如果我们正在处理的NiFi是一个复杂的工作流，那么构建工作流有多容易呢？我们最初考虑使用Java做同样的事情，但后来意识到这可以在NiFi中用最少的努力来完成。请注意，我们从Kafka处理的数据中有80%是简单的JSON，而20%是复杂的数据<
浏览 0提问于2019-04-18得票数 3
回答已采纳
1回答
我必须在Spark中实现流作业的偏移管理，这是从Kafka流中读取的。然而，尽管这个过程已经在正式文档中描述过，但它实际上并没有给出一个代码示例，说明如何从检查点实际存储和检索偏移量。相反，它含糊不清地说
这是否意味着，如果我只向Spark上下文提供检查点目录，它将自动存储偏移量？当应用程序重新启动时，如何检索上一次偏移读呢？和给出了一些关于如何使用</
浏览 0提问于2019-07-05得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
云直播
对象存储
腾讯会议活动推荐
腾讯云开发者
