Amazon Athena/PrestoのLambda式・Lambda関数を触ってみた

サーバーレスのインタラクティブなクエリーサービスAmazon Athenaはバックエンドのクエリーエンジンに分散クエリーエンジンのPrestoを採用しています。 この度、Athena の利用する Presto のバージョンが 0.172 にアップデートされ、Lambda式・Lambda関数に対応したため、Prestoのオフィシャルドキュメントを頼りに実際に動かしてみました。

なお、最新の EMR である 5.10.0 の Presto のバージョンは 0.187 です。

Lambda 式について

Presto 公式ドキュメントからの転載となりますが、以下のように、ほとんどの SQL 式を使えます。

x -> x + 1
(x, y) -> x + y
x -> regexp_like(x, 'a+')
x -> x[1] / x[2]
x -> IF(x > 0, x, -x)
x -> COALESCE(x, 0)
x -> CAST(x AS JSON)

一方で、以下には対応していません。

• サブクエリー x -> 2 + (SELECT 3)
• 集約関数 x -> max(y)
• TRY 関数

なお、最新の Presto 0.192 は TRY 関数に対応しています

x -> x + TRY(1 / 0)

Lambda 関数を使ってみる

Lambda 式を関数として渡してみましょう。

Presto 公式ドキュメントにLambda関数の定義とその実行例があるため、各関数に対して例をひとつだけ選び、動かしてみます。

また、関数の理解を深めるため、Presto の Lambda 関数を Python で書き換えます。

filter(array<T>, function<T, boolean>) → ARRAY<T>

array 型を引数に受け取り、 function が true となるエレメントだけの array 型を返します。

SELECT filter(ARRAY [5, -6, NULL, 7], x -> x > 0)
-- [5, 7]

Python 版

l = [5, -6, None, 7]
filter(lambda x : x > 0, l) # 例1
[x for x in l if x > 0] # 例2

map_filter(map<K, V>, function<K, V, boolean>) → MAP<K,V>

map 型を引数に受け取り、 function が true となるエレメントだけの map 型を返します。

SELECT map_filter(MAP(ARRAY[10, 20, 30], ARRAY['a', NULL, 'c']), (k, v) -> v IS NOT NULL)
-- {10=a, 30=c}

Python 版

d = dict(zip([10,20,30], ['a',None,'c']))
dict(filter(lambda (k,v):v is not None, d.iteritems())) # 例1
{k:v for k,v in d.iteritems() if v is not None} # 例2

reduce(array<T>, initialState S, inputFunction<S, T, S>, outputFunction<S, R>) → R

array 型を引数に受け取り、reduce した結果を返します。

SELECT reduce(ARRAY [5, 20, 50], 0, (s, x) -> s + x, s -> s)
-- 75

Python 版

reduce(lambda x, y: x+y, [5, 20, 50]) # 例1
((5+20)+50) # 例2

loop を使うなら

result = 0
list = [5, 20, 50]
for num in list:
    result = result + num

Python3 では "put in functools, a loop is more readable most of the times" という理由により reduce は組み込み関数から削除され、functools モジュールの1関数となっています。

transform(array<T>, function<T, U>) → ARRAY<U>

array 型を引数に受け取り、 各エレメントに function を実行した array 型を返します。

SELECT transform(ARRAY [5, 6], x -> x + 1)
-- [6, 7]

Python 版

map(lambda x:x+1, [5, 6]) # 例1
[x+1 for x in [5, 6]]     # 例2

transform_keys(map<K1, V>, function<K1, V, K2>) → MAP<K2,V>

map 型を引数に受け取り、 各エントリーに function を実行した結果でキーを更新した map 型を返します。

SELECT transform_keys(MAP(ARRAY [1, 2, 3], ARRAY ['a', 'b', 'c']), (k, v) -> k + 1)
-- {2=a, 3=b, 4=c}

Python 版

d = dict(zip([1,2,3], ['a','b','c']))
dict(map(lambda (k,v): (k+1,v), d.iteritems()))  # 例1
{k+1:v for k,v in d.iteritems()}                 # 例2

transform_values(map<K, V1>, function<K, V1, V2>) → MAP<K, V2>

map 型を引数に受け取り、 各エントリーに function を実行した結果でバリューを更新した map 型を返します。

SELECT transform_values(MAP(ARRAY [1, 2, 3], ARRAY [10, 20, 30]), (k, v) -> v + 1)
-- {1 -> 11, 2 -> 22, 3 -> 33}

Python 版

d = dict(zip([1,2,3], [10,20,30]))
dict(map(lambda (k,v): (k,v+1), d.iteritems())) # 例1
{k:v+1 for k,v in d.iteritems()}                # 例2

zip_with(array<T>, array<U>, function<T, U, R>) → array<R>

2組の Array 型を引数に受け取り、両 Array のエレメントが引数の function を実行した結果の Array 型を返します

SELECT zip_with(ARRAY[1, 2], ARRAY[3, 4], (x, y) -> x + y)
-- [4, 6]

Python 版

map(lambda (x,y):x+y, zip([1,2], [3,4])) # 例1
[x+y for x,y in zip([1,2], [3,4])]       # 例2

最後に

今回のLambda式・関数を利用すると ELB のアクセスログに対して Athena を使ってクエリー部分をゴニョゴニョすることも瞬殺です。

SELECT transform(split(url_extract_query('http://example.com/?a=1&b=2&c=3'), '&'), x -> split(x, '='))
-- [[a, 1], [b, 2], [c, 3]]

Amazon Athena の適用範囲が一段と広がりました。

参照

• Amazon Athena Engine Upgrades, Support for Lambda Expressions, and Ability to Skip Headers
• 6.19. Lambda Expressions and Functions - Presto 0.172 Documentation