使用 Kafka 进行混合云数据编排

2025年5月15日 | 阅读时间 6 分钟

引言

混合云正变得越来越流行，因为它结合了公共云的灵活性和低成本，以及私有云的控制和安全性。在这种环境中，数据编排涉及数据在这两个域之间的无缝移动、转换和处理。Apache Kafka 因其可扩展性、可靠性和灵活性而成为实现混合云数据编排的强大工具。

本文档探讨了使用 Kafka 进行混合云数据编排的架构、理论和实现，包括带有输出的示例程序。

理解混合云数据编排

什么是混合云？

混合云是一种同时使用私有云和公共云的设置，允许数据和应用程序在它们之间移动。这有助于组织获得两者的优势。

混合云编排中的挑战

数据延迟：管理跨地域分布式环境的延迟。
数据安全：确保安全数据传输。
数据一致性：维护私有云和公共云中数据的一致性。
可扩展性：处理可变工作负载。

Kafka 在混合云编排中的作用

Kafka 通过以下方式促进混合云编排：

充当私有云和公共云之间流式传输数据的中心枢纽。
实现实时数据处理。
支持高吞吐量和容错数据管道。
使用连接器与其他系统无缝集成。

混合云 Kafka 架构的组成部分

1. 内部数据中心

内部数据中心通常托管 Kafka 代理和处理敏感数据的关键应用程序。这些可能包括处理金融交易、个人信息或专有业务逻辑的应用程序。在内部托管 Kafka 代理可确保数据主权和符合法规要求。

示例

Configuration for on-premise Kafka broker:
broker.id=1
log.dirs=/var/lib/kafka/data
zookeeper.connect=localhost:2181
listeners=PLAINTEXT://on-prem-host:9092
num.network.threads=3
num.io.threads=8
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
Start the Kafka broker:
bin/kafka-server-start.sh config/server.properties

2. 公共云提供商

公共云提供商托管 Kafka 代理和利用可伸缩资源的云原生应用程序。此设置有助于弹性并易于与其他云服务集成。

示例

Configuration for public cloud Kafka broker:
broker.id=2
log.dirs=/var/lib/kafka/data
zookeeper.connect=cloud-zookeeper:2181
listeners=PLAINTEXT://cloud-host:9092
num.network.threads=3
num.io.threads=8
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
Start the Kafka broker:
bin/kafka-server-start.sh config/server.properties

预期输出

云 Kafka 代理成功启动，日志确认监听器已就绪

3. Kafka Connect

Kafka Connect 在环境之间搭建数据源和数据槽的桥梁，实现内部数据中心和公共云之间数据流的无缝集成。它促进数据复制、转换和流式传输。

示例

创建 Kafka Connect 配置文件 (replicator-config.json)

{
  "name": "hybrid-cloud-connector",
  "config": {
    "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector",
    "tasks.max": "4",
    "src.kafka.bootstrap.servers": "on-prem-host:9092",
    "dest.kafka.bootstrap.servers": "cloud-host:9092",
    "topic.whitelist": "critical-data",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter"
  }
}

使用配置启动 Kafka Connect

bin/connect-standalone.sh config/connect-standalone.properties replicator-config.json

预期输出

Kafka Connect 服务启动，日志确认主题复制

INFO Replicating topic critical-data from on-prem-host:9092 to cloud-host:9092

4. Schema Registry

Schema Registry 确保混合设置中数据格式的一致性，防止序列化和反序列化问题。它提供了一个集中的模式管理存储库。

示例

使用 cURL 注册模式

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
    --data '{"schema": "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}, {\"name\": \"age\", \"type\": \"int\"}]}"}' \
    http://schema-registry:8081/subjects/User-value/versions

输出

Schema Registry 存储模式，返回一个模式 ID

Hybrid Cloud Data Orchestration Using Kafka

5. 监控和管理工具

Confluent Control Center、Prometheus 和 Grafana 等监控和管理工具在观察混合云设置中的 Kafka 基础设施方面发挥着至关重要的作用。

Prometheus 示例

Kafka 指标的 Prometheus 配置 (prometheus.yml)

scrape_configs:
  - job_name: 'kafka'
    static_configs:
      - targets: ['on-prem-host:9092', 'cloud-host:9092']
Start Prometheus:
./prometheus --config.file=prometheus.yml

Grafana 面板

将 Kafka 指标面板导入 Grafana 以可视化指标，例如：

主题延迟
分区健康状况
代理 CPU 使用率

架构模式

主动-主动

在主动-主动模式中，Kafka 集群在私有云和公共云中运行。这确保了高可用性和负载均衡。消息使用 MirrorMaker 等工具进行双向复制。

示例

MirrorMaker configuration for active-active replication:
bin/kafka-mirror-maker.sh \
  --consumer.config config/on-prem-consumer.properties \
  --producer.config config/cloud-producer.properties \
  --whitelist '.*'

输出

主动-被动

在主动-被动模式中，主集群在一个环境中运行，而辅助集群充当备份。

示例

MirrorMaker configuration for active-passive replication:
bin/kafka-mirror-maker.sh \
  --consumer.config config/on-prem-consumer.properties \
  --producer.config config/cloud-producer.properties \
  --whitelist 'critical-data'

输出

事件复制

集群之间的数据复制确保了一致性和灾难恢复。Kafka 的 MirrorMaker 或 Confluent Replicator 实现了这一点。

示例

MirrorMaker configuration for replication:
bin/kafka-mirror-maker.sh \
  --consumer.config config/on-prem-consumer.properties \
  --producer.config config/cloud-producer.properties \
  --whitelist '.*'

使用 Kafka 实施混合云数据编排

步骤 1：设置 Kafka

安装 Kafka：按照私有和公共环境的 Kafka 安装步骤进行操作。
配置代理：确保 Kafka 代理在混合设置中正确联网。

代理的示例配置

broker.id=1
log.dirs=/tmp/kafka-logs
zookeeper.connect=localhost:2181
listeners=PLAINTEXT://private-cloud-broker:9092
advertised.listeners=PLAINTEXT://public-cloud-broker:9092
num.network.threads=3
num.io.threads=8
log.retention.hours=168

步骤 2：创建主题

创建用于混合环境的 Kafka 主题。

示例命令

bin/kafka-topics.sh --create --topic hybrid-topic --bootstrap-server private-cloud-broker:9092 --partitions 3 --replication-factor 2

步骤 3：数据生产者

私有云生产者示例

from kafka import KafkaProducer
import json
producer = KafkaProducer(
    bootstrap_servers='private-cloud-broker:9092',  # Kafka broker address
    value_serializer=lambda v: json.dumps(v).encode('utf-8')  # Serialize data to JSON format
)
data = {"key": "value", "timestamp": "2024-12-24T12:00:00Z"}
try:
    future = producer.send('hybrid-topic', value=data)
    # Block until the message is sent (optional, for error handling)
    result = future.get(timeout=10)  
    print(f"Data sent to 'hybrid-topic': {data}")
except Exception as e:
    print(f"Error sending data: {e}")

# Close the producer
producer.close()

输出

步骤 4：数据消费者

公共云消费者示例

from kafka import KafkaConsumer
import json
consumer = KafkaConsumer(
    'hybrid-topic',  # Topic to subscribe to
    bootstrap_servers='public-cloud-broker:9092',  # Kafka broker address
    auto_offset_reset='earliest',  # Start reading messages from the beginning
    enable_auto_commit=True,  # Automatically commit offsets
    group_id='hybrid-consumer-group',  # Consumer group ID
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))  # Deserialize JSON messages
)

print("Listening to messages on topic 'hybrid-topic'...")
try:
    for message in consumer:
        print(f"Received: {message.value}")
except KeyboardInterrupt:
    print("\nConsumer stopped.")
except Exception as e:
    print(f"Error while consuming messages: {e}")
finally:
    consumer.close()
    print("Consumer closed.")    

输出

步骤 5：使用 Kafka Streams 进行数据转换

示例程序

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

public class DataTransformation {
    public static void main(String[] args) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> source = builder.stream("hybrid-topic");
        source.mapValues(value -> value.toUpperCase())
              .to("transformed-topic");

        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}

输出

步骤 6：使用连接器集成云服务

示例：AWS S3 Sink 连接器

{
  "name": "s3-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "topics": "hybrid-topic",
    "s3.bucket.name": "my-s3-bucket",
    "s3.region": "us-east-1",
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "org.apache.kafka.connect.storage.StringConverter"
  }
}

处理挑战

1. 安全挑战

a) 使用 SSL/TLS 保护数据

SSL/TLS 确保混合云设置中 Kafka 组件之间的加密通信。

生产者示例

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers='cloud-broker:9093',
    security_protocol='SSL',
    ssl_cafile='/path/to/ca.pem',
    ssl_certfile='/path/to/client-cert.pem',
    ssl_keyfile='/path/to/client-key.pem',
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

data = {"key": "hybrid-data", "timestamp": "2024-12-24T12:00:00Z"}
producer.send('hybrid-topic', value=data)
print("Secure message sent to hybrid-topic.")
producer.close()

输出

消费者示例

from kafka import KafkaConsumer
import json

consumer = KafkaConsumer(
    'hybrid-topic',
    bootstrap_servers='on-prem-broker:9093',
    security_protocol='SSL',
    ssl_cafile='/path/to/ca.pem',
    ssl_certfile='/path/to/client-cert.pem',
    ssl_keyfile='/path/to/client-key.pem',
    value_deserializer=lambda v: json.loads(v.decode('utf-8'))
)

print("Listening to secure messages...")
for message in consumer:
    print(f"Received secure message: {message.value}")

输出

b) 使用 SASL 进行身份验证

SASL 为混合云添加了一个身份验证层。

生产者示例

from kafka import KafkaProducer
producer = KafkaProducer(
    bootstrap_servers='cloud-broker:9094',
    security_protocol='SASL_SSL',
    sasl_mechanism='PLAIN',
    sasl_plain_username='admin',
    sasl_plain_password='password123',
    value_serializer=lambda v: v.encode('utf-8')
)
producer.send('hybrid-topic', value='Authenticated message')
print("Message sent with SASL authentication.")
producer.close()

2. 数据延迟挑战

a) 优化生产者以实现低延迟

分区、压缩和批处理减少数据延迟。

生产者示例

from kafka import KafkaProducer
producer = KafkaProducer(
    bootstrap_servers='hybrid-broker:9092',
    value_serializer=lambda v: v.encode('utf-8'),
    acks='all',  # Ensure durability
    compression_type='snappy',  # Lightweight compression
    batch_size=65536,  # Batch size for better throughput
    linger_ms=5  # Small delay to collect more messages
)
for i in range(100):
    producer.send('low-latency-topic', value=f"Message {i}")
print("Low-latency messages sent.")
producer.close()

b) 优化消费者以实现高吞吐量

调整获取设置可提高消费者性能。

消费者示例

from kafka import KafkaConsumer
consumer = KafkaConsumer(
    'low-latency-topic',
    bootstrap_servers='hybrid-broker:9092',
    group_id='latency-group',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    fetch_min_bytes=1024,  # Min data to fetch in one poll
    fetch_max_wait_ms=500  # Max wait time for fetching
)
print("Listening to low-latency topic...")
for message in consumer:
    print(f"Received: {message.value.decode('utf-8')}")

输出

3. 故障恢复挑战

a) 使用日志压缩

日志压缩确保只保留数据的最新版本。

生产者示例

producer = KafkaProducer(
    bootstrap_servers='hybrid-broker:9092',
    key_serializer=lambda k: k.encode('utf-8'),
    value_serializer=lambda v: v.encode('utf-8')
)
producer.send('compaction-topic', key='user1', value='Data1')
producer.send('compaction-topic', key='user1', value='Updated Data1')
print("Messages sent with log compaction.")
producer.close()

消费者示例

from kafka import KafkaConsumer
consumer = KafkaConsumer(
    'compaction-topic',
    bootstrap_servers='hybrid-broker:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='compaction-group'
)
print("Listening to compacted topic...")
for message in consumer:
    print(f"Received: {message.key.decode('utf-8')} -> {message.value.decode('utf-8')}")

输出

b) 使用消费者组确保高可用性

消费者组分配负载并确保容错。

消费者组示例

from kafka import KafkaConsumer
consumer = KafkaConsumer(
    'ha-topic',
    bootstrap_servers='hybrid-broker:9092',
    group_id='ha-consumer-group',
    auto_offset_reset='earliest',
    enable_auto_commit=True
)	
print("High availability consumer group active...")
for message in consumer:
    print(f"HA Consumer received: {message.value.decode('utf-8')}")

输出

c) 使用 MirrorMaker 进行灾难恢复

MirrorMaker 在内部部署和云集群之间复制数据。

MirrorMaker 命令

bin/kafka-mirror-maker.sh \
  --consumer.config config/on-prem-consumer.properties \
  --producer.config config/cloud-producer.properties \
  --whitelist '.*'

消费者配置 (on-prem-consumer.properties)

bootstrap.servers=on-prem-broker:9092
group.id=mirror-group
key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

生产者配置 (cloud-producer.properties)

bootstrap.servers=cloud-broker:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

下一个主题Kafka 在增强现实 (AR) 应用程序中的作用

使用 Kafka 进行混合云数据编排

引言

理解混合云数据编排

什么是混合云？

Kafka 在混合云编排中的作用

混合云 Kafka 架构的组成部分

1. 内部数据中心

2. 公共云提供商

3. Kafka Connect

4. Schema Registry

5. 监控和管理工具

架构模式

主动-主动

主动-被动

事件复制

使用 Kafka 实施混合云数据编排

步骤 1：设置 Kafka

步骤 2：创建主题

步骤 3：数据生产者

步骤 4：数据消费者

步骤 5：使用 Kafka Streams 进行数据转换

步骤 6：使用连接器集成云服务

处理挑战

1. 安全挑战

2. 数据延迟挑战

3. 故障恢复挑战

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

Kafka 教程

Kafka 安装

Kafka CLI

Kafka 编程

实时示例

Kafka 监控

Kafka Connect

Kafka Streams

杂项

使用 Kafka 进行混合云数据编排

引言

理解混合云数据编排

什么是混合云？

Kafka 在混合云编排中的作用

混合云 Kafka 架构的组成部分

1. 内部数据中心

2. 公共云提供商

3. Kafka Connect

4. Schema Registry

5. 监控和管理工具

架构模式

主动-主动

主动-被动

事件复制

使用 Kafka 实施混合云数据编排

步骤 1：设置 Kafka

步骤 2：创建主题

步骤 3：数据生产者

步骤 4：数据消费者

步骤 5：使用 Kafka Streams 进行数据转换

步骤 6：使用连接器集成云服务

处理挑战

1. 安全挑战

2. 数据延迟挑战

3. 故障恢复挑战

相关帖子

Kafka 实时分析设置

Kafka 到 Cosmos DB 数据流中的延迟优化

针对基因组数据流优化的 Kafka

系统迁移中的 Kafka 组件

节能 Kafka 集群

监控和调整 Kafka 到 Cosmos DB 管道

Kafka 和数字孪生架构

事件驱动架构：Kafka 到 Cosmos DB 集成

Kafka 中的数据捕获

动态主题分区策略

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器