Kafka Streams 在监控场景的应用与实践

news2024/12/24 10:09:34

作者：来自 vivo 互联网服务器团队- Pang Haiyun

介绍 Kafka Streams 的原理架构，常见配置以及在监控场景的应用。

一、背景

在当今大数据时代，实时数据处理变得越来越重要，而监控数据的实时性和可靠性是监控能力建设最重要的一环。随着监控业务需求的变化和技术的发展，需要能够实时处理和分析庞大的数据流。作为一种流式处理平台，Kafka Streams 为处理实时数据提供了强大的支持。本文将重点介绍如何利用 Kafka Streams 进行实时数据处理，包括其基本原理、功能和实际应用。通过本文的学习，读者将能够深入了解 Kafka Streams 的优势、在监控场景的应用及实践。

二、Kafka Streams 的基本概念

Kafka Streams 是一个开源的流式处理框架，基于 Kafka 消息队列构建，能够处理无限量的数据流。与传统的批处理不同，Kafka Streams 允许用户以流式处理的方式实时处理数据，而且处理延迟仅为毫秒级。

通过 Kafka Streams ，用户可以进行数据的实时转换、聚合、过滤等操作，同时能够与 Kafka Connect 和 Kafka Producer/Consumer 无缝集成。Kafka Streams 也是一个客户端程序库，用于处理和分析存储在 Kafka 中的数据，并将得到的数据写回 Kafka 或发送到外部系统。

Kafka、Storm、Flink 和 Spark 是大数据领域常用的工具和框架。

1、区别：

  • Kafka 是一个分布式消息系统，主要用于构建实时数据管道和事件驱动的应用程序。它提供了高吞吐量、持久性、可伸缩性和容错性，主要用于数据的发布和订阅。

  • Storm 是一个分布式实时计算系统，用于处理实时数据流。它提供了低延迟、高吞吐量的实时计算能力，适用于实时数据处理和流式计算。

  • Flink 是一个流处理引擎，提供了精确一次的状态处理和事件时间处理等特性。它支持流处理和批处理，并提供了统一的 API 和运行时环境。

  • Spark 是一个通用的大数据处理框架，提供了批处理和流处理的功能。Spark 提供了丰富的数据处理和计算功能，包括 SQL 查询、机器学习、图处理等。

2、Kafka 的优势：

  • 持久性和可靠性：Kafka 提供了数据持久化的功能，能够确保数据不丢失，并且支持数据的持久存储和重放。

  • 可伸缩性：Kafka 集群可以很容易地进行水平扩展，支持大规模数据处理和高并发访问。

  • 灵活性：Kafka 可以与各种不同的数据处理框架集成，作为数据源或数据目的地，使其在实时数据处理的场景中具有广泛的适用性。

总的来说，Kafka 的优势在于其高吞吐量、持久性和可靠性，以及灵活的集成能力，使其成为构建实时数据管道和事件驱动应用程序的理想选择。

2.1 Stream 处理拓扑

2.1.1 流

流是 Kafka Streams 提出的最重要的抽象概念：它表示一个无限的，不断更新的数据集。流是一个有序的，可重放（反复的使用），不可变的容错序列，数据记录的格式是键值对（key-value）。这里的 key 主要记录的是 value 的索引，决定了 Kafka 和 Kafka Streams 中数据的分区，即数据如何路由到 Topic 的特定分区。value 是主要后续处理器要处理的数据。

图片

2.1.2 处理器拓扑

处理器拓扑是一个由流（边缘）连接的流处理（节点）的图。通过 Kafka Streams ，我们可以编写一个或多个的计算逻辑的处理器拓扑，用于对数据进行多步骤的处理。

2.1.3 流处理器

流处理器是处理器拓扑中的一个节点；它表示一个处理的步骤，用来转换流中的数据（从拓扑中的上游处理器一次接受一个输入消息，并且随后产生一个或多个输出消息到其下游处理器中）。

在拓扑中有两个特别的处理器：

  • 源处理器（Source Processor）：源处理器是一个没有任何上游处理器的特殊类型的流处理器。它从一个或多个 Kafka 主题生成输入流。通过消费这些主题的消息并将它们转发到下游处理器。

  • sink 处理器（Sink Processor）：sink 处理器是一个没有下游流处理器的特殊类型的流处理器。它接收上游流处理器的消息发送到一个指定的 Kafka 主题。

图片

(图片来源: Kafka 官网)

Kafka Streams 提供2种方式来定义流处理器拓扑：Kafka  Streams DSL 提供了更常用的数据转换操作，如 map 和 filter；低级别  Processor API 允许开发者定义和连接自定义的处理器，以及和状态仓库交互。处理器拓扑仅仅是流处理代码的逻辑抽象。

2.2 时间

在流处理方面有一些重要的时间概念，它们是建模和集成一些操作的重要元素，例如定义窗口的时间界限。

时间在流中的常见概念如下：

  • 事件时间 - 当一个事件或数据记录发生的时间点，就是最初创建的“源头”。

  • 处理时间 - 事件或数据消息发生在流处理应用程序处理的时间点。即，记录已被消费。处理时间可能是毫秒，小时，或天等。比原始事件时间要晚。

  • 摄取时间 - 事件或数据记录是 Kafka broker 存储在 topic 分区的时间点。与事件时间的差异是，当记录由 Kafka broker 追加到目标 topic 时，生成的摄取时间戳，而不是消息创建时间（“源头”）。与处理时间的差异是处理时间是流处理应用处理记录时的时间。比如，如果一个记录从未被处理，那么就没有处理时间，但仍然有摄取时间。

Kafka Streams 通过 TimestampExtractor 接口为每个数据记录分配一个时间戳。该接口的具体实现了基于数据记录的实际内容检索或计算获得时间戳，例如嵌入时间戳字段提供的事件时间语义，或使用其他的方法，比如在处理时返回当前的 wall-clock（墙钟）时间，从而产生了流应用程序的处理时间语义。因此开发者可以根据自己的业务需要选择执行不同的时间。例如，每条记录时间戳描述了流的时间增长（尽管记录在 stream 中是无序的）并利用时间依赖性来操作，如 join。

最后，当一个 Kafka Streams 应用程序写入记录到 Kafka 时，它将分配时间戳到新的消息。时间戳分配的方式取决于上下文：

  • 当通过处理一些输入记录（例如，在 process()函数调用中触发的 context.forward()）生成新的输出记录时，输出记录时间戳直接从输入记录时间戳继承。

  • 当通过周期性函数（如 punctuate()）生成新的输出记录时。输出记录时间戳被定义为流任务的当前内部时间（通过 context.timestamp() 获取）。

  • 对于聚合，生成的聚合更新的记录时间戳将被最新到达的输入记录触发更新。

本部分简要介绍了 Kafka Streams 的基本概念，下一部分将介绍 Kafka Streams 的在监控场景的应用实践。

三、Kafka Streams 在监控场景的应用

3.1 链路分布示意图

图片

3.2 示例：使用 Kafka Streams 来处理实时数据

流式处理引擎（如 Kafka Streams）与监控数据 ETL 可以为业务运维带来诸多好处，例如实时数据分析、实时监控、事件驱动的架构等。在本部分，我们将重点介绍  Kafka Streams 与监控数据 ETL 的集成，以及如何在监控数据 ETL 中利用 Kafka Streams 进行实时数据处理。

在监控数据ETL架构中，Kafka Streams 扮演着举足轻重的角色。它可以作为一个独立的数据处理服务来处理实时的数据流，并将处理结果输出到其他存储组件（例如，ES、VM等）中。同时，它也可以作为多个数据源之间的数据交换和通信的桥梁，扮演着数据总线的角色。Kafka Streams 的高可用性、高吞吐量和流式处理能力使得它成为监控数据ETL架构中的重要组件之一。

下面给出一个示例，演示了如何将 Kafka Streams 作为监控数据 ETL 来处理实时的数据。假设我们有一个监控数据流 TopicA，我们希望对这些数据进行实时的分析，并将分析结果输出到另一个 TopicB。我们可以创建一个 Kafka Streams 来处理这个需求：

//创建配置类
Properties props = new Properties();
//设置订阅者
props.put(StreamsConfig.APPLICATION_ID_CONFIG, "stream-processing-service");
//设置servers地址
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092");
 
StreamsBuilder builder = new StreamsBuilder();
//构建流
KStream<String, String> userActions = builder.stream("TopicA");
//对流进行处理
KTable<String, Long> userClickCounts = userActions
 
    .filter((key, value) -> value.contains("click"))
 
    .groupBy((key, value) -> value.split(":")[0])
 
    .count();
//流写回Kafka
userClickCounts.toStream().to("TopicB", Produced.with(Serdes.String(), Serdes.Long()));
 
KafkaStreams streams = new KafkaStreams(builder.build(), props);
 
streams.start();

在这个示例中，我们创建了一个 Kafka Streams 监控数据 ETL，用于处理实时的监控数据流。它对数据进行了过滤、分组和统计分析，并将结果输出到 TopicB。通过这个 ETL，我们可以很容易地实现实时的数据处理功能，并且能够与其他数据源和数据存储组件进行无缝的集成。

3.3 监控 ETL 的流处理示意图

图片

本部分介绍了 Kafka Streams 的在监控场景的应用实践，下一部分将深入探讨 Kafka Streams 的运作原理及实时数据处理的常见操作，并阐述 Kafka Streams 如何实现这些操作。

四、监控数据 ETL 中 Kafka Streams 的运作原理

4.1 架构

Kafka Streams 通过生产者和消费者，并利用 Kafka 自有的能力来提供数据平行性，分布式协调性，故障容错和操作简单性，从而简化了应用程序的开发，在本节中，我们将描述 Kafka Streams 是如何工作的。

下图展示了 Kafka Streams 应用程序的解剖图，让我们来看一下。

图片

(图片来源: Kafka 官网)

Kafka 消费者通过消费1个或多个 Topic 拿到数据，形成输入 Kafka 流，经过处理器拓扑对数据进行统一处理形成输出 Kafka 流，将数据写入1个或多个出流 Topic，这是 kafka 流整体的运行流程。

4.1.1 Stream 分区和任务

Kafka 分区数据的消息层用于存储和传输，Kafka Streams  分区数据用于处理， 在这两种情况下，这种分区规划和设计使数据具有弹性，可扩展，高性能和高容错的能力。Kafka Streams 使用了分区和任务的概念，基于 Kafka 主题分区的并行性模型。在并发环境里，Kafka  Streams 和 Kafka 之间有着紧密的联系：

  • 每个流分区是完全有序的数据记录队列，并映射到 Kafka 主题的分区。

  • 流的数据消息与主题的消息映射。

  • 数据记录中的 keys 决定了 Kafka 和 Kafka Streams  中数据的分区，即，如何将数据路由到指定的分区。

应用程序的处理器拓扑通过将其分成多个任务来进行扩展，更具体点说，Kafka Streams 根据输入流分区创建固定数量的任务，其中每个任务分配一个输入流的分区列表（即，Kafka 主题）。分区对任务的分配不会改变，因此每个任务是应用程序并行性的固定单位。然后，任务可以基于分配的分区实现自己的处理器拓扑；他们还可以为每个分配的分区维护一个缓冲，并从这些记录缓冲一次一个地处理消息。作为结果，流任务可以独立和并行的处理而无需手动干预。

重要的是要理解 Kafka Streams 不是资源管理器，而是可在任何地方都能“运行”的流处理应用程序库。多个实例的应用程序在同一台机器上执行，或分布多个机器上，并且任务可以通过该库自动的分发到这些运行的实例上。分区对任务的分配永远不会改变；如果一个应用程式实例失败，则这些被分配的任务将自动地在其他的实例重新创建，并从相同的流分区继续消费。

下面展示了2个分区，每个任务分配了输出流的1个分区。

图片

(图片来源: Kafka 官网)

4.1.2 线程模型

Kafka Streams 允许用户配置线程数，可用于平衡处理应用程序的实例。每个线程的处理器拓扑独立的执行一个或多个任务。例如，下面展示了一个流线程运行2个流任务。

图片

(图片来源: Kafka 官网)

启动更多的流线程或更多应用程序实例，只需复制拓扑逻辑（即复制代码到不同的机器上运行），达到并行处理处理不同的 Kafka 分区子集的目的。要注意的是，这些线程之间不共享状态。因此无需协调内部的线程。这使它非常简单在应用实例和线程之间并行拓扑。Kafka 主题分区的分配是通过 Kafka Streams 利用 Kafka 的协调功能在多个流线程之间透明处理。

如上所述，Kafka Streams 扩展流处理应用程序是很容易的：你只需要运行你的应用程序实例，Kafka Streams 负责在实例中运行的任务之间分配分区。你可以启动多个应用程序线程处理多个输入的 Kafka 主题分区。这样，所有运行中的应用实例，每个线程（即运行的任务）至少有一个输入分区可以处理。

4.1.3 故障容错

Kafka Streams 基于 Kafka 分区的高可用和副本故障容错能力。因此，当流数据持久到 Kafka，即使应用程序故障，如果需要重新处理它，它也是可用的。Kafka  Streams 中的任务利用 Kafka 消费者客户端提供的故障容错的能力来处理故障。如果任务故障，Kafka Streams 将自动的在剩余运行中的应用实例重新启动该任务。

此外，Kafka Streams 还确保了本地状态仓库对故障的稳定性。对于每个状态仓库都维持一个追踪所有的状态更新的变更日志主题。这些变更日志主题也分区，因此，每个本地状态存储实例，在任务访问仓里，都有自己的专用的变更日志分区。变更主题日志也启用了日志压缩，以便可以安全的清除旧数据，以防止主题无限制的增长。如果任务失败并在其他的机器上重新运行，则  Kafka Streams 在恢复新启动的任务进行处理之前，重放相应的变更日志主题，保障在故障之前将其关联的状态存储恢复。故障处理对于终端用户是完全透明的。

请注意，任务（重新）初始化的成本通常主要取决于通过重放状态仓库变更日志主题来恢复状态的时间。为了减少恢复时间，用户可以配置他们的应用程序增加本地状态的备用副本（即完全的复制状态）。当一个任务迁移发生时，Kafka Streams 尝试去分配任务给应用实例，提前配置了备用副本的应用实例就可以减少任务（重新）初始化的成本。

4.2 创建流

记录流（KStreams）或变更日志流（KTable或GlobalkTable）可以从一个或多个 Kafka 主题创建源流，（而 KTable 和 GlobalKTable，只能从单个主题创建源流）。

KStreamBuilder builder = new KStreamBuilder();
 
KStream<String, GenericRecord> source1 = builder.stream("topic1", "topic2");
KTable<String, GenericRecord> source2 = builder.table("topic3", "stateStoreName");
GlobalKTable<String, GenericRecord> source2 = builder.globalTable("topic4", "globalStoreName");

4.3 流回写 Kafka

在处理结束后，开发者可以通过 KStream.to 和 KTable.to 将最终的结果流（连续不断的）写回 Kafka 主题。

joined.to("topic4");

如果已经通过上面的to方法写入到一个主题中，但是如果你还需要继续读取和处理这些消息，可以从输出主题构建一个新流，Kafka Streams 提供了便利的方法，through:

// equivalent to
//
// joined.to("topic4");
// materialized = builder.stream("topic4");
KStream materialized = joined.through("topic4");

4.4 流程序的配置与启执行

除了定义的 topology，开发者还需要在运行它之前在 StreamsConfig 配置他们的应用程序，Kafka Streams 配置的完整列表可以在这里找到。

Kafka Streams 中指定配置和生产者、消费者客户端类似，通常，你创建一个 java.util.Properties，设置必要的参数，并通过 Properties 实例构建一个 StreamsConfig 实例。

import java.util.Properties;
import org.apache.kafka.streams.StreamsConfig;
Properties settings = new Properties();
// Set a few key parameters
settings.put(StreamsConfig.APPLICATION_ID_CONFIG, "my-first-streams-application");
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092");
settings.put(StreamsConfig.ZOOKEEPER_CONNECT_CONFIG, "zookeeper1:2181");
 
// Any further settings
settings.put(... , ...);
 
// Create an instance of StreamsConfig from the Properties instance
StreamsConfig config = new StreamsConfig(settings);

除了 Kafka Streams 自己配置参数，你也可以为 Kafka 内部的消费者和生产者指定参数。根据你应用的需要。类似于 Streams 设置，你可以通过 StreamsConfig 设置任何消费者和/或生产者配置。请注意，一些消费者和生产者配置参数使用相同的参数名。例如，用于配置 TCP 缓冲的 send.buffer.bytes 或 receive.buffer.bytes。用于控制客户端请求重试的 request.timeout.ms 和 retry.backoff.ms。如果需要为消费者和生产者设置不同的值，可以使用 consumer. 或 producer. 作为参数名称的前缀。

Properties settings = new Properties();
 
// Example of a "normal" setting for Kafka Streams
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker-01:9092");
// Customize the Kafka consumer settings
streamsSettings.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 60000);
// Customize a common client setting for both consumer and producer
settings.put(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG, 100L);
// Customize different values for consumer and producer
settings.put("consumer." + ConsumerConfig.RECEIVE_BUFFER_CONFIG, 1024 * 1024);
settings.put("producer." + ProducerConfig.RECEIVE_BUFFER_CONFIG, 64 * 1024);
 
// Alternatively, you can use
settings.put(StreamsConfig.consumerPrefix(ConsumerConfig.RECEIVE_BUFFER_CONFIG), 1024 * 1024);
settings.put(StremasConfig.producerConfig(ProducerConfig.RECEIVE_BUFFER_CONFIG), 64 * 1024);

你可以在应用程序代码中的任何地方使用 Kafka Streams ，常见的是在应用程序的 main() 方法中使用。

首先，先创建一个 KafkaStreams 实例，其中构造函数的第一个参数用于定义一个 topology builder（Streams DSL的KStreamBuilder，或 Processor API 的 TopologyBuilder）。

第二个参数是上面提到的 StreamsConfig 的实例。

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.processor.TopologyBuilder;
// Use the builders to define the actual processing topology, e.g. to specify
// from which input topics to read, which stream operations (filter, map, etc.)
// should be called, and so on.
KStreamBuilder builder = ...;  // when using the Kafka Streams DSL
//
// OR
//
TopologyBuilder builder = ...; // when using the Processor API
// Use the configuration to tell your application where the Kafka cluster is,
// which serializers/deserializers to use by default, to specify security settings,
// and so on.
StreamsConfig config = ...;
KafkaStreams streams = new KafkaStreams(builder, config);

在这点上，内部结果已经初始化，但是处理还没有开始。你必须通过调用 start() 方法启动 Kafka Streams 线程：

// Start the Kafka Streams instance
streams.start();

捕获任何意外的异常，设置 java.lang.Thread.UncaughtExceptionHandler。每当流线程由于意外终止时，将调用此处理程序。

streams.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
    public uncaughtException(Thread t, throwable e) {
        // here you should examine the exception and perform an appropriate action!
    }
);

close() 方法结束程序。

// Stop the Kafka Streams instance
streams.close();

现在，运行你的应用程序，像其他的 Java 应用程序一样（Kafka Sterams 没有任何特殊的要求）。同样，你也可以打包成 jar，通过以下方式运行：

# Start the application in class com.example.MyStreamsApp
# from the fat jar named path-to-app-fatjar.jar.
$ java -cp path-to-app-fatjar.jar com.example.MyStreamsApp

当应用程序实例开始运行时，定义的处理器拓扑将被初始化成1个或多个流任务，可以由实例内的流线程并行的执行。如果处理器拓扑定义了状态仓库，则这些状态仓库在初始化流任务期间（重新）构建。这一点要理解，当如上所诉的启动你的应用程序时，实际上 Kafka Streams 认为你发布了一个实例。现实场景中，更常见的是你的应用程序有多个实例并行运行（如，其他的 JVM 中或别的机器上）。在这种情况下，Kafka Streams 会将任务从现有的实例中分配给刚刚启动的新实例。

五、监控数据 ETL 中 Kafka Streams 参数及其调优

5.1 必配参数：

  1. bootstrap.servers：这是 Kafka 集群的地址列表，Kafka Streams 使用它来初始化与 Kafka 的连接。

  2. key.deserializer 和 value.deserializer：这些配置定义了流中键和值的序列化和反序列化器。

  3. auto.offset.reset：当没有初始偏移量或偏移量无效时，这个配置定义了 Kafka Streams 如何处理。

  4. group.id：这对于使用 Kafka Streams 的消费者组来说很重要，它定义了消费者组的ID。

5.2 基础参数：

  1. num.stream.threads：定义 Kafka Streams 应用程序中的线程数，默认与处理器的逻辑核心数相等。

  2. state.dir：定义 Kafka Streams 存储状态的本地目录。

  3. threading.max.instances：定义每个主题分区的最大线程实例数，默认与分区数相等。

  4. threading.instances：定义每个主题分区的线程实例数，默认与分区数相等。

5.3 消费者参数：

  1. enable.auto.commit：自动提交偏移量，默认值为"true"，建议设置为"false"，以便更好地控制偏移量的提交。

  2. commit.interval.ms：提交偏移量的频率，默认值为5000ms，可以根据需要进行调整。

  3. max.poll.records：一次拉取的消息数量，默认值为1000，可以根据网络带宽和处理能力进行调整。

5.4 生产者参数：

  1. batch.size：批量发送消息的大小，默认值通常是16384（字节），可以根据网络带宽和 Kafka 集群的性能进行调整。

  2. linger.ms：消息在生产者缓冲区中的最小停留时间，默认值为100ms，可以根据需要进行调整。

  3. compression.type：压缩类型，可以提高网络带宽利用率，但会增加 CPU 开销。默认值为"none"，可以根据需要设置为"gzip"、“snappy"或"lz4”。

对于 Kafka 的调优参数，可以根据实际的应用场景和性能需求进行调整，以达到最佳的性能和稳定性。

六、监控数据 ETL 中 Kafka Streams 的分区倾斜问题原因和解决方式

6.1 原因

分区倾斜是监控数据 ETL 的 Kafka Streams 在处理大规模数据流时遇到的常见问题。分区倾斜指的是在一个流处理应用程序中，某个分区的消息消费速度远远慢于其他分区，或某个分区的延迟积压数据远大于其他分区，导致  Kafka Streams 的实时性受到限制。

产生分区倾斜的原因可能包括：

  1. 数据分布不均匀：原始数据在 Kafka 主题的分区中分布不均匀，导致某些分区的消息量远大于其他分区。

  2. 消费者实例数量不足：在 Kafka Streams 应用程序中，消费者的实例数量不足，无法充分处理所有分区的消息。

  3. 消费者负载不均衡：消费者的负载不均衡（包括但不限于某些消费者实例处理的分区数大于其他实例），导致某些消费者实例处理的消息量远大于其他实例。

  4. 消费者实例负载不均衡：消费者实例性能不一致或性能被挤占，导致消费能力不均衡，消费速率异常小于平均消费速率

6.2 解决方案

  1. 数据均衡策略：在设计 Kafka 主题分区分配策略时，可以采用如轮询（Round-robin）或范围（Range）等均衡策略，使得数据在各个分区之间均匀分布。

  2. 增加消费者实例：根据应用程序的实际情况，适当增加消费者的实例数量，以提高整个系统的处理能力，例如扩容。

  3. 负载均衡策略：在消费者组内部实现负载均衡，如使用均匀分配消费者（Uniform Distribution Consumer）等策略，确保消费者实例之间的负载均衡，例如重启或剔除倾斜分区实例使 Kafka Streams 的分区进行重新分配。

  4. 优化消费者处理逻辑：分析消费者处理消息的速度慢的原因，优化处理逻辑，提高消费者的处理能力。

  5. 调整批次大小和窗口函数：通过调整 Kafka Streams 的批次大小和窗口函数等参数，降低消费者的处理压力。

  6. 使用侧输出：对于一些处理速度较慢的分区，可以考虑使用侧输出将部分消息引流至其他系统处理，减轻消费者负载。

七、总结

本文介绍了 Kafka Streams 在监控场景中的应用，阐述了 Kafka Streams 的基本概念，包括流、处理器拓扑、流处理器、时间概念等，举例说明了 Kafka Streams 在监控实时数据ETL中的具体应用，并详细解释了 Kafka Streams 的运作原理，包括其架构、创建流、流回写 Kafka、流程序配置与启执行等内容。文章还介绍了 Kafka Streams 的参数及其调优方法，以及可能出现的分区倾斜问题及其解决方法。

本文意在让读者对于 Kafka 流在监控业务的实际应用有所认识，并且了解 Kafka 流的基本概念和原理，阅读本文后对构建自己 Kafka 流应用程序有所帮助，能够理解在监控数据 ETL 常见分区倾斜的原理和解决方式。

引用：Kafka 官网 https://kafka.apache.org/

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2264661.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

数据分析思维（五）：分析方法——假设检验分析方法

数据分析思维（五）：分析方法——假设检验分析方法

数据分析并非只是简单的数据分析工具三板斧——Excel、SQL、Python&#xff0c;更重要的是数据分析思维。没有数据分析思维和业务知识&#xff0c;就算拿到一堆数据&#xff0c;也不知道如何下手。 推荐书本《数据分析思维——分析方法和业务知识》&#xff0c;本文内容就是提取…
阅读更多...
解读DiffusionNER: Boundary Diffusion for Named Entity Recognition

解读DiffusionNER: Boundary Diffusion for Named Entity Recognition

content 摘要1. 图1图21. 上方&#xff1a;扩散过程与实体边界2. 下方&#xff1a;网络结构&#xff08;Sentence Encoder Entity Decoder&#xff09;3. 关键思想小结 摘要 主要内容分为四个部分&#xff1a; 模型定位与基本原理&#xff1a; 提出了DiffusionNER模型将命名…
阅读更多...
【QSS样式表 - ⑥】：QPushButton控件样式

【QSS样式表 - ⑥】：QPushButton控件样式

文章目录 QPushBUtton控件样式QSS示例 QPushBUtton控件样式 常用子控件 常用伪状态 QSS示例 代码&#xff1a; QPushButton {background-color: #99B5D1;color: white;font-weigth: bold;border-radius: 20px; }QPushButton:hover {background-color: red; }QPushButton:p…
阅读更多...
数字经济下的 AR 眼镜

数字经济下的 AR 眼镜

目录 1. &#x1f4c2; AR 眼镜发展历史 1.1 AR 眼镜相关概念 1.2 市面主流 XR 眼镜 1.3 AR 眼镜大事记 1.4 国内外 XR 眼镜 1.5 国内 AR 眼镜四小龙 2. &#x1f531; 关键技术 2.1 AR 眼镜近眼显示原理 2.2 AR 眼镜关键技术 2.3 AR 眼镜技术难点 3. &#x1f4a…
阅读更多...
smb和nfs双栈协议共享目录

smb和nfs双栈协议共享目录

1 简介 NFS和SAMBA协议都是文件共享&#xff0c;Linux客户端常用于NFS协议访问远程共享目录&#xff0c;Windows客户端常用于SAMBA协议访问远程共享目录。 2 环境 合计使用三台服务器&#xff0c;服务器都位于同一个子网&#xff08;10.0.0.0/19&#xff09;、同一个安全组…
阅读更多...
Day13 用Excel表体验梯度下降法

Day13 用Excel表体验梯度下降法

Day13 用Excel表体验梯度下降法 用所学公式创建Excel表 用Excel表体验梯度下降法 详见本Day文章顶部附带资源里的Excel表《梯度下降法》&#xff0c;可以对照表里的单元格公式进行理解&#xff0c;还可以多尝试几次不同的学习率 η \eta η来感受&#xff0c;只需要更改学习率…
阅读更多...
Python获取系统负载并打印折线图

Python获取系统负载并打印折线图

#! /opt/py36/bin/python import psutil import matplotlib.pyplot as plt import time# 创建一个空列表&#xff0c;用于存储负载数据 load_data []# 循环收集负载数据 while True:# 获取当前系统负载load_avg psutil.getloadavg()# 将平均负载添加到load_data列表中load_da…
阅读更多...
RCE 命令执行漏洞 过滤模式 基本的过滤问题 联合ctf题目进行实践

RCE 命令执行漏洞 过滤模式 基本的过滤问题 联合ctf题目进行实践

前言 知道RCE 命令执行分为 代码执行 和 命令执行 原理 &#xff1a; 就是用户的输入被当做命令或者代码执行了 从而造成了危害 代码执行 除了eval php代码执行漏洞的函数还有 eval()、a ssert()、 preg_replace()、 create_function()、 array_map()、 call_user_func(…
阅读更多...
Leetcode打卡：考场就坐

Leetcode打卡：考场就坐

执行结果&#xff1a;通过 题目&#xff1a; 855 考场就坐 在考场里&#xff0c;有 n 个座位排成一行&#xff0c;编号为 0 到 n - 1。 当学生进入考场后&#xff0c;他必须坐在离最近的人最远的座位上。如果有多个这样的座位&#xff0c;他会坐在编号最小的座位上。(另外&am…
阅读更多...
数据结构（哈希表（上）纯概念版）

数据结构（哈希表（上）纯概念版）

前言 在软件开发和计算机科学中&#xff0c;数据结构的选择直接影响到程序的性能和效率。不同的数据结构适用于不同的场景&#xff0c;合理地选择合适的数据结构是高效编程的关键之一。哈希表&#xff08;哈希表&#xff08;Hash Table&#xff09;作为一种高效的键值对存储结…
阅读更多...
【机器学习与数据挖掘实战】案例06：基于Apriori算法的餐饮企业菜品关联分析

【机器学习与数据挖掘实战】案例06：基于Apriori算法的餐饮企业菜品关联分析

【作者主页】Francek Chen 【专栏介绍】 ⌈ ⌈ ⌈机器学习与数据挖掘实战 ⌋ ⌋ ⌋ 机器学习是人工智能的一个分支，专注于让计算机系统通过数据学习和改进。它利用统计和计算方法，使模型能够从数据中自动提取特征并做出预测或决策。数据挖掘则是从大型数据集中发现模式、关联…
阅读更多...
深入解析 Spring WebFlux：原理与应用

深入解析 Spring WebFlux：原理与应用

优质博文&#xff1a;IT-BLOG-CN WebFlux 是 Spring Framework 5 引入的一种响应式编程框架&#xff0c;和Spring MVC同级&#xff0c;旨在处理高并发和低延迟的非阻塞应用。这是一个支持反应式编程模型的新Web框架体系。 顺便一提&#xff0c;Spring Cloud Gateway在实现上是…
阅读更多...
C语言基础——指针（4）

C语言基础——指针（4）

一&#xff0e; 字符指针变量 字符指针变量的使用和整型指针变量的使用方法相似&#xff0c;以下是其基本使用方法的例子&#xff1a; &#xff08;1&#xff09;字符指针变量还有一种使用方法&#xff1a; const char* p "abcd" 需…
阅读更多...
『 Linux 』高级IO (一)

『 Linux 』高级IO (一)

文章目录 内容回顾及铺垫五种IO模型不同类型IO的区别非阻塞IOfcntl( ) 多路转接 - select( )select( ) 的基本使用 - SelectServer服务器 内容回顾及铺垫 在博客『 Linux 』基础IO/文件IO (万字)中介绍了对IO的认识; IO实际上为Input/Output,输入输出; 以网络协议栈的视角来看,…
阅读更多...
Spark-Streaming集成Kafka

Spark-Streaming集成Kafka

Spark Streaming集成Kafka是生产上最多的方式&#xff0c;其中集成Kafka 0.10是较为简单的&#xff0c;即&#xff1a;Kafka分区和Spark分区之间是1:1的对应关系&#xff0c;以及对偏移量和元数据的访问。与高版本的Kafka Consumer API 集成时做了一些调整&#xff0c;下面我们…
阅读更多...
「下载」智慧城市包括哪些方面：大数据公共服务平台、城市运行指挥中心、城市综合治理平台、城市体检综合运营平台解决方案

「下载」智慧城市包括哪些方面：大数据公共服务平台、城市运行指挥中心、城市综合治理平台、城市体检综合运营平台解决方案

在当今信息化高速发展的时代&#xff0c;智慧城市已成为全球城市发展的新趋势。系列全面而创新的智慧城市解决方案&#xff0c;旨在助力城市实现智慧化转型&#xff0c;提升城市管理效率&#xff0c;增强市民生活质量。 智慧城市最新解决方案&#xff0c;标准规范顶层设计指南、…
阅读更多...
ChatGPT生成接口文档实践案例（二）

ChatGPT生成接口文档实践案例（二）

不难发现&#xff0c;两个方案都出色地完成了接口文档的生成&#xff0c;但笔者更喜欢Response 2的表达&#xff0c;因为其描述更加全面。 还可以让ChatGPT生成符合OpenAPI 3.0规范的接口文档&#xff0c;以便于项目相关成员阅读&#xff0c;如图5-13所示。 为什么要生成OpenAP…
阅读更多...
【解决】Linux更新系统内核后Nvidia-smi has failed...

【解决】Linux更新系统内核后Nvidia-smi has failed...

问题概述 由于服务器(操作系统为 RedHat 9)宕机&#xff0c;重启后&#xff0c;系统内核自动更新了&#xff0c;然后输入 nvidia-smi 发现报了下面的异常&#xff1a; NVIDIA-SMI has failed because it couldnt communicate with the NVIDIA driver. Make sure that the late…
阅读更多...
Docker Compose 安装 Harbor

Docker Compose 安装 Harbor

我使用的系统是rocky Linux 9 1. 准备环境 确保你的系统已经安装了以下工具&#xff1a; DockerDocker ComposeOpenSSL&#xff08;用于生成证书&#xff09;#如果不需要通过https连接的可以不设置 1.1 安装 Docker 如果尚未安装 Docker&#xff0c;可以参考以下命令安装&…
阅读更多...
PCIe_Host驱动分析_设备枚举

PCIe_Host驱动分析_设备枚举

往期内容 本文章相关专栏往期内容&#xff0c;PCI/PCIe子系统专栏&#xff1a; 嵌入式系统的内存访问和总线通信机制解析、PCI/PCIe引入 深入解析非桥PCI设备的访问和配置方法 PCI桥设备的访问方法、软件角度讲解PCIe设备的硬件结构 深入解析PCIe设备事务层与配置过程 PCIe的三…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023