フェデレーティッドクエリの記述 - Amazon Athena
単一データソースのクエリ複数のデータソースのクエリ

フェデレーティッドクエリの記述

1 つ、または複数のデータコネクタを設定してアカウントにデプロイしたら、それらを Athena クエリで使用できます。

単一データソースのクエリ

このセクションの例では、Athena CloudWatch コネクタが設定され、アカウントにデプロイされていることを前提としています。他のコネクタを使用する場合も、同じ方法でクエリを実行します。

CloudWatch コネクタを使用する Athena クエリを作成する

  1. https://console.aws.amazon.com/athena/ で Athena コンソールを開きます。

  2. Athena クエリエディタで、FROM 句に次の構文を使用する SQL クエリを作成します。

    MyCloudwatchCatalog.database_name.table_name

例

以下の例では、/var/ecommerce-engine/order-processor CloudWatch Logs ロググループの all_log_streams ビューへの接続に Athena CloudWatch コネクタを使用します。all_log_streams ビューは、ロググループ内のすべてのログストリームのビューです。このクエリ例では、返される行数を 100 に制限します。

例 
SELECT * FROM "MyCloudwatchCatalog"."/var/ecommerce-engine/order-processor".all_log_streams limit 100;

次の例では、前の例と同じビューの情報を解析します。この例では、注文 ID とログレベルを抽出し、レベル INFO のメッセージをフィルタリングします。

例 
SELECT log_stream as ec2_instance,
                        Regexp_extract(message '.*orderId=(\d+) .*', 1) AS orderId,
                        message                                         AS
                        order_processor_log,
                        Regexp_extract(message, '(.*):.*', 1)           AS log_level
            FROM
                    "MyCloudwatchCatalog"."/var/ecommerce-engine/order-processor".all_log_streams
            WHERE Regexp_extract(message, '(.*):.*', 1) != 'INFO'

以下の画像は、サンプル結果を示しています。

注記

この例は、データソースがカタログとして Athena に登録されているクエリを示しています。また、lambda:MyLambdaFunctionName. 形式を使用して、データソースコネクタ Lambda 関数を参照することもできます。

Athena クエリ出力の例。

複数のデータソースのクエリ

さらに複雑な例として、次の図に示すようなアプリケーションインフラストラクチャを持つ e コマース会社があるとします。

さまざまなデータソースを使用した e コマースインフラストラクチャの例。

以下は、図内の番号付き項目についての説明です。

  1. トランザクションレコードが Amazon EMR の HBase に保存されている、セキュア VPC での支払い処理

  2. アクティブな注文を保存して、処理エンジンがそれらに迅速にアクセスできるようにする Redis

  3. E メールアドレスや配送先住所などの顧客アカウントデータ用の Amazon DocumentDB

  4. Fargate のオートスケーリングを使用する e コマースサイトのための Amazon Aurora 内の製品カタログ

  5. 注文プロセッサのログイベントを格納する CloudWatch Logs

  6. Amazon RDS 上の Write-Once-Read-Many データウェアハウス

  7. 配送追跡データを保存するための DynamoDB

この e コマースアプリケーションのデータアナリストが、一部の注文の状態が誤って報告されていることを発見したとします。配送されたにもかかわらず保留中と表示されている注文がある一方、配送済みと表示されていても出荷されていない注文があります。

アナリストは、遅延している注文の数と、e コマースインフラストラクチャ全体で影響を受けた注文が共通している内容を知りたいと考えています。情報のソースを個別に調査する代わりに、アナリストはデータソースをフェデレーションして、必要な情報を 1 つのクエリで取得します。1 つの場所にデータを抽出する必要はありません。

アナリストのクエリは、以下の Athena データコネクタを使用します。

  • CloudWatch Logs – 注文処理サービスからのログを取得し、regex マッチングと抽出を使用して WARN または ERROR イベントがある注文をフィルタリングします。

  • Redis – Redis インスタンスからアクティブな注文を取得します。

  • CMDB – 注文処理サービスを実行して WARN または ERROR メッセージをログに記録した Amazon EC2 インスタンスの ID と状態を取得します。

  • DocumentDB – 影響を受けた注文について、Amazon DocumentDB から顧客の E メールとアドレスを取得します。

  • DynamoDB – 配送テーブルから配送ステータスと追跡詳細を取得し、報告されたステータスと実際のステータス間における潜在的な相違を特定します。

  • HBase – 支払い処理サービスから、影響を受けた注文に関する支払いステータスを取得します。

例
注記

この例は、データソースがカタログとして Athena に登録されているクエリを示しています。また、lambda:MyLambdaFunctionName. 形式を使用して、データソースコネクタ Lambda 関数を参照することもできます。

--Sample query using multiple Athena data connectors.
WITH logs 
     AS (SELECT log_stream, 
                message                                          AS 
                order_processor_log, 
                Regexp_extract(message, '.*orderId=(\d+) .*', 1) AS orderId, 
                Regexp_extract(message, '(.*):.*', 1)            AS log_level 
         FROM 
     "MyCloudwatchCatalog"."/var/ecommerce-engine/order-processor".all_log_streams 
         WHERE  Regexp_extract(message, '(.*):.*', 1) != 'INFO'), 
     active_orders 
     AS (SELECT * 
         FROM   redis.redis_db.redis_customer_orders), 
     order_processors 
     AS (SELECT instanceid, 
                publicipaddress, 
                state.NAME 
         FROM   awscmdb.ec2.ec2_instances), 
     customer 
     AS (SELECT id, 
                email 
         FROM   docdb.customers.customer_info), 
     addresses 
     AS (SELECT id, 
                is_residential, 
                address.street AS street 
         FROM   docdb.customers.customer_addresses),
     shipments 
     AS ( SELECT order_id, 
                 shipment_id, 
                 from_unixtime(cast(shipped_date as double)) as shipment_time,
                 carrier
        FROM lambda_ddb.default.order_shipments),
     payments
     AS ( SELECT "summary:order_id", 
                 "summary:status", 
                 "summary:cc_id", 
                 "details:network" 
        FROM "hbase".hbase_payments.transactions)
         
SELECT _key_            AS redis_order_id, 
       customer_id, 
       customer.email   AS cust_email, 
       "summary:cc_id"  AS credit_card,
       "details:network" AS CC_type,
       "summary:status" AS payment_status,
       status           AS redis_status, 
       addresses.street AS street_address, 
       shipments.shipment_time as shipment_time,
       shipments.carrier as shipment_carrier,
       publicipaddress  AS ec2_order_processor, 
       NAME             AS ec2_state, 
       log_level, 
       order_processor_log 
FROM   active_orders 
       LEFT JOIN logs 
              ON logs.orderid = active_orders._key_ 
       LEFT JOIN order_processors 
              ON logs.log_stream = order_processors.instanceid 
       LEFT JOIN customer 
              ON customer.id = customer_id 
       LEFT JOIN addresses 
              ON addresses.id = address_id 
       LEFT JOIN shipments
              ON shipments.order_id = active_orders._key_
       LEFT JOIN payments
              ON payments."summary:order_id" = active_orders._key_

以下の画像は、クエリの結果の例を示しています。

Athena での横串検索からのサンプル結果。