透過 PipelineDP 計算私人統計資料

1. 事前準備

您可能會認為匯總統計資料並未洩露任何所屬個人的資訊。不過，攻擊者可以透過多種方式，從匯總統計資料中得知個人的機密資訊。

在本程式碼研究室中，您將瞭解如何透過 PipelineDP 以差異化隱私匯總產生私人統計資料，進而保護個人隱私權。PipelineDP 是一種 Python 架構，可讓您透過批次處理系統 (例如 Apache Spark 和 Apache Beam) 對大型資料集套用差異化隱私。如要進一步瞭解如何在 Go 中計算差異化隱私的統計資料，請參閱 Beam 的隱私權程式碼研究室。

私人代表輸出內容的產生方式不會洩漏資料中個人的任何私人資訊。差異化隱私是差異化隱私的重要概念，可協助您保護使用者隱私。系統會採取跨多位使用者的資料匯總程序，所有去識別化方法都會使用匯總，但並非所有匯總方法都能進行去識別化。另一方面，差異化隱私能提供可量化的資訊，確保資訊外洩和隱私權。

必要條件

• 熟悉 Python
• 熟悉基本資料匯總
• 體驗 pandas、Spark 和 Beam

課程內容

• 差異化隱私基本概念
• 如何透過 PipelineDP 計算差異化隱私的摘要統計資料
• 如何運用其他隱私權和公用程式參數調整結果

軟硬體需求

• 如果想在自己的環境中執行程式碼研究室，您的電腦必須安裝 Python 3.7 以上版本。
• 如果想在沒有自己的環境的情況下按照程式碼研究室的說明操作，您需要存取 Colaboratory 存取權。

2. 瞭解差異化隱私

如要進一步瞭解差異化隱私，請查看這個簡單的範例。

假設你任職於線上時尚零售商的行銷部門，想要瞭解哪些產品最有可能銷售。

這張圖表顯示消費者造訪商店網站時最先瀏覽哪些產品：T 恤、跳衣、襪子或牛仔褲。T 恤是最受歡迎的商品，襪子則是最受歡迎的商品。

這看起來很實用，但還有一個小提示。如想將額外資訊納入考量 (例如客戶是否完成購物或他們瀏覽了哪類產品)，可能會導致資料中出現個別使用者。

這張圖表顯示，只有一位客戶先看跳躍器，然後再實際購買：

從隱私權的角度來看，這並不是件好事。匿名統計資料不應透露個人貢獻，這時你該怎麼做？只要將隨機雜訊加入長條圖，就能降低這些資料的準確度！

這個長條圖未必「完全」正確，但仍然很實用，且不會顯示個別貢獻：

差異化隱私是為個別貢獻內容新增的隨機雜訊量。

這個範例過於簡化。如何實作差異化隱私的做法較為複雜，而且可能有一些非預期的導入細節。與密碼學類似，建立本身的差異化隱私實作方式可能不是一個好主意。請改用 PipelineDP。

3. 下載並安裝 PipelineDP

您可以在本文件中找到所有相關程式碼和圖表，因此不必安裝 PipelineDP 。

如要試用 PipelineDP，您可以自行執行或稍後再使用：

• 下載並安裝 PipelineDP：

pip install pipeline-dp

如要使用 Apache Beam 執行範例，請按照下列步驟操作：

• 下載並安裝 Apache Beam：

pip install apache_beam

您可以在 PipelineDP/examples/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)

正如先前的所學，這些統計資料可以揭露您資料集中個人的資訊。舉例來說，一個人先看到跳躍器後，就只有 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 用於其他進階用途。如一開始所述，您想要查看三項統計資料。PipelineDP 可讓您同時計算多個統計資料，但前提是 AggregateParams 執行個體中的這些參數具有相同的參數，以供稍後查看。如此一來，不僅能更簡潔快速地一次計算多個指標，對隱私權的影響也更為重要。

如果您記得自己提供給 NaiveBudgetAccountant 類別的 epsilon 和 delta 參數，這些項目就代表資料稱為隱私預算，用來評估從資料中洩露的使用者隱私程度。

還有一個重點是，隱私權預算是累加的，如果執行某個管道的 epsilon Lesson 以及 delta 學到一次，您會支出 DISTANCE,學到 的 預算。如果您第二次跑步，所花費的總預算為 (2Chicago, 2 )。同樣地，如果您以 NaiveBudgetAccountant 方法計算多個統計資料，並連續 假設 STAR 的隱私預算，那麼總預算是 2 約期，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()

可以對照原始統計資料，對照這三項私人統計數據。視加入的雜訊而定，結果實際上可能會超出合理範圍。在本例中，跳躍者的轉換率和總轉換價值為負值，因為加入的干擾值約為零。為了進一步分析與處理，最好不要手動對私人統計資料進行後續處理，但如果您想將這些圖表加入報表，可以將下限值設為零即可，不必違反隱私權保證。

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. 選用：調整隱私和公用程式參數

您看到了本程式碼研究室中提到的幾個參數，例如 epsilon、delta 和 max_partitions_contributed 參數。可以大致區分為兩個類別：「隱私參數」和「公用程式參數」。

隱私權參數

epsilon 和 delta 參數會量化您提供的差異化隱私防護機制。更精確地說，這些指標是衡量潛在攻擊者可以從匿名輸出中獲取的資訊量。參數的值越高，攻擊者取得的資料就越多，進而帶來隱私權風險。另一方面，epsilon 和 delta 參數的值越低，您必須在輸出內容中加入的雜訊越多，且每個分區中所需的不重複使用者人數就越多，才能保留在去識別化的輸出內容中。在此情況下，必須兼顧實用性和隱私權。

在 PipelineDP 中，當您在 NaiveBudgetAccountant 執行個體中設定總隱私預算時，必須指定對匿名輸出內容的隱私保證。需注意的是，如要維持隱私保證，請務必針對每項匯總作業小心使用單獨的 NaiveBudgetAccountant 例項，或是多次執行管道，以免過度使用預算。

如要進一步瞭解差異化隱私和隱私權參數的意義，請參閱差異化隱私閱讀清單。

公用程式參數

公用程式參數不會影響隱私權保證，但會影響準確率，因此會影響輸出的實用性。這些樣本會在 AggregateParams 執行個體中提供，並用來調整加入的雜訊。

AggregateParams 執行個體中提供的公用程式參數是 max_partitions_contributed 參數，適用於所有匯總作業。分區與 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 程式碼研究室，並學到許多有關差異化隱私和 PipelineDP 的知識。

瞭解詳情

• GitHub 存放區
• PipelineDP
• Beam 隱私權程式碼研究室