private_heavy_hitters.ipynb

Kernel: Python 3

Copyright 2022 The TensorFlow Authors.

#@title Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

隐私保护频繁项

在 TensorFlow.org 上查看

在 Google Colab 中运行

在 Github 上查看源代码

下载笔记本

注：本 Colab 已经过验证，可与最新发布版本的 tensorflow_federated pip 软件包一起使用。本 Colab 可能不会更新为适用于 main。

本教程展示了如何使用 tff.analytics.heavy_hitters.iblt.build_iblt_computation API 来构建联合分析计算，以发现群体中最常见的字符串（隐私保护频繁项）。

环境设置

请运行以下代码来确保您的环境已正确设置。如果未看到问候语，请参阅安装指南查看说明。

#@test {"skip": true}

# tensorflow_federated_nightly also bring in tf_nightly, which
# can causes a duplicate tensorboard install, leading to errors.
!pip install --quiet tensorflow-text-nightly
!pip install --quiet --upgrade tensorflow-federated

import collections

import numpy as np
import tensorflow as tf
import tensorflow_federated as tff
import tensorflow_text as tf_text

np.random.seed(0)
tff.backends.test.set_sync_test_cpp_execution_context()

tff.federated_computation(lambda: 'Hello, World!')()

b'Hello, World!'

背景：联合分析中的隐私保护频繁项

考虑以下设置：每个客户端都有一个字符串列表，每个字符串都来自一个开集，这意味着它可以是任意的。目标是在联合设置中隐秘发现最常见的字符串（频繁项）及其数量。本 Colab 演示了使用以下隐私属性解决此问题的解决方案：

• 安全聚合：计算聚合字符串计数，使服务器不可能学习任何客户端的单个值。请参阅 tff.federated_secure_sum，以了解更多信息。

• 差分隐私 (DP)：一种广泛使用的方法，用于限制和量化分析中敏感数据的隐私泄露。可以将用户级中央 DP 应用于频繁项结果。

安全聚合 API tff.federated_secure_sum 支持整数向量的线性和。如果字符串来自大小为 n 的闭集，那么将每个客户端的字符串编码为大小为 n 的向量相当轻松：让向量的索引 i 处的值是闭集中第 i 个字符串的计数。随后，您可以安全地对所有客户端的向量求和，以获得整个群体中的字符串计数。但是，如果字符串来自一个开集，那么如何适当地对它们进行编码以实现安全求和并非显而易见。在这项工作中，可以将字符串编码为可逆布隆查找表 (IBLT)，这是一种概率数据结构，能够以高效的方式对大型（或开放）域中的项目进行编码。IBLT sketch 可以线性求和，因此它们与安全求和兼容。

可以使用 tff.analytics.heavy_hitters.iblt.build_iblt_computation 创建将每个客户端的本地字符串编码为 IBLT 结构的 TFF 计算。这些结构通过加密安全多方计算协议安全地汇总到服务器可以解码的聚合 IBLT 结构中。随后，服务器可以返回排名靠前的频繁项。以下部分展示了如何使用此 API 创建 TFF 计算并使用莎士比亚数据集运行模拟。

加载并预处理联合莎士比亚数据

莎士比亚数据集包含莎士比亚戏剧的角色台词。本示例中选择了一部分角色（即客户端）。预处理器将每个角色的台词转换为字符串列表，并删除任何只有标点符号或符号的字符串。

# Load the simulation data.
source, _ = tff.simulation.datasets.shakespeare.load_data()

# Preprocessing function to tokenize a line into words.
def tokenize(ds):
  """Tokenizes a line into words with alphanum characters."""
  def extract_strings(example):
    return tf.expand_dims(example['snippets'], 0)

  def tokenize_line(line):
    return tf.data.Dataset.from_tensor_slices(tokenizer.tokenize(line)[0])

  def mask_all_symbolic_words(word):
    return tf.math.logical_not(
        tf_text.wordshape(word, tf_text.WordShape.IS_PUNCT_OR_SYMBOL))

  tokenizer = tf_text.WhitespaceTokenizer()
  ds = ds.map(extract_strings)
  ds = ds.flat_map(tokenize_line)
  ds = ds.map(tf_text.case_fold_utf8)
  ds = ds.filter(mask_all_symbolic_words)
  return ds

batch_size = 5

def client_data(n: int) -> tf.data.Dataset:
  return tokenize(source.create_tf_dataset_for_client(
      source.client_ids[n])).batch(batch_size)

# Pick a subset of client devices to participate in the computation.
dataset = [client_data(n) for n in range(10)]

模拟

要运行模拟以发现莎士比亚数据集中最常见的单词（频繁项），首先需要使用 tff.analytics.heavy_hitters.iblt.build_iblt_computation API 和以下参数创建一个 TFF 计算：

• capacity：IBLT sketch 的容量。此数字应该大致等于一轮计算中可能出现的唯一字符串的总数。默认值为 1000。如果此数字太小，解码可能由于散列值的冲突而失败。如果此数字太大，它会消耗更多不必要的内存。

• string_max_bytes：IBLT 中字符串的最大长度。默认值为 10。必须为正数。长度超过 string_max_bytes 的字符串将被截断。

• max_words_per_user：每个客户端允许贡献的最大字符串数。如果不是 None，则必须是正整数。默认值为 None，这意味着所有客户端都贡献了其所有字符串。

• max_heavy_hitters：要返回的最大项目数。如果解码结果的项目数超过此数量，将按估计计数递减排序并返回前 max_heavy_hitters 个项目。默认值为 None，这意味着返回结果中的所有频繁项。

• secure_sum_bitwidth：用于安全求和的位宽。默认值为 None，该值会禁用安全求和。如果不是 None，则必须处于 [1,62] 范围内。请参阅 tff.federated_secure_sum。

• multi_contribution：是否允许每个客户端为每个唯一单词贡献多个计数或仅贡献一个计数。默认值为 True。当需要差分隐私时，此参数可以改善效果。

• batch_size：每批数据集中的元素数量。默认值为 1，表示输入数据集由 tf.data.Dataset.batch(1) 处理。必须是正整数。

max_words_per_user = 8
iblt_computation = tff.analytics.heavy_hitters.iblt.build_iblt_computation(
    capacity=100,
    string_max_bytes=20,
    max_words_per_user=max_words_per_user,
    max_heavy_hitters=10,
    secure_sum_bitwidth=32,
    multi_contribution=False,
    batch_size=batch_size)

现在已准备好使用 TFF 计算 iblt_computation 和预处理输入数据集运行模拟。iblt_computation 的输出有以下四个属性：

• 客户端：参与计算的客户端的标量值。

• heavy_hitters：聚合的频繁项列表。

• heavy_hitters_counts：聚合的频繁项的计数列表。

• num_not_decoded：未成功解码的字符串的标量值。

def run_simulation(one_round_computation: tff.Computation, dataset):
  output = one_round_computation(dataset)
  heavy_hitters = output.heavy_hitters
  heavy_hitters_counts = output.heavy_hitters_counts
  heavy_hitters = [word.decode('utf-8', 'ignore') for word in heavy_hitters]

  results = {}
  for index in range(len(heavy_hitters)):
    results[heavy_hitters[index]] = heavy_hitters_counts[index]
  return output.clients, dict(results)

clients, result = run_simulation(iblt_computation, dataset)
print(f'Number of clients participated: {clients}')
print('Discovered heavy hitters and counts:')
print(result)

Number of clients participated: 10
Discovered heavy hitters and counts:
{'to': 8, 'the': 8, 'and': 7, 'you': 4, 'i': 4, 'a': 3, 'he': 3, 'your': 3, 'is': 3, 'of': 2}

使用差分隐私的隐私保护频繁项

为了获得具有中心 DP 的隐私保护频繁项，将 DP 机制应用于开集直方图。其思想是向聚合直方图中的字符串计数添加噪声，然后只保留计数高于某个阈值的字符串。噪声和阈值取决于 (epsilon, delta)-DP 预算，有关详细算法和证明，请参阅此文档。作为后处理步骤，噪声计数四舍五入为整数，这不会削弱 DP 保证。请注意，当需要 DP 时，您会发现更少的频繁项。这是因为阈值步骤会过滤掉计数较低的字符串。

iblt_computation = tff.analytics.heavy_hitters.iblt.build_iblt_computation(
    capacity=100,
    string_max_bytes=20,
    max_words_per_user=max_words_per_user,
    secure_sum_bitwidth=32,
    multi_contribution=False,
    batch_size=batch_size)

clients, result = run_simulation(iblt_computation, dataset)

# DP parameters
eps = 20
delta = 0.01

# Calculating scale for Laplace noise
scale = max_words_per_user / eps

# Calculating the threshold
tau = 1 + (max_words_per_user / eps) * np.log(max_words_per_user / (2 * delta))

result_with_dp = {}
for word in result:
  noised_count = result[word] + np.random.laplace(scale=scale)
  if noised_count >= tau:
    result_with_dp[word] = int(noised_count)
print(f'Discovered heavy hitters and counts with central DP:')
print(result_with_dp)

Discovered heavy hitters and counts with central DP:
{'the': 8, 'you': 4, 'to': 7, 'tear': 3, 'and': 7, 'i': 3}