Kafka Streams 中的数据屏蔽技术

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

数据屏蔽简介

什么是数据屏蔽？

数据屏蔽将敏感数据转换为安全、隐藏的形式，确保原始数据保持隐藏状态，同时保留其在特定用例中的可用性。它广泛用于测试、分析和生产环境等场景，以保护敏感信息。

Kafka Streams 中数据屏蔽的重要性

Kafka Streams 实时处理大量数据，通常包括个人身份信息 (PII)、信用卡详细信息和医疗记录等敏感信息。在 Kafka Streams 中实施数据屏蔽可以确保

数据隐私： 保护敏感数据免遭未经授权的访问。
合规性： 遵守数据保护法规和指南。
运营安全： 最大限度地降低数据泄露的风险。

静态数据屏蔽 (SDM)

静态数据屏蔽应用于静态数据。此技术通常用于在共享或进一步处理文件或数据库中存储的数据集之前对其进行清理。

示例：屏蔽文件中的敏感数据

此示例演示了如何读取 CSV 文件，屏蔽敏感数据（如社会安全号码），并将屏蔽后的数据保存到新文件中。

输入文件 (data.csv)

id,name,ssn
1,John Doe,123-45-6789
2,Jane Smith,987-65-4321
3,Bob Johnson,111-22-3333   

代码实现

import pandas as pd

input_file = "data.csv"
output_file = "masked_data.csv"
data = pd.read_csv(input_file)
data['ssn'] = data['ssn'].apply(lambda x: "XXX-XX-XXXX")
data.to_csv(output_file, index=False)
print("Data masking complete. Masked data saved to", output_file)   

输出文件 (masked_data.csv)

Data Masking Techniques in Kafka Streams

动态数据屏蔽 (DDM)

动态数据屏蔽应用于传输中的数据，使其非常适合 Kafka Streams。此示例演示了如何实时屏蔽 Kafka 中的敏感信息。

示例：使用 Kafka Streams 实时屏蔽 SSN

此示例展示了在 Kafka 中处理数据流时如何动态屏蔽 SSN。

前提条件

必须安装并运行 Kafka。
创建 Kafka 主题

kafka-topics.sh --create --topic input-topic --bootstrap-server localhost:9092
kafka-topics.sh --create --topic output-topic --bootstrap-server localhost:9092   

将示例数据生产者发送到 input-topic

kafka-console-producer.sh --topic input-topic --bootstrap-server localhost:9092
>{"id":1,"name":"John Doe","ssn":"123-45-6789"}
>{"id":2,"name":"Jane Smith","ssn":"987-65-4321"}   

Kafka Streams Java 应用程序

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.util.Properties;

public class DynamicDataMasking {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "dynamic-masking-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> inputStream = builder.stream("input-topic");
        KStream<String, String> maskedStream = inputStream.mapValues((ValueMapper<String, String>) value -> {
            try {
                ObjectMapper mapper = new ObjectMapper();
                ObjectNode jsonNode = (ObjectNode) mapper.readTree(value);

                if (jsonNode.has("ssn")) {
                    jsonNode.put("ssn", "XXX-XX-XXXX");
                }

                return jsonNode.toString();
            } catch (Exception e) {
                e.printStackTrace();
                return value;
            }
        });

        maskedStream.to("output-topic");

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

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}   

从 output-topic 消费消息

 
kafka-console-consumer.sh --topic output-topic --from-beginning --bootstrap-server localhost:9092   

预期输出

数据脱敏技术

1. 擦除

擦除涉及用固定模式替换敏感信息以进行模糊处理。

示例：擦除信用卡号

输入

实施

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import java.util.Properties;
public class RedactionExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "redaction-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> stream = builder.stream("input-topic");

        KStream<String, String> redactedStream = stream.mapValues(value -> {
            try {
                ObjectMapper mapper = new ObjectMapper();
                ObjectNode jsonNode = (ObjectNode) mapper.readTree(value);

                if (jsonNode.has("credit_card")) {
                    jsonNode.put("credit_card", "XXXX-XXXX-XXXX-XXXX");
                }

                return jsonNode.toString();
            } catch (Exception e) {
                e.printStackTrace();
                return value;
            }
        });

        redactedStream.to("output-topic");
        KafkaStreams streams = new KafkaStreams(builder.build(), props);
        streams.start();
    }
}   

输出

2. 替换

替换用逼真但虚构的值替换数据。

示例：用假名替换姓名

输入

实施

import java.util.Map;
import java.util.HashMap;

public class SubstitutionExample {
    private static final Map<String, String> pseudonyms = new HashMap<>();

    static {
        pseudonyms.put("John Doe", "James Smith");
        pseudonyms.put("Jane Doe", "Emily Johnson");
    }

    public static String substituteName(String value) {
        return pseudonyms.getOrDefault(value, "Anonymous");
    }
RedactionExample)
}   

输出

3. 标记化

标记化用令牌替换敏感字段，从而实现安全映射。

示例：标记化 SSN

输入

实施

import java.util.UUID;
import java.util.Map;
import java.util.HashMap;
public class TokenizationExample {
    private static final Map<String, String> tokenStore = new HashMap<>();

    public static String tokenize(String value) {
        return tokenStore.computeIfAbsent(value, k -> UUID.randomUUID().toString());
    }
}   

输出

4. 加密

加密通过将重要数据更改为秘密代码来保护它。

示例：加密电子邮件地址

输入

实施

import javax.crypto.Cipher;
import javax.crypto.KeyGenerator;
import javax.crypto.SecretKey;
import javax.crypto.spec.SecretKeySpec;
import java.util.Base64;

public class EncryptionExample {
    private static SecretKey secretKey;

    static {
        try {
            KeyGenerator keyGen = KeyGenerator.getInstance("AES");
            keyGen.init(128);
            secretKey = keyGen.generateKey();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static String encrypt(String data) {
        try {
            Cipher cipher = Cipher.getInstance("AES");
            cipher.init(Cipher.ENCRYPT_MODE, secretKey);
            return Base64.getEncoder().encodeToString(cipher.doFinal(data.getBytes()));
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }

    // Apply this encryption logic in the Kafka Streams pipeline
}   

输出

5. 随机化

随机化随机排列数据集中的数据，以模糊原始关系。

示例：随机化电话号码

输入

[
    {"id": 1, "phone": "123-456-7890"},
    {"id": 2, "phone": "987-654-3210"},
    {"id": 3, "phone": "555-555-5555"}
]   

实施

import java.util.Collections;
import java.util.List;
import java.util.ArrayList;

public class ShufflingExample {
    public static List<String> shuffle(List<String> values) {
        Collections.shuffle(values);
        return values;
    }

    // Apply shuffling logic in the Kafka Streams pipeline
}   

输出

6. 置空屏蔽

置空用 null 替换敏感字段。

示例：置空 SSN

输入

实施

public class NullificationExample {
    public static String nullify(String value) {
        return null;
    }

    // Apply this nullification logic in the Kafka Streams pipeline
}   

输出

这些技术中的每一种都提供了一种在保持一定数据可用性的同时保护敏感信息的方法，具体取决于用例。

在 Kafka Streams 中实现数据屏蔽

设置 Kafka 和 Kafka Streams

在实现数据屏蔽之前，请确保您已设置 Kafka 和 Kafka Streams。

安装 Kafka： 从 https://kafka.apache.org/downloads Kafka Downloads 下载并设置 Apache Kafka。
创建主题： 为 Kafka Streams 处理创建输入和输出主题。
kafka-topics.sh --create --topic input-topic --bootstrap-server localhost:9092
生产示例数据： 将示例 JSON 数据生产到 input topic。
kafka-console-producer.sh --topic input-topic --bootstrap-server localhost:9092

使用 Kafka Streams 进行数据屏蔽

下面是一个使用 Kafka Streams 屏蔽 JSON 数据中敏感字段的示例。

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.ValueMapper;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;

import java.util.Properties;

public class DataMaskingExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "data-masking-app");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> inputStream = builder.stream("input-topic");

        KStream<String, String> maskedStream = inputStream.mapValues((ValueMapper<String, String>) value -> {
            try {
                ObjectMapper mapper = new ObjectMapper();
                ObjectNode jsonNode = (ObjectNode) mapper.readTree(value);
                if (jsonNode.has("ssn")) {
                    jsonNode.put("ssn", "XXX-XX-XXXX"); // Redacting SSN
                }
                return jsonNode.toString();
            } catch (Exception e) {
                e.printStackTrace();
                return value;
            }
        });

        maskedStream.to("output-topic");

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

        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}   

说明

从输入主题读取： inputStream 从 input-topic 读取未屏蔽的数据。
屏蔽敏感字段： mapValues 方法通过擦除 SSN 字段来修改 JSON 数据。
写入输出主题： 屏蔽后的数据被写入 output-topic。

运行应用程序

编译并运行 Java 应用程序，然后从 output topic 消费消息以验证屏蔽效果。

kafka-console-consumer.sh --topic output-topic --from-beginning --bootstrap-server localhost:9092

预期输出

Kafka Streams 高级数据屏蔽

集成第三方库

可以集成 DataMasker 或 Apache Avro 等库来实现更复杂的数据屏蔽。

使用查找表的实时动态屏蔽

可以通过使用查找表根据特定规则或角色替换数据来增强动态屏蔽。

屏蔽操作的审计和日志记录

集成日志框架（例如 Logback 或 Log4j）以跟踪屏蔽操作，以确保合规性和调试。

挑战与最佳实践

性能影响

数据屏蔽可能会在流处理中引入延迟。优化序列化和反序列化操作以最大程度地减少性能影响。

平衡安全性和可用性

确保屏蔽后的数据对下游应用程序保持足够的可用性，同时保护敏感信息。

确保合规性

定期审计屏蔽逻辑，以符合不断变化的法规和标准。

下一主题节能 Kafka 集群

Kafka Streams 中的数据屏蔽技术

数据屏蔽简介

什么是数据屏蔽？

Kafka Streams 中数据屏蔽的重要性

静态数据屏蔽 (SDM)

动态数据屏蔽 (DDM)

数据脱敏技术

1. 擦除

2. 替换

3. 标记化

4. 加密

5. 随机化

6. 置空屏蔽

在 Kafka Streams 中实现数据屏蔽

设置 Kafka 和 Kafka Streams

Kafka Streams 高级数据屏蔽

集成第三方库

使用查找表的实时动态屏蔽

屏蔽操作的审计和日志记录

挑战与最佳实践

性能影响

平衡安全性和可用性

确保合规性

联系信息

关注我们

教程

面试题

在线编译器

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 Streams 中的数据屏蔽技术

数据屏蔽简介

什么是数据屏蔽？

Kafka Streams 中数据屏蔽的重要性

静态数据屏蔽 (SDM)

动态数据屏蔽 (DDM)

数据脱敏技术

1. 擦除

2. 替换

3. 标记化

4. 加密

5. 随机化

6. 置空屏蔽

在 Kafka Streams 中实现数据屏蔽

设置 Kafka 和 Kafka Streams

Kafka Streams 高级数据屏蔽

集成第三方库

使用查找表的实时动态屏蔽

屏蔽操作的审计和日志记录

挑战与最佳实践

性能影响

平衡安全性和可用性

确保合规性

相关帖子

使用 Kafka 进行事件驱动的网络安全

使用 Kafka 在零售业中实现超个性化

Kafka-to-Cosmos DB 管道中的模式演进和序列化

Kafka 延迟

Kafka 中的内存管理

Kafka 控制器

Kafka Streams API 中的数据流和推荐

Kafka 连接器

使用 Kafka 的 BigQuery 数据仓库

Kafka 中的事件流版本控制

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器