S3のデータをAmazon Athenaを使って分析する

Amazon Athenaは対話型クエリサービスで、標準的なSQLを使ってAmazon S3の直接データを直接分析することを簡単にしてくれます。Athenaはサーバレスなので、インフラを構築したり管理する必要はなく、今すぐにデータ分析を始めることができます。Athenaはデータをロードしたり、複雑なETL処理をする必要すらありません。S3に保存されているデータに直接クエリすることができます。

Athenaは、クエリを実行する際に分散SQLエンジンのPrestoを利用しています。また、テーブルを作成、削除、変更、パーティションするためにApache Hiveも利用しています。Hive互換のDDL文や、ANSI SQL文をAthenaクエリエディタ内で書くことができます。複雑なJOINやウィンドウ関数、そして複雑なデータ型をAthenaで使うこともできます。Athenaはschema-on-readとして知られるアプローチを取っていて、クエリを実行する時にデータに対してスキーマを定義することができます。これによって、データロードやETLを必要としていません。

Athenaはクエリ毎にスキャンしたデータの量に応じて課金します。データをパーティションしたり、圧縮したり、またはApache Parquet等の列指向フォーマットに変換することでコストを抑えパフォーマンスを向上させることができます。詳しくはAthenaの料金ページをご覧ください。

この記事では、既に決められた形式のテキストファイルで生成されるElastic Load Balancingのログに対して、どのようにAthenaを使うかをお見せします。テーブルを作成し、Athenaで使われる形式でデータをパーティションして、それをParquetに変換してから、クエリのパフォーマンスを比較してみます。

今回の例では、生ログはAmazon S3に以下の形式で保存されています。年、月、日と分かれたプレフィックスがあって、トータルで2570個のオブジェクトが1 TBのデータを持っています。

テーブルを作成する

もしApache Hiveを使ったことがあれば、Athenaでのテーブル作成は見慣れたものかと思います。クエリエディタでDDL文を書くか、ウィザードを使うか、もしくはJDBCドライバを使ってテーブルを作成することができます。以下のDDL文をAthenaクエリエディタにコピー＆ペーストしてテーブルを作成してみます。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_raw_native (
  request_timestamp string, 
  elb_name string, 
  request_ip string, 
  request_port int, 
  backend_ip string, 
  backend_port int, 
  request_processing_time double, 
  backend_processing_time double, 
  client_response_time double, 
  elb_response_code string, 
  backend_response_code string, 
  received_bytes bigint, 
  sent_bytes bigint, 
  request_verb string, 
  url string, 
  protocol string, 
  user_agent string, 
  ssl_cipher string, 
  ssl_protocol string ) 
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
         'serialization.format' = '1','input.regex' = '([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*):([0-9]*) ([.0-9]*) ([.0-9]*) ([.0-9]*) (-|[0-9]*) (-|[0-9]*) ([-0-9]*) ([-0-9]*) \\\"([^ ]*) ([^ ]*) (- |[^ ]*)\\\" (\"[^\"]*\") ([A-Z0-9-]+) ([A-Za-z0-9.-]*)$' ) 
LOCATION 's3://athena-examples/elb/raw/';

CREATE TABLE文で指定されている正規表現に注目して下さい。任意の正規表現を使って、Athenaにテキスト内の各行をどのように解釈するかを指定することができます。また、JSON等の他の形式にもAthenaを使ってクエリすることができます。CSV、TSV、またはJSON形式を処理する場合には、正規表現は必要ありません。成功すると、テーブルとそのスキーマがデータカタログ(左側)に現れます。Athenaは内部にデータカタログを持っていて、テーブル、データベース、パーティションに関する情報を保存するために使われています。これはとても耐久性が高く、管理を必要としません。DDLクエリかコンソールを通じてこのカタログを操作することができます。

これでAmazon S3上に保存されたデータに対してテーブルを作成したので、もうデータをクエリする準備ができました。このelb_logs_raw_nativeテーブルがs3://athena-examples/elb/raw/というプレフィックスを向いていることに注目して下さい。これによって、このプレフィックス以下に例えば新しい月のデータなどをさらに追加すると、テーブルは自動的に成長します。簡単なクエリを実行してみましょう:

SELECT * FROM elb_logs_raw_native WHERE elb_response_code = '200' LIMIT 100;

これだけで、インフラの構築もETLもなしに、すべてのログに対してクエリすることができるようになったのです。

データをパーティションする

お客様はしばしばデータを時系列形式で保存していて、1日、1ヶ月、1年の特定のデータにだけクエリする必要があります。パーティションを行わないと、Athenaはクエリを実行する時にテーブル全体をスキャンします。パーティションをすることで、Athenaを特定のパーティションに制限することができるので、スキャンするデータ量を減らし、コストを抑え、そしてパフォーマンスを向上させることができます。

AthenaはApache Hive方式のデータパーティションを使います。皆さんはデータを複数の軸、例えば月、週、日、時間、または顧客ID、もしくはそれらすべてを使って、パーティションすることができます。

パーティションを使うためには、まずスキーマ定義をパーティションを含む様に変更して、その後パーティションのメタデータをAthenaにロードする必要があります。同じCREATE TABLE文を使いますが、パーティションを有効化してみましょう。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_raw_native_part (
  request_timestamp string, 
  elb_name string, 
  request_ip string, 
  request_port int, 
  backend_ip string, 
  backend_port int, 
  request_processing_time double, 
  backend_processing_time double, 
  client_response_time double, 
  elb_response_code string, 
  backend_response_code string, 
  received_bytes bigint, 
  sent_bytes bigint, 
  request_verb string, 
  url string, 
  protocol string, 
  user_agent string, 
  ssl_cipher string, 
  ssl_protocol string ) 
PARTITIONED BY(year string, month string, day string) 
ROW FORMAT SERDE 'org.apache.hadoop.hive.serde2.RegexSerDe'
WITH SERDEPROPERTIES (
         'serialization.format' = '1','input.regex' = '([^ ]*) ([^ ]*) ([^ ]*):([0-9]*) ([^ ]*):([0-9]*) ([.0-9]*) ([.0-9]*) ([.0-9]*) (-|[0-9]*) (-|[0-9]*) ([-0-9]*) ([-0-9]*) \\\"([^ ]*) ([^ ]*) (- |[^ ]*)\\\" (\"[^\"]*\") ([A-Z0-9-]+) ([A-Za-z0-9.-]*)$' ) 
LOCATION 's3://athena-examples/elb/raw/';

CREATE TABLE文のPARTITIONED BY句に注目してください。データは年、月、日でパーティションされています。Resultsセクションでは、パーティションされたテーブルではパーティションをロードする様にリマインドしています。

ALTER TABLE ADD PARTITION文を使うと、パーティションに関連するメタデータを読み込むことができます。例えばs3://athena-examples/elb/raw/2015/01/01/からデータをロードするには、以下のクエリを実行します:

ALTER TABLE elb_logs_raw_native_part ADD PARTITION (year='2015',month='01',day='01') location 's3://athena-examples/elb/raw/2015/01/01/'
show partitions elb_logs_raw_native_part

これでWHERE句でパーティションを指定することで、各クエリを制限することができ、データのスキャン量は少なくなり早く終わります。例えばこちら:

SELECT distinct(elb_response_code),
         count(url)
FROM elb_logs_raw_native_part
WHERE year='2015'
        AND month= '01'
        AND day='01'
GROUP BY  elb_response_code

もし多数のパーティションを持っている場合には、手動で指定するのは効率が悪くなります。JDBCドライバを使ってこの処理を自動がすることができます。もしデータが既にHiveパーティション形式になっていれば、これをする必要はありません。

データを列指向形式に変換する

AthenaではApache ParquetやApache ORC等のオープンソースの列指向形式を使うことができます。データを列指向形式に変換することで、クエリのパフォーマンスを向上させるだけでなく、コストも抑えることができます。

列指向形式へのデータ変換方法はいくつかあります。この記事では、Amazon EMR上でたった20行程度のPySparkスクリプトを動かして、データをApache Parquetに変換してみます。スクリプトは同時にデータを年、月、日でパーティションします。この記事を公開した時には、US-eastにある2ノードのr3.8xlargeクラスタで1 TBのログファイルを130 GBの圧縮されたApache Parquetファイル(87%の圧縮率)に$5で変換することができました。

こちらが、Amazon S3上のファイルの配置になります:

ファイルの配置に注目してください。この形式のパーティションは、key=value形式で指定されているので、Athenaに自動的にパーティションとして認識されます。これによって、msck repair table <tablename>コマンドを使ってすべてのパーティションを自動的にロードすることができます。これはHiveがパーティションされたデータを認識する方法と似ています。もしデータが上記のkey-value形式ではない場合、以前お話した様に手動でパーティションをロードして下さい。

テーブルをParquetのデータ上に作成します。スキーマは同一のままで、Snappyを使ってファイルを圧縮しています。

CREATE EXTERNAL TABLE IF NOT EXISTS elb_logs_pq (
  request_timestamp string,
  elb_name string,
  request_ip string,
  request_port int,
  backend_ip string,
  backend_port int,
  request_processing_time double,
  backend_processing_time double,
  client_response_time double,
  elb_response_code string,
  backend_response_code string,
  received_bytes bigint,
  sent_bytes bigint,
  request_verb string,
  url string,
  protocol string,
  user_agent string,
  ssl_cipher string,
  ssl_protocol string )
PARTITIONED BY(year int, month int, day int) 
STORED AS PARQUET
LOCATION 's3://athena-examples/elb/parquet/'
tblproperties ("parquet.compress"="SNAPPY");

カタログにすべてのパーティションを認識させるために、msck repair table elb_logs_pqを実行します。クエリが完了すると、すべてのパーティションを表示することができます。

msck repair table elb_logs_pq
show partitions elb_logs_pq

パフォーマンスを比較する

同じクエリのパフォーマンスをテキストファイルとParquetファイルの間で比較することができます。

SELECT elb_name,
       uptime,
       downtime,
       cast(downtime as DOUBLE)/cast(uptime as DOUBLE) uptime_downtime_ratio
FROM 
    (SELECT elb_name,
        sum(case elb_response_code
        WHEN '200' THEN
        1
        ELSE 0 end) AS uptime, sum(case elb_response_code
        WHEN '404' THEN
        1
        ELSE 0 end) AS downtime
    FROM elb_logs_pq
    GROUP BY  elb_name)

圧縮・パーティション・列指向データへのクエリ

生テキストファイルへのクエリ

SELECT elb_name,
       uptime,
       downtime,
       cast(downtime as DOUBLE)/cast(uptime as DOUBLE) uptime_downtime_ratio
FROM 
    (SELECT elb_name,
        sum(case elb_response_code
        WHEN '200' THEN
        1
        ELSE 0 end) AS uptime, sum(case elb_response_code
        WHEN '404' THEN
        1
        ELSE 0 end) AS downtime
    FROM elb_logs_raw_native
    GROUP BY  elb_name)

Athenaはクエリ毎にデータをスキャンした量で課金します。列指向形式に変換し、圧縮してパーティションすることで、コストを削減できるだけでなくパフォーマンスも向上します。以下の表でデータを列指向形式に変換した際のコストを比較しています。

(*Snappyで圧縮)

まとめ

Amazon Athenaによって、インフラを全く管理する必要なく、S3上のデータを標準SQLで分析することができます。また、JDBCドライバを使って、ビジネスインテリジェンスツールからもAthenaにアクセスすることができます。Athenaはクエリ毎にスキャンしたデータ量で課金します。この記事から明らかな様に、オープンソースの形式にデータを変換することで、コストを抑えるだけでなく、パフォーマンスを改善することもできます。

Amazon AthenaはUS-East (N. Virginia)とUS-West 2 (Oregon)リージョンで利用可能です。詳細は、Amazon Athena製品ページ、またはAmazon Athenaユーザーガイドをご覧ください。

原文: Analyzing Data in S3 using Amazon Athena (翻訳: SA岩永)