问题标签 [stream-processing]

我一直在ETL使用 Apache Kafka 实现数据管道。我使用 Kafka Connect 进行提取和加载。

Connect 将读取源数据并以 JSON 的形式写入可用的 Kafka 主题实际数据。

在转换阶段，我想从 Kafka 主题读取 JSON 数据，然后需要根据一些自定义业务逻辑转换为 SQL 查询，然后需要写入输出 Kafka 主题。

到目前为止，我已经编写了一个生产者-消费者应用程序，它从主题中读取并进行转换，然后写入输出主题。

是否可以使用 Kafka 流 API 实现相同的目标？如果是，请提供一些样品。

我想在 Flink 中创建一个 BucketingSink ，它将所有文件写入同一个文件夹中，但将文件名设置为当前时间戳而不是递增计数器。例如：

部分 0-1514107452000.avro

部分 0-1514021052000.avro

...

这是我的代码：

有人知道吗？谢谢

I have two Input Stream. I would like to merge two stream element based on same ID. Here is the code details

What I would like to get the result is MergeResult, which is based on the same id in Foo and Bar.

Also, for some Foo and Bar which has mismatched id, I would like to keep in the memory, I wonder if there is a clean way to do it because it is stateful.

More importantly, the source elements are in order. If there are ID duplicates found, the strategy should be first matched first served. That means if Foo(1, "foo-1"), Foo(1, "foo-2") and Bar(1, "Bar-1"), the match should be MergeResult(1, "foo-1", "Bar-1") .

I am looking at some solutions from akka stream at the moment. If there are some good solution like Spark, Flink and so on, that would be helpful as well.

Thanks in advance.

我尝试使用 kafka 版本 0.11.0.2 运行以下程序。我编译它并创建了一个可运行的 jar，它从设置 kafka 的服务器运行。通过生产者/输入主题发布到kafka的输入是在消费者/输出主题中获得的，没有应用处理逻辑，我没有得到字数。（Kafka 设置为单个分区，复制因子为 1）。

我如何让它工作？我在这里做错了什么？

运行流束管道，我使用 avroIO 从 gcs 流式传输文件/记录，然后创建每分钟/每小时桶以聚合事件并将其添加到 BQ。如果管道失败，我如何才能正确恢复并仅处理未处理的事件？我不想重复计算事件。我想的一种方法是写入扳手或大表，但可能是写入 BQ 成功但数据库失败，反之亦然？如何在流式管道中以可靠一致的方式保持状态以仅处理未处理的事件？我想确保 BQ 中的最终汇总数据是不同事件的准确计数，而不是计数不足或过度计数？火花流管道如何解决这个问题（我知道他们有一些检查点目录来管理查询和数据帧的状态）？是否有任何推荐的技术可以准确地解决流管道中的此类问题？

我们使用 Kafka Streams 库进行流处理。但是我们有太多用于连接 kafka 代理的 tcpv6/pipe 开放描述符的问题。

我们使用的 Kafka Stream 代码与上面的代码类似，不同之处在于我们有 10 多个用于 StreamsBuilder 的 KStream。

如果有人能就这个问题提供任何帮助，我们将不胜感激。谢谢！

我正在使用 0.11.0.2 版的 Kafka Streams。

为了利用API，我使用builder 方法transform创建了自己的 API 。问题是我对某些字段/方法的 javadoc 不够清楚。StateStoreSupplierStores.create

提到的变更日志将如何表示？

这 4 个值如何相互影响？每个窗口都有最大数量的元素 - windowSize？一旦到达新窗口就开始了？每个窗口都可以划分numSegments为 RocksDB 磁盘上的文件吗？重复意味着键和值都相同，并且仅在同一窗口中检测到？

这里暗示了什么样的缓存？

I have the following system in mind: A master program that polls a list of tasks to see if they should be launched (based on some trigger information). The tasks themselves are container images in some repository. Tasks are executed as jobs on a Kubernetes cluster to ensure that they are run to completion. The master program is a container executing in a pod that is kept running indefinitely by a replication controller.

However, I have not stumbled upon this pattern of launching jobs from a pod. Every tutorial seems to be assuming that I just call kubectl from outside the cluster. Of course I could do this but then I would have to ensure the master program's availability and reliability through some other system. So am I missing something? Launching one-off jobs from inside an indefinitely running pod seems to me as a perfectly valid use case for Kubernetes.

我有两个独立的事件流，一个流我有小时桶，另一个流我有 4 小时桶，是否可以加入这两个流。我如何确定要加入两个流上的哪些窗口？我可以在一个流上有一个滑动窗口并将其与另一个流上的固定窗口连接吗？示例用例是我将一个流划分为固定的每分钟/每小时桶，但希望它们加入窗口必须的 24 小时滚动/滑动桶与相同的开始时间对齐。有可能在 spark 中这样做吗？

Apache Beam 或 Spark Streaming 中的流处理中的水印估计有多准确。我的数据源是来自 gcs/s3 的文件，但我使用与每个事件关联的事件时间作为窗口函数的时间戳。关于这些流处理引擎如何计算这种启发式或估计的任何想法，以及是否有办法衡量这个估计有多糟糕。

我的用例我有几个服务器在 gcs/S3 上生成事件日志，然后我从我的流处理引擎以流的方式读取这些文件。因此，由于文件系统中断和故障或服务器在几个小时内无法刷新日志事件，可能会出现延迟。因此，在我的流处理管道中，正确性是聚合某些事件时的重要方面之一。所以我很好奇这个水印估计是如何计算的