[Amazon Connect]録音データの検索やS3を参照するQuickSight ダッシュボードを作ってみた

こんにちは、洲崎です。
Amazon Connectは通話の詳細情報を「コンタクト追跡レコード」(以下CTR)に保存します。
CTRは通話録音を再生する際に利用するデータですが、Amazon Connectの仕様上、2年までしか保存されません。
録音自体はS3に保存されますが、2年以上前の通話を聞き起こしたい場合、CTRの情報をみながらS3パスを掘って探す必要があります。
今回はCTRをAmazon ConnectからS3に吐き出し、QuickSight(Athena)から見ることで期間に縛られることなく検索できる仕組みを実装しました。

実装画面

ダッシュボードでは通話開始日時、終了日時、エージェント名、S3のURLを飛ぶボタンを設定しています。
検索の条件は日時範囲指定、エージェント名、キュー名を設定しています。
この項目以外にも、CTRにある情報であれば表示したり検索の条件として設定することができます。(電話番号等)
S3 URLのボタンを押すと該当の通話録音があるページに遷移し、そこから音声データをダウンロードすることができます。(事前にマネジメントコンソールにログインしていることが前提です)

構成図

Amazon ConnectのCTRをKinesis Firehoseを経由してS3に保存します。(間にKinesis Streamsを挟んでも可)
CTRを保存しているS3を対象にGlueクローラーを実施しテーブルを作成し、QuickSight(Athena)で見にいく構成になります。

やってみる

Kinesis Data Firehose

CTRを格納するS3バケットを作成しておきます。
ソースをDirect、送信先をS3でKinesis Firehoseを作成します。
S3は作成したバケットを指定し、S3バケットプレフィックスはcontact-trace-records/、エラープレフィックスはerror/とします。
その他はデフォルトで作成しました。(状況によってバッファサイズやバッファ間隔を調整します)
Kinesis Data Firehoseの設定内容

マネジメントコンソールからAmazon Connectに飛んで、データストリーミングでKinesis Data Firehoseを指定します。

通話テストをして、S3にcontact-trace-records/ができてることを確認します。

Glue クローラー

マネジメントコンソールからAWS Glueに飛び、左のサイドメニューから「Crawlers」をクリックします。

「Create crawler」をクリックして、「Add a date source」をクリックします。

Data sourceをS3、S3 pathに作成したS3を「Browse」をクリックしてcontact-trace-records/を指定します。

IAMロールは新しいのを作成します。(何か既存のもので権限が足りればそれでもOKです)

OutputのTargetは特に指定がなければ"default"を選択します。(Glueのデータベースです)

作成できたら、右上の”Run crawler”をクリックします。

50秒ほどで処理が完了します。ステータスがCompletedになっているか確認します。

Athena

マネジメントコンソールからAthenaに飛び、クエリエディタを開き「テーブルとビュー」に作成したGlueテーブルがあることを確認します。
Glue クローラーで作成したテーブル情報を確認します。

SHOW CREATE TABLE `default`.`contact_trace_records`;

S3バケットのCTRデータを元に、ある程度Glue クローラーがテーブルを作成してくれてることが分かります。

CREATE EXTERNAL TABLE `default.contact_trace_records`(
  `awsaccountid` string COMMENT 'from deserializer', 
  `awscontacttracerecordformatversion` string COMMENT 'from deserializer', 
  `agent` struct<arn:string,aftercontactworkduration:int,aftercontactworkendtimestamp:string,aftercontactworkstarttimestamp:string,agentinteractionduration:int,connectedtoagenttimestamp:string,customerholdduration:int,hierarchygroups:string,longestholdduration:int,numberofholds:int,routingprofile:struct<arn:string,name:string>,statetransitions:string,username:string> COMMENT 'from deserializer', 
  `agentconnectionattempts` int COMMENT 'from deserializer', 
  `answeringmachinedetectionstatus` string COMMENT 'from deserializer', 
  `attributes` string COMMENT 'from deserializer', 
  `campaign` struct<campaignid:string> COMMENT 'from deserializer', 
  `channel` string COMMENT 'from deserializer', 
  `connectedtosystemtimestamp` string COMMENT 'from deserializer', 
  `contactdetails` string COMMENT 'from deserializer', 
  `contactid` string COMMENT 'from deserializer', 
  `customerendpoint` struct<address:string,type:string> COMMENT 'from deserializer', 
  `disconnectreason` string COMMENT 'from deserializer', 
  `disconnecttimestamp` string COMMENT 'from deserializer', 
  `initialcontactid` string COMMENT 'from deserializer', 
  `initiationmethod` string COMMENT 'from deserializer', 
  `initiationtimestamp` string COMMENT 'from deserializer', 
  `instancearn` string COMMENT 'from deserializer', 
  `lastupdatetimestamp` string COMMENT 'from deserializer', 
  `mediastreams` array<struct<type:string>> COMMENT 'from deserializer', 
  `nextcontactid` string COMMENT 'from deserializer', 
  `previouscontactid` string COMMENT 'from deserializer', 
  `queue` struct<arn:string,dequeuetimestamp:string,duration:int,enqueuetimestamp:string,name:string> COMMENT 'from deserializer', 
  `recording` struct<deletionreason:string,location:string,status:string,type:string> COMMENT 'from deserializer', 
  `recordings` array<struct<deletionreason:string,fragmentstartnumber:string,fragmentstopnumber:string,location:string,mediastreamtype:string,participanttype:string,starttimestamp:string,status:string,stoptimestamp:string,storagetype:string>> COMMENT 'from deserializer', 
  `references` array<string> COMMENT 'from deserializer', 
  `scheduledtimestamp` string COMMENT 'from deserializer', 
  `systemendpoint` struct<address:string,type:string> COMMENT 'from deserializer', 
  `tags` string COMMENT 'from deserializer', 
  `transfercompletedtimestamp` string COMMENT 'from deserializer', 
  `transferredtoendpoint` string COMMENT 'from deserializer', 
  `voiceidresult` string COMMENT 'from deserializer')
PARTITIONED BY ( 
  `partition_0` string, 
  `partition_1` string, 
  `partition_2` string, 
  `partition_3` string)
ROW FORMAT SERDE 
  'org.openx.data.jsonserde.JsonSerDe' 
WITH SERDEPROPERTIES ( 
  'paths'='AWSAccountId,AWSContactTraceRecordFormatVersion,Agent,AgentConnectionAttempts,AnsweringMachineDetectionStatus,Attributes,Campaign,Channel,ConnectedToSystemTimestamp,ContactDetails,ContactId,CustomerEndpoint,DisconnectReason,DisconnectTimestamp,InitialContactId,InitiationMethod,InitiationTimestamp,InstanceARN,LastUpdateTimestamp,MediaStreams,NextContactId,PreviousContactId,Queue,Recording,Recordings,References,ScheduledTimestamp,SystemEndpoint,Tags,TransferCompletedTimestamp,TransferredToEndpoint,VoiceIdResult') 
STORED AS INPUTFORMAT 
  'org.apache.hadoop.mapred.TextInputFormat' 
OUTPUTFORMAT 
  'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION
  's3://xxxxxxxxxxx/contact-trace-records/'
TBLPROPERTIES (
  'CrawlerSchemaDeserializerVersion'='1.0', 
  'CrawlerSchemaSerializerVersion'='1.0', 
  'UPDATED_BY_CRAWLER'='ctr20230119', 
  'averageRecordSize'='2572', 
  'classification'='json', 
  'compressionType'='none', 
  'objectCount'='4', 
  'recordCount'='4', 
  'sizeKey'='10291', 
  'typeOfData'='file')

PARTITIONED BY(Partition Projection)は正しく設定できていない為、CREATE EXTERNAL TABLEをVS Code等のエディタに貼り付けて、修正します。
'projection.date.range'や'projection.date.format'は要件に応じて変更します。

/* PARTITIONED BYをdateに置き換え */
CREATE EXTERNAL TABLE `default./*別の名前に置き換える/*`

/* PARTITIONED BYをdateに置き換え */
PARTITIONED BY ( 
  `date` string)

/* TBLPROPERTIESに追記 */
'has_encrypted_data'='false', 
  'projection.enabled'='true',
  'projection.date.type'='date',
  'projection.date.range'='2022/01/01,NOW+9HOURS',
  'projection.date.format'='yyyy/MM/dd',
  'projection.date.interval'='1',
  'projection.date.interval.unit'='DAYS',
  'storage.location.template'='s3://xxxxxxxxxxx/contact-trace-records/${date}',
/* 最後にこれを残す */
'typeOfData'='file')

修正できたら、Athenaのクエリエディタに貼り付けて実行します。
テーブル情報でパーティション等が設定できていることを確認します。

QuickSight

QuickSightのページに飛び、「データセット」を作成します。
データソースは「Athena」を選択します。

作成したGlueテーブルを指定し、「カスタムSQLを使用」をクリックします。

CTRに対してクエリをかけるSQLを入力します。

SELECT 
    ConnectedToSystemTimestamp,
    DisconnectTimestamp,
    Channel,
    ContactId,
    Queue.name,
    Agent.username,
    Recording.location,
    CustomerEndpoint.address AS CustomerEndpoint,
    SystemEndpoint.address AS SystemEndpoint,
    date
FROM "default"."/*作成したテーブル名*/"

クエリを高速化するために、SPICEを有効にします。

作成できたら、データセットを編集します。ConnectedToSystemTimestamp、DisConnectedToSystemTimestamp,date等を文字列からData型に変換します。

UTCではなくJST時間で表示したい場合は、下記記事を参考にAddTime関数でフィールドを作成します。

「計算フィールドを追加」で、下記記事を参考にS3のURLフィールドを作成します。

データセットの作成が完了したら、「分析」で表示したい項目をビジュアルタイプ「テーブル」で出してみます。
フィールドリストから「値」にドラックアンドドロップすることで、シートにデータが表示されます。

シートの鉛筆マークを押して「フィールドのスタイル設定」のURLオプションを変えることで、S3のURLを飛ぶアクションを作成することができます。

フィルターを作成して、3点マークから「シートに追加」することで、フィルター含めたシートを作成することができます。

出したい情報をシートで見ることができたら、右上の「共有」ボタンから「ダッシュボードの公開」で完成です！

最後に

Amazon ConnectのCTRを分析するQuickSightダッシュボードを作ってみました。
ベースはこれで、あとは必要に応じてPartition Projectionの調整や、QuickSightの「パラメーター」や「コントロール」等の機能でカスタマイズしていくとより良いダッシュボードが作れそうです。

ではまた！コンサルティング部の洲崎でした。

参考