Amazon Comprehend と Amazon Kinesis Data Firehose を使用して、ストリーミング データから機密データをほぼリアルタイムで編集する

データと洞察をほぼリアルタイムで配信することで、企業は顧客のニーズに迅速に対応できます。 リアルタイム データは、ソーシャル メディア、IoT デバイス、インフラストラクチャの監視、コール センターの監視など、さまざまなソースから取得できます。 複数のソースから取り込まれるデータの幅と深さのため、企業は顧客のプライバシーを保護し、機密データがエンド システムからアクセスされないようにするためのソリューションを探しています。 以前は、誤検出や欠落データにフラグを立てる可能性のある個人識別情報 (PII) ルール エンジンに依存するか、ストリーミング データ内の PII を識別するためにカスタム機械学習 (ML) モデルを構築して維持する必要がありました。 また、これらのエンジンやモデルをサポートするために必要なインフラストラクチャを実装して維持する必要もありました。

このプロセスを合理化し、コストを削減するために、次を使用できます。 Amazon Comprehendは、ML を使用して非構造化テキスト内の人、場所、感情、トピックなどの洞察と関係を見つける自然言語処理 (NLP) サービスです。 Amazon Comprehend ML 機能を使用して、顧客の E メール、サポートチケット、製品レビュー、ソーシャルメディアなどの PII を検出して編集できるようになりました。 ML の経験は必要ありません。 たとえば、ドキュメントのインデックスを作成する前に、サポート チケットとナレッジ記事を分析して PII エンティティを検出し、テキストを編集できます。 その後、ドキュメントは PII エンティティから解放され、ユーザーはデータを使用できます。 PII エンティティを秘匿化することで、顧客のプライバシーを保護し、地域の法律や規制に準拠することができます。

この投稿では、Amazon Comprehend をストリーミング アーキテクチャに実装して、PII エンティティをほぼリアルタイムで編集する方法を学びます。 Amazon Kinesis データ ファイアホース 　 AWSラムダ.

この投稿では、Kinesis Data Firehose を使用してストリーミング アーキテクチャに取り込まれた選択フィールドからのデータを編集することに焦点を当てています。そこでは、エンドユーザーまたはダウンストリーム アプリケーションによる消費のために、データの追加の派生コピーを作成、保存、および維持します。 使用している場合 Amazon Kinesisデータストリーム または、PII リダクション以外の追加のユース ケースがある場合は、次を参照してください。 Amazon Kinesis Data Analytics、Amazon Translate、Amazon ComprehendのSQL関数を使用して、ストリーミングデータを翻訳、編集、分析します、使用方法を示します Amazon Kinesis データ分析スタジオ を搭載 アパッチゼッペリン & ApacheFlink ストリーミング データ内のテキスト フィールドをインタラクティブに分析、翻訳、編集します。

ソリューションの概要

次の図は、ストリーミング データの PII リダクションをリアルタイムで実行するためのアーキテクチャの例を示しています。 Amazon シンプル ストレージ サービス （Amazon S3）、 Kinesis DataFirehoseデータ変換, Amazon Comprehend, AWSラムダ. さらに、 AWS SDK for Python（Boto3） Lambda 関数用。 図に示されているように、S3 RAW バケットには編集されていないデータが含まれており、S3 編集済みバケットには Amazon Comprehend を使用した後に編集されたデータが含まれています。 DetectPiiEntities Lambda 関数内の API。

関係する費用

Kinesis Data Firehose、Amazon S3、および Lambda のコストに加えて、このソリューションには Amazon Comprehend からの使用コストが発生します。 支払う金額は、PII を含むレコードの合計数と Lambda 関数によって処理される文字の要素です。 詳細については、次を参照してください。 Amazon Kinesis Data Firehose の料金, Amazon Comprehendの料金, AWSLambdaの価格.

例として、10,000 件のログ レコードがあり、PII を秘匿化するキー値が 500 文字であるとします。 10,000 件のログ レコードのうち、50 件に PII が含まれていると特定されました。 費用の詳細は次のとおりです。

PII コストを含む:

• 各キー値のサイズ = 500 文字 (1 単位 = 100 文字)
• レコードあたりのユニット数 (100 文字) (最小は 3 ユニット) = 5
• 合計ユニット数= 10,000（レコード）x 5（レコードあたりのユニット数）x 1（レコードあたりのAmazon Comprehendリクエスト）= 50,000
• ユニットあたりの価格= $ 0.000002
  • ContainsPiiEntities API を使用して PII でログ レコードを識別するための総コスト = 0.1 USD [50,000 ユニット x 0.000002 USD] 

PII コストの編集:

• PII を含む合計ユニット = 50 (レコード) x 5 (レコードあたりのユニット) x 1 (レコードあたりの Amazon Comprehend リクエスト) = 250
• ユニットあたりの価格= $ 0.0001
  • DetectPiiEntities API を使用して PII の場所を特定するための総コスト = [ユニット数] x [ユニットあたりのコスト] = 250 x $0.0001 = $0.025

識別と編集の総コスト:

• 総コスト: $0.1 (フィールドに PII が含まれている場合の検証) + $0.025 (PII を含むフィールドの編集) = $0.125

AWSCloudFormationを使用してソリューションをデプロイする

この投稿では、 AWS CloudFormation ストリーミング データのリダクション template、繰り返し可能な展開を可能にする実装の完全な詳細を提供します。 デプロイ時に、このテンプレートは 3 つの SXNUMX バケットを作成します。XNUMX つは Amazon Kinesis Data Generator (KDG) から取り込まれた生のサンプル データを保存するためのもので、もう XNUMX つはリダクションされたデータを保存するためのものです。 さらに、Kinesis Data Firehose 配信ストリームを作成します。 DirectPUT 入力として、Amazon Comprehend を呼び出す Lambda 関数 含まれているPiiエンティティ & PiiEntities の検出 PII データを識別して編集するための API。 Lambda 関数は、環境変数へのユーザー入力に依存して、PII について検査する必要があるキー値を決定します。

このソリューションの Lambda 関数では、ペイロード サイズが 100 KB に制限されています。 テキストが 100 KB を超えるペイロードが提供された場合、Lambda 関数はそれをスキップします。

ソリューションを展開するには、次の手順を実行します。

1. 米国東部 (バージニア北部) で CloudFormation スタックを起動する us-east-1:
   
2. スタック名を入力し、他のパラメーターはデフォルトのままにします
3. 選択 AWS CloudFormationがカスタム名でIAMリソースを作成する可能性があることを認めます。
4. 選択する スタックを作成.

リソースを手動でデプロイする

AWS CloudFormation を使用する代わりに手動でアーキテクチャを構築する場合は、このセクションの手順を完了してください。

S3バケットを作成します

次の手順で S3 バケットを作成します。

1. Amazon S3コンソールで、 バケット ナビゲーションペインに表示されます。
2. 選択する バケットを作成する.
3. 生データ用に XNUMX つのバケットを作成し、秘匿化されたデータ用に XNUMX つのバケットを作成します。
4. 作成したバケットの名前をメモします。

Lambda関数を作成する

Lambda 関数を作成してデプロイするには、次の手順を実行します。

1. Lambdaコンソールで、 関数を作成する.
2. 選択する 最初から作成者.
3. 関数名、 入る AmazonComprehendPII-Redact.
4. ランタイム、選択する Pythonの3.9.
5. アーキテクチャ選択 x86_64.
6. 実行の役割選択 Lambda アクセス許可を持つ新しいロールを作成する.
7. 関数を作成したら、次のコードを入力します。

   import json
   import boto3
   import os
   import base64
   import sys
   
   def lambda_handler(event, context):
       
       output = []
       
       for record in event['records']:
           
           # Gathers keys from enviroment variables and makes a list of desired keys to check for PII
           rawkeys = os.environ['keys']
           splitkeys = rawkeys.split(", ")
           print(splitkeys)
           #decode base64
           #Kinesis data is base64 encoded so decode here
           payloadraw=base64.b64decode(record["data"]).decode('utf-8')
           #Loads decoded payload into json
           payloadjsonraw = json.loads(payloadraw)
           
           # Creates Comprehend client
           comprehend_client = boto3.client('comprehend')
           
           
           # This codes handles the logic to check for keys, identify if PII exists, and redact PII if available. 
           for i in payloadjsonraw:
               # checks if the key found in the message matches a redact
               if i in splitkeys:
                   print("Redact key found, checking for PII")
                   payload = str(payloadjsonraw[i])
                   # check if payload size is less than 100KB
                   if sys.getsizeof(payload) < 99999:
                       print('Size is less than 100KB checking if value contains PII')
                       # Runs Comprehend ContainsPiiEntities API call to see if key value contains PII
                       pii_identified = comprehend_client.contains_pii_entities(Text=payload, LanguageCode='en')
                       
                       # If PII is not found, skip over key
                       if (pii_identified['Labels']) == []:
                           print('No PII found')
                       else:
                       # if PII is found, run through redaction logic
                           print('PII found redacting')
                           # Runs Comprehend DetectPiiEntities call to find exact location of PII
                           response = comprehend_client.detect_pii_entities(Text=payload, LanguageCode='en')
                           entities = response['Entities']
                           # creates redacted_payload which will be redacted
                           redacted_payload = payload
                           # runs through a loop that gathers necessary values from Comprehend API response and redacts values
                           for entity in entities:
                               char_offset_begin = entity['BeginOffset']
                               char_offset_end = entity['EndOffset']
                               redacted_payload = redacted_payload[:char_offset_begin] + '*'*(char_offset_end-char_offset_begin) + redacted_payload[char_offset_end:]
                           # replaces original value with redacted value
                           payloadjsonraw[i] = redacted_payload
                           print(str(payloadjsonraw[i]))
                   else:
                       print ('Size is more than 100KB, skipping inspection')
               else:
                   print("Key value not found in redaction list")
           
           redacteddata = json.dumps(payloadjsonraw)
           
           # adds inspected record to record
           output_record = {
               'recordId': record['recordId'],
               'result': 'Ok',
               'data' : base64.b64encode(redacteddata.encode('utf-8'))
           }
           output.append(output_record)
           print(output_record)
           
       print('Successfully processed {} records.'.format(len(event['records'])))
       
       return {'records': output}

8. 選択する 配備します.
9. ナビゲーションペインで、 コンフィギュレーション。
10. MFAデバイスに移動する 環境変数.
11. 選択する 編集.
12. キー、 入る keys.
13. 値、PII をリダクションするキー値をコンマとスペースで区切って入力します。 たとえば、次のように入力します。 Tweet1, Tweet2 この投稿の次のセクションで提供されるサンプル テスト データを使用している場合。
14. 選択する Save.
15. MFAデバイスに移動する 一般的な設定.
16. 選択する 編集.
17. の値を変更する タイムアウト を1分に設定します。
18. 選択する Save.
19. MFAデバイスに移動する 権限.
20. 下のロール名を選択します 実行の役割.
    にリダイレクトされます AWS IDおよびアクセス管理 (IAM) コンソール。
21. 権限を追加、選択する ポリシーを添付.
22. 入力します Comprehend 検索バーに移動し、ポリシーを選択します ComprehendFullAccess.
23. 選択する ポリシーを添付.

Firehose 配信ストリームを作成する

Firehose 配信ストリームを作成するには、次の手順を実行します。

1. Kinesis Data Firehoseコンソールで、 配信ストリームを作成する.
2. ソース選択 ダイレクトプット.
3. 開催場所選択 アマゾンS3.
4. 配信ストリーム名、 入る ComprehendRealTimeBlog.
5. ソース レコードを AWS Lambda で変換する選択 使用可能.
6. AWS Lambda関数、作成した関数の ARN を入力するか、関数を参照します AmazonComprehendPII-Redact.
7. バッファサイズ、値を 1 MB に設定します。
8. バッファ間隔、60 秒のままにします。
9. 宛先設定で、リダクションされたデータ用に作成した S3 バケットを選択します。
10. バックアップ設定で、生レコード用に作成した S3 バケットを選択します。
11. 許可、IAM ロールを作成または更新するか、適切なアクセス許可を持つ既存のロールを選択します。
12. 選択する 配信ストリームを作成する.

Kinesis Data Generator を使用してストリーミング データ ソリューションをデプロイする

Kinesis Data Generator (KDG) を使用してサンプルデータを Kinesis Data Firehose に取り込み、ソリューションをテストできます。 このプロセスを簡素化するために、Lambda 関数と CloudFormation テンプレートを提供して、 アマゾンコグニート ユーザーに割り当て、KDG を使用するための適切な権限を割り当てます。

1. ソフトウェア設定ページで、下図のように Amazon Kinesis Data Generator ページ、選択する CloudFormationを使用してCognitoユーザーを作成するスタックを作成するために、AWS CloudFormation コンソールにリダイレクトされます。
2. KDG へのログインに使用するユーザーのユーザー名とパスワードを指定します。
3. 他の設定はデフォルトのままにして、スタックを作成します。
4. ソフトウェア設定ページで、下図のように 出力 タブで、KDG UI リンクを選択します。
5. ユーザー名とパスワードを入力してログインします。

テスト レコードを送信し、Amazon S3 でリダクションを検証する

ソリューションをテストするには、次の手順を実行します。

1. 前の手順で作成した KDG URL にログインします。
2. AWS CloudFormation スタックがデプロイされたリージョンを選択します。
3. ストリーム/配信ストリーム、作成した配信ストリームを選択します (テンプレートを使用した場合、その形式は accountnumber-awscomprehend-blog).
4. 他の設定はデフォルトのままにします。
5. レコード テンプレートについては、独自のテストを作成するか、次のテンプレートを使用できます。以下に提供されているサンプル データをテストに使用している場合は、 AmazonComprehendPII-Redact ラムダ関数 Tweet1, Tweet2. CloudFormation 経由でデプロイされた場合は、環境変数を次のように更新します。 Tweet1, Tweet2 作成された Lambda 関数内。 サンプル テスト データは次のとおりです。

   {"User":"12345", "Tweet1":" Good morning, everybody. My name is Van Bokhorst Serdar, and today I feel like sharing a whole lot of personal information with you. Let's start with my Email address SerdarvanBokhorst@dayrep.com. My address is 2657 Koontz Lane, Los Angeles, CA. My phone number is 818-828-6231.", "Tweet2": "My Social security number is 548-95-6370. My Bank account number is 940517528812 and routing number 195991012. My credit card number is 5534816011668430, Expiration Date 6/1/2022, my C V V code is 121, and my pin 123456. Well, I think that's it. You know a whole lot about me. And I hope that Amazon comprehend is doing a good job at identifying PII entities so you can redact my personal information away from this streaming record. Let's check"}

6. 選択する データを送る、レコードがストリームに送信されるまで数秒待ちます。
7. 数秒後、KDG ジェネレーターを停止し、配信されたファイルの S3 バケットを確認します。

以下は、生の S3 バケット内の生データの例です。

{"User":"12345", "Tweet1":" Good morning, everybody. My name is Van Bokhorst Serdar, and today I feel like sharing a whole lot of personal information with you. Let's start with my Email address SerdarvanBokhorst@dayrep.com. My address is 2657 Koontz Lane, Los Angeles, CA. My phone number is 818-828-6231.", "Tweet2": "My Social security number is 548-95-6370. My Bank account number is 940517528812 and routing number 195991012. My credit card number is 5534816011668430, Expiration Date 6/1/2022, my C V V code is 121, and my pin 123456. Well, I think that's it. You know a whole lot about me. And I hope that Amazon comprehend is doing a good job at identifying PII entities so you can redact my personal information away from this streaming record. Let's check"}

以下は、リダクションされた S3 バケット内のリダクションされたデータの例です。

{"User":"12345", "Tweet1":"Good morning, everybody. My name is *******************, and today I feel like sharing a whole lot of personal information with you. Let's start with my Email address ****************************. My address is ********************************** My phone number is ************.", "Tweet"2: "My Social security number is ***********. My Bank account number is ************ and routing number *********. My credit card number is ****************, Expiration Date ********, my C V V code is ***, and my pin ******. Well, I think that's it. You know a whole lot about me. And I hope that Amazon comprehend is doing a good job at identifying PII entities so you can redact my personal information away from this streaming record. Let's check"}

秘匿化されたメッセージから機密情報が削除されているため、このデータをエンド システムと共有できるという確信が持てます。

掃除

このソリューションの実験が終了したら、AWS CloudFormation コンソールを使用してリソースをクリーンアップし、この例でデプロイされたすべてのリソースを削除します。 手動の手順に従った場合は、XNUMX つのバケットを手動で削除する必要があります。 AmazonComprehendPII-Redact 機能、 ComprehendRealTimeBlog ストリーム、のログ グループ ComprehendRealTimeBlog ストリーム、および作成されたすべての IAM ロール。

まとめ

この投稿では、PII リダクションを準リアルタイム ストリーミング アーキテクチャに統合し、飛行中にリダクションを実行することでデータ処理時間を短縮する方法を示しました。 このシナリオでは、リダクションされたデータをエンドユーザーに提供し、データ レイク管理者が後で使用できるように raw バケットを保護します。 また、Amazon Comprehend で追加の処理を構築して、トーンやセンチメントを識別し、データ内のエンティティを識別し、各メッセージを分類することもできます。

この投稿の一部として各サービスの個別の手順を提供し、アカウントに必要なリソースをプロビジョニングできる CloudFormation テンプレートも含めました。 このテンプレートは、概念実証またはテスト シナリオにのみ使用してください。 開発者ガイドを参照してください Amazon Comprehend, ラムダ, Kinesisデータ消防ホース サービスの制限について。

PII の識別と編集を開始するには、次を参照してください。 個人を特定できる情報（PII）. この投稿のアーキテクチャ例では、Kinesis Data Firehose データ変換を使用して、Amazon Comprehend API をほぼリアルタイムのデータと統合できます。 Kinesis Data Firehose を使用して準リアルタイムのデータを使用して構築できるものについて詳しくは、以下を参照してください。 Amazon Kinesis Data Firehose 開発者ガイド. このソリューションは、Amazon Comprehend と Kinesis Data Firehose が利用可能なすべての AWS リージョンで利用できます。

著者について

ジョー・モロッティ アマゾン ウェブ サービス (AWS) のソリューション アーキテクトであり、米国中西部のエンタープライズ顧客を支援しています。 彼は幅広い技術的役割を果たし、顧客の可能性の芸術を示すことを楽しんでいます。 余暇には、家族と一緒に新しい場所を探索したり、スポーツ チームのパフォーマンスを分析したりして充実した時間を過ごしています。

スリハーシュアダリ アマゾン ウェブ サービス (AWS) のシニア ソリューション アーキテクトであり、顧客がビジネスの成果からさかのぼって AWS で革新的なソリューションを開発するのを支援しています。 長年にわたり、彼はさまざまな業界のデータ プラットフォームの変革に関して複数の顧客を支援してきました。 彼の主な専門分野には、テクノロジー戦略、データ分析、およびデータ サイエンスが含まれます。 余暇には、テニスをしたり、テレビ番組をどんちゃん騒ぎしたり、タブラを楽しんだりしています。