《流处理》专题

我试图启动一个流数据流作业，其中包含n个管道。 基于配置的主题和每个主题对应的BQ表，我想在一个流作业内启动一个管道。 这里的运行时参数为bucket_name和config_json_path，用于所有与配置相关的数据集、BQ表、topics/Subscription和所有工作流选项。 这到底有没有可能？因为谷歌也提供了一对一的模板。不是很多对很多模板（例如三个主题-三个BQ表（三个数据流水线）

问题内容： 从我的理解在这里，“V8拥有世代垃圾收集器。物件的移动风靡随机，节点不能得到一个指向原始字符串数据写入到插座。” 因此，我不应该将来自TCP流的数据存储在字符串中，特别是当该字符串变得大于字节时。（希望我到目前为止是..） 那么，处理来自TCP套接字的所有数据的最佳方法是什么？到目前为止，我一直在尝试用作定界符，因为我认为它在某种程度上是唯一的，不会缠结其他事物。 将要收集的数据样本是

我想知道Kafka流是如何分配到主题的分区进行阅读的。据我所知，每个Kafka流线程都是一个消费者（该流有一个消费者组）。所以我猜消费者是随机分配到分区的。 话题P包含人称。它有两个分区。消息的关键是person ID，因此每个属于person的消息最终都位于同一个分区中。 主题O包含订单。它有两个分区。假设密钥也是（订购某样东西的人的）person-id。因此，在这里，属于一个人的每个订单消息总

让我们假设这样一个简单的例子： KStream使用带有转换器的转换操作来消除ORDER_主题中的重复消息，该转换器通过密钥/id将消息存储在持久本地状态存储中。这样，如果相同的顺序到达两次，它将被忽略。 现在一个新订单到达，它不是重复的，所以它存储在本地存储中，但在将其发送到VALIDATED_ORDER_TOPIC应用程序崩溃之前。 我想知道KStream中的事务保证是什么：记录是否已存储并提交

我正在学习akka流，但显然它与任何流媒体框架都相关：） 引用akka文件： Reactive Streams只是定义了一种通用机制，说明如何在不丢失、缓冲或资源耗尽的情况下跨异步边界移动数据 现在，据我所知，如果在streams之前，让我们以http服务器为例，当接收方没有完成请求时，请求就会出现，因此即将出现的新请求将收集在一个缓冲区中，该缓冲区将保存等待的请求，还有一个问题是，这个缓冲区的大

我有一个Pub/Sub主题，它会定期（通常每隔几天或几周一次，但有时更频繁）接收批量消息。我想启动一个批处理数据流作业来读取这些消息，执行一些转换，将结果写入Datastore，然后停止运行。当新一批消息发出时，我想启动一项新工作。我已经阅读了Apache Beam PythonSDK文档和许多SO问题，但仍不确定一些事情。 Pub/Sub IO可以作为非流作业的一部分读取吗？然后同一作业可以使用

我实际上正在部署一个Spark/Kafka/Cassandra应用程序，而我正面临一个不同解决方案的问题，所以我在这里听取您的建议。 > 我在Spark streaming中运行了一个很长时间的应用程序，用于处理Kafka中的Avro消息。根据消息的性质，我可以做一些不同的案例，最后在Cassandra中保存一个记录，所以只是这些技术的一个基本用例。 所以我正在寻找执行批处理作业的最佳实践。由于s

问题内容： 我有一个dao，它基本上使用hibernate将记录插入到一​​个表中，该dao用标记为注释，并且我有一个服务，该服务会生成其他一些东西，然后调用我的dao。我的服务也标注了使用。 我叫服务循环。我在dao上的插入内容是否可以批量或一个接一个地工作？我如何确定它们可以批量工作？hibernateTransaction Manager是否管理批处理插入？ 我正在使用Oracle DB。

需要读取spring批处理中的文件，对其进行处理并将其作为一个提要保存。一个提要包含50%的信息。当我必须持久化提要的最终结果时，我需要使用公共字段将它们组合起来，并像一个项目一样持久化。请参见下面的示例。 我需要保留的最终信息如下： 请建议我如何在我的Spring批工作中实现这一点。 谢谢

我使用的是spring批处理，和通常使用的一样，我有读取器、处理器和写入器。 我有两个问题 1>Reader查询所有200条记录（表中记录总大小为200，我给出了pageSize=200)，因此它得到所有200条记录，在处理器中，我们需要所有这些记录的列表，因为我们必须将每个记录与其他199条记录进行比较，以便将它们分组在不同的层中。因此我在想，如果我们能在处理步骤中得到那个列表，我就可以操纵它们

在happy path场景中，我有一个spring批处理工作，但现在我将重点放在错误处理上。 但是，在另一个测试中，我想证明一个不可预见的数据库错误会导致作业失败。为此，我创建了一个触发器，该触发器会导致对要插入的表的插入失败。 这似乎起作用了，在writer执行之后，在事务提交期间抛出异常，并且我得到以下日志消息： 这似乎也是预期的行为。问题是，这并不能阻止工作。该步骤退出到SimplyRetr

我有一个图像路径列表，我想在进程或线程之间划分，以便每个进程处理列表的某些部分。处理包括从磁盘加载图像，进行一些计算并返回结果。我正在使用Python 2.7 下面是我如何创建辅助进程 我所面临的问题是，当我在initializer函数中记录初始化时间时，我知道worker不是并行初始化的，而是每个worker都以5秒的间隔初始化，下面是供参考的日志 我尝试过使用将同时启动辅助线程 我知道Wind

这种情况应该经常发生：

我阅读了tornado和相关模块，以了解当接受套接字并解析和处理请求时，tornado会做什么。 然后我向tornado发送一些请求，所有请求都被设置为读取超时3秒，我在tornado上发现了一些连接，几分钟后，这些连接正常关闭。 我猜5秒处理程序是减缓tornado的事件循环，当tornado处理第二个处理程序时，它的连接已经关闭，如果tornado知道了，它就不需要再做下一步了。

但是我的接收器只在2个分区中获取数据，但是我配置了20个流线程，并且我验证了我的生产者正在写入所有20个分区，如何知道我的转换节点转发到我的FINAL_TOPIC的所有20个分区