使用 PipelineDP 计算私密统计信息

1. 准备工作

你可能会认为，汇总的统计信息不会泄露与其相关的个人的任何信息。但是，攻击者可以通过多种方式从汇总统计信息中获知有关个人的敏感信息。

在此 Codelab 中，您将学习如何使用 PipelineDP 的差分隐私汇总生成隐私统计信息，以保护个人的保护隐私。PipelineDP 是一个 Python 框架，可让您通过批处理系统（如 Apache Spark 和 Apache Beam）将差分隐私应用于大型数据集。如需详细了解如何在 Go 中计算差分隐私统计信息，请参阅 Privacy on Beam Codelab。

隐私是指生成输出时不会泄露数据中当事人的任何私密信息。您可以通过差分隐私来实现这一结果，这是针对匿名化的强烈隐私概念，它是为保护用户隐私而跨多个用户进行数据汇总的过程。所有匿名化方法都使用汇总方法，但并非所有汇总方法都能实现匿名化。另一方面，差分隐私可以针对信息泄露和隐私提供可衡量的保证。

前提条件

• 熟悉 Python
• 熟悉基本的数据汇总
• 拥有使用 Pandas、Spark 和 Beam 的经验

学习内容

• 差分隐私基础知识
• 如何使用 PipelineDP 计算差分隐私摘要统计信息
• 如何使用其他隐私参数和实用程序参数微调结果

所需条件

• 如果您想在自己的环境中运行 Codelab，则需要在计算机上安装 Python 3.7 或更高版本。
• 如果您想在没有自己的环境的情况下完成此 Codelab，则需要能够访问 Colaboratory。

2. 了解差分隐私

为了更好地了解差分隐私，我们来看看下面这个简单的示例。

假设您在一家线上时装零售商的营销部门工作，想要了解自己的哪些商品最有可能售出。

此图表显示了客户在访问该商店的网站时首先看到的产品：T 恤、套头衫、袜子或牛仔裤。T 恤是最受欢迎的商品，而袜子是最不受欢迎的商品。

这看起来有用，但有个问题。如果您想考虑更多信息（例如客户是否购买了商品或者他们后来浏览了哪款产品），那么您就有可能会泄露数据中的个人信息。

此图表显示，只有一位客户先看了套头衫，然后实际购买了商品：

从隐私保护的角度来看，这并不是好事。匿名化的统计信息不应透露具体的贡献，那么您该怎么做？您在条形图中添加随机噪声，以降低条形图的准确性！

此条形图并非完全准确，但依然很有用，并且不会透露具体的贡献：

差分隐私是指添加适量的随机噪声，以掩盖个人贡献。

此示例过于简单。差分隐私的正确实现涉及更多领域，并且带来了许多意想不到的细微差别。与加密类似，自行创建差分隐私实现可能不是一个好主意。您可以改用 PipelineDP。

3. 下载并安装 PipelineDP

您无需安装 PipelineDP 即可按照此 Codelab 中的说明操作，因为您可以在本文档中找到所有相关代码和图表。

如需使用 PipelineDP，可以自行运行它，也可以稍后使用：

• 下载并安装 PipelineDP：

pip install pipeline-dp

如果要使用 Apache Beam 运行该示例，请执行以下操作：

• 下载并安装 Apache Beam：

pip install apache_beam

您可以在 PipelineDP/examples/codelab/ 目录中找到此 Codelab 的代码和数据集。

4. 计算每个产品首次浏览的转化指标

假设您在一家线上时装零售商工作，想要了解首次查看时，哪些不同的产品类别带来的转化次数和价值最高。您想与营销代理机构以及其他内部团队分享这些信息，但又希望防止任何客户的信息泄露。

如需计算用户在网站上查看的每件产品的转化指标，请执行以下操作：

1. 查看 PipelineDP/examples/codelab/ 目录中有关您网站的访问情况的模拟数据集。

此屏幕截图是数据集的示例。它包含用户的 ID、用户浏览过的产品、访问者是否转化以及转化价值（如果已转化）。

user_id	product_view_0	product_view_1	product_view_2	product_view_3	product_view_4	has_conversion	conversion_value
0	牛仔裤	t_shirt	t_shirt	无	无	false	0.0
1	牛仔裤	t_shirt	牛仔裤	套头衫	无	false	0.0
2	t_shirt	套头衫	t_shirt	t_shirt	无	true	105.19
3	t_shirt	t_shirt	牛仔裤	无	无	false	0.0
4	t_shirt	袜子	牛仔裤	牛仔裤	无	false	0.0

您感兴趣的指标如下：

• view_counts：您网站的访问者最先看到每件商品的次数。
• total_conversion_value：访问者在转化时花费的总金额。
• conversion_rate：访问者的转化率。

2. 以公开方式生成指标：

conversion_metrics = df.groupby(['product_view_0'
                               ])[['conversion_value', 'has_conversion']].agg({
                                   'conversion_value': [len, np.sum],
                                   'has_conversion': np.mean
                               })
conversion_metrics = conversion_metrics.rename(
   columns={
       'len': 'view_counts',
       'sum': 'total_conversion_value',
       'mean': 'conversion_rate'
   }).droplevel(
       0, axis=1)

如前所述，这些统计信息可以揭示数据集中个人的相关信息。例如，在第一次看到套头衫后，只有一个人完成转化。22 次观看时，转化率约为 0.05。现在，您需要将每个条形图转换为不公开的条形图。

3. 使用 pipeline_dp.NaiveBudgetAccountant 类定义您的隐私参数，然后指定要用于分析的 epsilon 和 delta 参数。

如何设置这些参数取决于您的具体问题。如需详细了解它们，请参阅“可选：调整差分隐私参数”。

以下代码段使用示例值：

budget_accountant = pipeline_dp.NaiveBudgetAccountant(
     total_epsilon=1, total_delta=1e-5)

4. 初始化 LocalBackend 实例：

ops = pipeline_dp.LocalBackend()

您可以使用 LocalBackend 实例，因为您在本地运行此程序，没有其他框架（如 Beam 或 Spark）。

5. 初始化 DPEngine 实例：

dp_engine = pipeline_dp.DPEngine(budget_accountant, ops)

借助 PipelineDP，您可以通过 pipeline_dp.AggregateParams 类指定更多参数，这会影响私有统计信息的生成。

params = pipeline_dp.AggregateParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     metrics=[pipeline_dp.Metrics.COUNT],
     max_partitions_contributed=1,
     max_contributions_per_partition=1)

6. 指定要计算 count 指标并使用 LAPLACE 噪声分布。
7. 将 max_partitions_contributed 实参设置为 1 值。

该参数限定用户可以贡献多少不同的访问次数。您希望用户每天访问网站一次，而不关心用户是否在一天中多次访问网站。

8. 将 max_contributions_per_partitions 实参设置为 1 值。

该参数用于指定在这种情况下单个访问者可对单个分区或产品类别做出多少贡献。

9. 创建一个 data_extractor 实例，用于指定 privacy_id、partition 和 value 字段的位置。

您的代码应如以下代码段所示：

def run_pipeline(data, ops):
 budget_accountant = pipeline_dp.NaiveBudgetAccountant(
     total_epsilon=1, total_delta=1e-5)

 dp_engine = pipeline_dp.DPEngine(budget_accountant, ops)

 params = pipeline_dp.AggregateParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     metrics=[pipeline_dp.Metrics.COUNT],
     max_partitions_contributed=1, # A single user can only contribute to one partition.
     max_contributions_per_partition=1, # For a single partition, only one contribution per user is used.
     )

 data_extractors = pipeline_dp.DataExtractors(
     privacy_id_extractor=lambda row: row.user_id,
     partition_extractor=lambda row: row.product_view_0
     value_extractor=lambda row: row.has_conversion)

 dp_result = dp_engine.aggregate(data, params, data_extractors)

 budget_accountant.compute_budgets()

 return dp_result

10. 添加以下代码，将您的 Pandas DataFrame 转换为一个行列表，您可以从中直接计算差分隐私统计信息：

rows = [index_row[1] for index_row in df.iterrows()]
dp_result_local = run_pipeline(rows, ops) # Returns generator
list(dp_result_local)

恭喜！您计算了您的首个差分隐私统计信息！

此图表在您先前计算的非私密计数旁边显示了差分隐私计数结果：

运行代码时获得的条形图可能与此图表不同，这没关系。由于差分隐私中的噪声，您每次运行代码时都会得到不同的条形图，但您可以看到它们与原始的非私密条形图类似。

请注意，切勿为保护隐私而多次运行流水线，确保隐私安全。如需了解详情，请参阅“计算多项统计信息”。

5. 使用公共分区

在上一部分中，您可能已经注意到，您丢弃了某个分区的所有访问数据，即首次在您的网站上看到袜子的访问者。

这是由于分区选择或阈值造成的，当输出分区的存在取决于用户数据本身时，这是确保差分隐私保证的重要步骤。在这种情况下，如果只是输出中存在某个分区，就可能会导致数据中是否存在个别用户。如需详细了解此内容侵犯隐私权的原因，请参阅这篇博文。为了防止这种侵犯隐私权的行为，PipelineDP 只会保留包含足够数量用户的分区。

如果输出分区列表不依赖于用户私人数据，则无需执行此分区选择步骤。您的例子其实就是这种情况，因为您知道客户首先可能看到的所有产品类别。

如需使用分区，请执行以下操作：

1. 创建可能的分区列表：

public_partitions_products = ['jeans', 'jumper', 'socks', 't-shirt']

2. 将列表传递给 run_pipeline() 函数，该函数会将其设置为 pipeline_dp.AggregateParams 类的附加输入：

run_pipeline(
    rows, ops, total_delta=0, public_partitions=public_partitions_products)
  # Returns generator

params = pipeline_dp.AggregateParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     metrics=[pipeline_dp.Metrics.COUNT],
     max_partitions_contributed=1,
     max_contributions_per_partition=1,
     public_partitions=public_partitions_products)

如果您使用公共分区和 LAPLACE 噪声，可以将 total_delta 参数设置为 0 值。

现在，您可以在结果中看到所有分区或产品的数据都得到报告。

公开分区不仅可以让您保留更多分区，而且还会增加大约一半的干扰数据，因为您不会在分区选择上花费任何隐私预算，因此原始计数和隐私计数之间的差异与上一次运行相比略小。

使用公开部分时，请注意以下两点：

• 从原始数据派生分区列表时，请务必小心谨慎。如果不以差分隐私方式执行此操作，您的流水线将不再提供差分隐私保证。如需了解详情，请参阅“高级：从数据派生分区”。
• 如果某些公开分区没有数据，则需要对这些分区应用噪声以保护差分隐私。例如，如果您的数据集或网站中未出现过裤子之类的其他商品，则该商品仍然是噪声数据，结果可能会显示一些产品访问次数，而实际上没有任何产品访问次数。

高级：从数据派生分区

如果您在同一流水线中对同一组非公共输出分区运行多次汇总，则可以使用 dp_engine.select_private_partitions() 方法派生一次分区列表，然后将这些分区作为 public_partitions 输入提供给每次汇总。这不仅从隐私的角度来看是安全的，而且还可以减少添加噪声，因为对于整个流水线，在分区选择上仅使用一次隐私预算。

def get_private_product_views(data, ops):
 """Obtains the list of product_views in a private manner.

   This does not calculate any private metrics; it merely obtains the list of
   product_views but does so while making sure the result is differentially private.
   """

 # Set the total privacy budget.
 budget_accountant = pipeline_dp.NaiveBudgetAccountant(
     total_epsilon=1, total_delta=1e-5)

 # Create a DPEngine instance.
 dp_engine = pipeline_dp.DPEngine(budget_accountant, ops)

 # Specify how to extract privacy_id, partition_key, and value from a   
 # single element.
 data_extractors = pipeline_dp.DataExtractors(
     partition_extractor=lambda row: row.product_view_0,
     privacy_id_extractor=lambda row: row.user_id)

 # Run aggregation.
 dp_result = dp_engine.select_partitions(
     data, pipeline_dp.SelectPrivatePartitionsParams(
       max_partitions_contributed=1),
     data_extractors=data_extractors)

 budget_accountant.compute_budgets()
 return dp_result

6. 计算多项统计信息

现在，您已经了解了 PipelineDP 的工作原理，接下来可以了解如何将它用于一些更高级的用例。正如本课开始时提到的，您对三种统计信息感兴趣。借助 PipelineDP，您可以同时计算多个统计信息，前提是这些统计信息在 AggregateParams 实例中共享相同参数（稍后将对此进行介绍）。一次性计算多个指标不仅更清晰、更轻松，在隐私保护方面也有优势。

还记得为 NaiveBudgetAccountant 类提供的 epsilon 和 delta 参数，它们表示所谓的“隐私预算”，用于衡量从数据中泄露的用户隐私程度。

关于隐私预算的一个重要注意事项是它具有累加性。如果使用特定 epsilon ➘ 和 delta 表格一次运行流水线，则需支出 (过滤 ； ) 预算。如果第二次运行，则总预算为 (2{3}, 2➘)。同样，如果使用 NaiveBudgetAccountant 方法计算多项统计信息，并连续使用 过滤 探索 ，则支出总预算 为 (2 该类, 2 结算 )。这意味着您的隐私保证程度会降低。

为避免出现这种情况，您需要使用单个 NaiveBudgetAccountant 实例，并在需要针对相同数据计算多项统计信息时使用具有总预算。然后，您需要指定要用于每次汇总的 epsilon 和 delta 值。最终，您最终会获得相同的整体隐私保证，但特定汇总的 epsilon 和 delta 值越高，准确性就越高。

如需查看实际操作，您可以计算 count、mean 和 sum 统计信息。

您可以根据以下两种不同指标计算统计信息：conversion_value 指标用于根据最先查看的产品来推断产生的收入金额；has_conversion 指标，用于计算您网站的访问者数量和平均转化率。

对于每个指标，您需要单独指定用于指导计算私密统计信息的参数。您将隐私预算分配给这两个指标。您通过“has_conversion”指标计算了两种统计信息，因此您想要将其三分之二分配给初始预算，并将另一三分之一分配给 conversion_value 指标。

如需计算多个统计信息，请执行以下操作：

1. 使用要用于这三种统计信息的总 epsilon 和 delta 值来设置隐私预算会计：

budget_accountant = pipeline_dp.NaiveBudgetAccountant(
     total_epsilon=1, total_delta=0)

2. 初始化 DPEngine 以计算您的指标：

 dp_engine = pipeline_dp.DPEngine(budget_accountant, ops)

3. 指定此指标的参数。

params_conversion_value_metrics = pipeline_dp.AggregateParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     metrics=[pipeline_dp.Metrics.SUM],
     max_partitions_contributed=1,
     max_contributions_per_partition=1,
     min_value=0,
     max_value=100,
     public_partitions=public_partitions,
     budget_weight=1/3)

最后一个参数可视需要指定隐私预算的权重。您可以为所有对象指定相同的权重，但如前所述，您需要将此参数设置为三分之一。

您还可以设置 min_value 和 max_value 实参，以指定应用于分区中隐私单元贡献的值的下限和上限。如果要计算不公开的总和或平均值，则必须使用这些参数。不要期望值为负值，因此您可以假设 0 和 100 为合理的上下限。

4. 提取相关数据，然后将其传递给聚合函数：

data_extractors_conversion_value_metrics = pipeline_dp.DataExtractors(
     privacy_id_extractor=lambda row: row.user_id,
     partition_extractor=lambda row: row.product_view_0,
     value_extractor=lambda row: row.conversion_value)

 dp_result_conversion_value_metrics = (
     dp_engine.aggregate(data, params_conversion_value_metrics,
         data_extractors_conversion_value_metrics))

5. 按照相同的步骤，根据 has_conversion 变量计算这两个指标：

params_conversion_rate_metrics = pipeline_dp.AggregateParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     metrics=[pipeline_dp.Metrics.COUNT, pipeline_dp.Metrics.MEAN],
     max_partitions_contributed=1,
     max_contributions_per_partition=1,
     min_value=0,
     max_value=1,
     public_partitions=public_partitions,
     budget_weight=2/3)

 data_extractors_conversion_rate_metrics = pipeline_dp.DataExtractors(
     privacy_id_extractor=lambda row: row.user_id,
     partition_extractor=lambda row: row.product_view_0,
     value_extractor=lambda row: row.has_conversion)

 dp_result_conversion_rate_metrics = (
     dp_engine.aggregate(data, params_conversion_rate_metrics,
         data_extractors_conversion_rate_metrics))

唯一的变化是 pipeline_dp.AggregateParams 实例，在此实例中，您现在将 mean 和 count 定义为聚合，并将隐私预算的三分之二分配给此计算。由于您希望为两种统计信息设置相同的贡献边界，并基于同一个 has_conversion 变量进行计算，因此您可以在同一 pipeline_dp.AggregateParams 实例中将它们合并，并同时计算它们。

6. 调用 budget_accountant.compute_budgets() 方法：

budget_accountant.compute_budgets()

您可以绘制所有三种私有统计信息与原始统计信息的对比情况。根据所添加的干扰数据，您会发现结果实际上可能会超出合理范围。在此例中，由于添加的噪声在 0 左右为对称，因此您看到连衣裤的转化率和总转化价值都为负。为了进行进一步的分析和处理，最好不要手动对私人统计信息进行后处理，但如果您想将这些图表添加到报告中，只需事后将最小值设为 0 即可，而不会违反隐私权保证。

7. 使用 Beam 运行流水线

如今的数据处理要求您处理大量数据，以至于无法在本地处理这些数据。相反，许多人使用框架进行大规模数据处理（例如 Beam 或 Spark），并在云端运行他们的流水线。

只需对代码进行细微更改，PipelineDP 即可支持 Beam 和 Spark。

如需使用 private_beam API 通过 Beam 运行流水线，请执行以下操作：

1. 初始化 runner 变量，然后创建一个流水线，在其中将隐私操作应用于 rows 的 Beam 表示形式：

runner = fn_api_runner.FnApiRunner()  # local runner

with beam.Pipeline(runner=runner) as pipeline:
   beam_data = pipeline | beam.Create(rows)

2. 使用必需的隐私权参数创建 budget_accountant 变量：

budget_accountant = pipeline_dp.NaiveBudgetAccountant(
               total_epsilon=1, total_delta=0)

3. 创建一个 pcol（专用集合）变量，以确保任何汇总作业都满足您的隐私权要求：

pcol = beam_data | pbeam.MakePrivate(
                                 budget_accountant=budget_accountant,
                                 privacy_id_extractor=lambda 
                                                    row: row.user_id)

4. 在适当的类中指定专用聚合的参数。

在这里，由于您计算商品浏览次数的总和，因此使用了 pipeline_dp.aggregate_params.SumParams() 类。

5. 将聚合参数传递给 pbeam.Sum 方法以计算统计信息：

dp_result = pcol | pbeam.Sum(params)

6. 最后，您的代码应如以下代码段所示：

import pipeline_dp.private_beam as pbeam
runner = fn_api_runner.FnApiRunner()  # local runner

with beam.Pipeline(runner=runner) as pipeline:
   beam_data = pipeline | beam.Create(rows)
   budget_accountant = pipeline_dp.NaiveBudgetAccountant(
               total_epsilon=1, total_delta=0)

   # Create private collection.
   pcol = beam_data | pbeam.MakePrivate(
                              budget_accountant=budget_accountant,
                              privacy_id_extractor=lambda row:  
                                                         row.user_id)
   # Specify parameters.
   params = pipeline_dp.aggregate_params.SumParams(
     noise_kind=pipeline_dp.NoiseKind.LAPLACE,
     max_partitions_contributed=1,
     max_contributions_per_partition=1,
     min_value=0,
     max_value=100,
     public_partitions=public_partitions_product_views,
     partition_extractor=lambda row: row.product_view_0,
     value_extractor=lambda row:row.conversion_value)
   dp_result = pcol | pbeam.Sum(params)

   budget_accountant.compute_budgets()
   dp_result | beam.Map(print)

8. 可选：调整隐私设置和实用程序参数

您已经了解了此 Codelab 中提到的几个参数，例如 epsilon、delta 和 max_partitions_contributed 参数。您可以大致将其分为两类：隐私权参数和实用程序参数。

与隐私保护有关的参数

epsilon 和 delta 参数可量化您通过差分隐私提供的隐私。更确切地说，它们可以衡量潜在攻击者可以从匿名化输出中获得的有关数据的信息量。参数值越高，攻击者获得的数据信息就越多，这会带来隐私风险。另一方面，epsilon 和 delta 参数的值越低，为使输出变为匿名状态而需要添加的噪声就越多，每个分区中所需的唯一身份用户数量也就越多。在这种情况下，您需要在实用性和隐私性之间进行权衡。

在 PipelineDP 中，当您在 NaiveBudgetAccountant 实例中设置总隐私预算时，您需要为匿名化输出指定所需的隐私保证。需要注意的是，如果您希望自己的隐私权保证得到保障，则需要谨慎地对每次汇总使用单独的 NaiveBudgetAccountant 实例，或者多次运行流水线，以避免过度使用预算。

如需详细了解差分隐私以及隐私参数的含义，请参阅差分隐私阅读清单。

实用程序参数

实用参数不会影响隐私保证，但会影响输出的准确性，进而影响输出的效用。这些值在 AggregateParams 实例中提供，用于调整添加的噪声。

max_partitions_contributed 参数是 AggregateParams 实例中提供的且适用于所有汇总的实用程序参数。分区对应于 PipelineDP 汇总操作所返回数据的键，因此 max_partitions_contributed 参数会限制用户可以对输出贡献的不同键值对的数量。如果用户贡献的键数超出了 max_partitions_contributed 参数的值，系统会舍弃部分贡献，使其计入 max_partitions_contributed 形参的确切值。

同样，大多数聚合都有一个 max_contributions_per_partition 参数。AggregateParams 实例中也提供了这些变量，每个汇总的这些变量可以有不同的值。它们绑定了用户针对每个键的贡献。

添加到输出中的噪声根据 max_partitions_contributed 和 max_contributions_per_partition 参数进行调节，因此需要做出权衡：为每个参数分配的值越大，意味着保留的数据越多，但结果的噪声也越大。

某些汇总需要 min_value 和 max_value 参数，用于指定每个用户所能贡献内容的边界。如果用户贡献的值小于分配给 min_value 形参的值，则该值会增大为形参的值。同样，如果用户贡献的值大于 max_value 参数的值，则该值会递减为参数值。若要保留更多原始值，您必须指定更大的上下限。噪声会按边界的大小进行缩放，因此边界越大，您可以保留更多数据，但最终得到的结果会更加嘈杂。

最后，noise_kind 参数在 PipelineDP 中支持两种不同的噪声机制：GAUSSIAN 和 LAPLACE 噪声。LAPLACE 分布以较低的贡献边界改善效用，这就是 PipelineDP 默认使用它的原因。不过，如果要使用 GAUSSIAN 分布噪声，可以在 AggregateParams 实例中指定。

9. 恭喜

太棒了！您学完了 PipelineDP Codelab，并学习了很多有关差分隐私和 PipelineDP 的知识。

了解详情

• GitHub 代码库
• PipelineDP
• Privacy on Beam Codelab