Prestoの残りのWINDOW関数について紹介

6記事連続ですが今回もPrestoのWINDOW関数の記事です。例によって僕はトレジャーデータで検証しています。

実は前回のframeの記事でWINDOW関数シリーズは終わりにしようかと思っていたのですが、ドキュメントを見ると、まだ紹介してない関数はあと3つだけになっており、試しに動かしてみたら予想とちょっと違う動きも見られたので切りよくこれらまで紹介することにしました。

ドキュメントはこの記事もここです。
参考: Window Functions

この記事で紹介する関数は次の3つです。
- first_value(x)
- last_value(x)
- nth_value(xoffset)

first_value の説明は、 Returns the first value of the window. とあり、
last_value の説明は、Returns the last value of the window. とどちらも簡潔に書かれています。WINDOW内の最初の値と最後の値を返してくれると読めますね。

そして、 nth_value は offset で指定した数値番目(数値は1以上)の値を返してくれます。

さて、いつもみたいにダミーデータを使ってやってみましょう。
試したコードは下記です。

WITH
    sample_table AS(
        SELECT
            id,
            category,
            value
        FROM
            UNNEST(
                ARRAY[1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                ARRAY['A', 'A', 'A', 'A', 'A', 'A', 'B', 'B', 'B', 'B'],
                ARRAY[20, 10, 20, 30, 5, 50, 15, 30, 20, 40]
            ) AS t(
                id,
                category,
                value
            )
    )
SELECT
    id,
    category,
    value,
    first_value(value) OVER(PARTITION BY category ORDER BY value) AS first_value_,
    last_value(value) OVER(PARTITION BY category ORDER BY value) AS last_value_,
    nth_value(value, 4) OVER(PARTITION BY category ORDER BY value) AS nth_value_4
FROM
    sample_table

さて、この結果はどうなるでしょうか?3関数とも、OVER(PARTITION BY category ORDER BY value) をつけたので、 categoryの値で、’A’ と ‘B’ に分けられ、 value 列の値でソートされた上で、関数が実行されそうですね。

僕は、 first_value(value) の結果は、category ‘A’と’B’ でそれぞれ一番小さい値である、5と15が返され、 last_value(value) の結果は、それぞれのカテゴリで一番大きい、50と40が返されると予想していました。
そして、nth_value(value, 4) は’A’,’B’それぞれの小さい方から4番目の値である、20と40が帰ってくると思っていました。

ところが実行結果は以下の通りです。

id, category, value, first_value_, last_value_, nth_value_4
5,  'A',      5,     5,            5,           NULL
2,  'A',      10,    5,            10,          NULL
1,  'A',      20,    5,            20,          20
3,  'A',      20,    5,            20,          20
4,  'A',      30,    5,            30,          20
6,  'A',      50,    5,            50,          20
7,  'B',      15,    15,           15,          NULL
9,  'B',      20,    15,           20,          NULL
8,  'B',      30,    15,           30,          NULL
10, 'B',      40,    15,           40,          40

いかがでしょうか。この結果が予想通りだった!って人は(元々仕様を正確に理解していた人以外に)いらっしゃいますか?

first_value_ 列はいいですね。行をcategory列でグループ分けして、valueでソートした後、それぞれのcategoryのグループの最初の値である 5と15 を返してくれています。

問題は、last_value_ 列と、 nth_value_4 列です。
last_value_ の方なんて、同じ行の value 列の値をそのまま返してきていて、関数を使った意味がないです。

なぜこのような挙動になるのかを理解するには、前回の記事で紹介したframeについて知る必要があります。
参考: WINDOW関数のframeのROWSモードについて

WINDOW関数には frameというオプションがあり、何も指定しないとこれが、
RANGE UNBOUNDED PRECEDING
になるのでした。

つまり、PARTITION BY で区切ったグループ内の、最初の行からその行と同じ値の行までを関数の適用範囲とするわけです。これにより、 last_value が指し示す最後の行とは、その行と(ORDER BYで指定した行の値が)同じ値の行まで、の範囲で見た場合の最後の行となり、その結果として同じ行のvalue列の値をそのまま返してきてきたわけです。

ちなみに、 last_value(value) ではなく、 last_value(id) とするとまた微妙に結果が変わります。
category Aで、value列が20の行が二つありますが、2行とも 3が返ってきます。

nth_value_4 の方も原理は同じです。最初の行からその行と同じ値の行まで、の範囲に含まれる行数が4行未満の場合、返す値がないのでNULLになってしまうのです。
未指定の場合 frame は ROWSモードではなくRANGEモードなので、 その行までに4行含まなくても、その行と同じ値の行まで含めて4行確保できれば値を返してくれます。
だから、結果の3行目(id:1の行)は20が返ってきているのです。

ここまでの説明で、 last_value と nth_value って使えない、もしくは使いにくいなと感じられた人も多いかと思います。
これらの関数でイメージ通りに PARTITION BY で区切った範囲の最終行や、 n番目の行の値を素直に出力したい場合は、 合わせて frameを指定してあげることで可能になります。
PARTITION BY で区切った範囲の最初から最後までなので、
ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
ですね。

要するに次のように書けば良いです。
(記述量が多くなるので、SELECT句内の該当行だけ書きました)

  last_value(VALUE) OVER(
    PARTITION BY
          category
      ORDER BY
      VALUE
        ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
  ) AS last_value_,
  nth_value(VALUE, 4) OVER(
    PARTITION BY
      category
    ORDER BY
      VALUE
    ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING
  ) AS nth_value_4

これで出力結果を載せる前に書いてた、イメージしてた通りの結果が得られます。

WINDOW関数のframeのROWSモードについて

またWINDOW関数の記事です。例によって、SQLクエリエンジンはPrestoを仮定し、僕はトレジャーデータで検証して記事を書いています。
今回は前回の記事でチラチラと登場していたframeについて説明を試みます。
(正直、僕もこれまであまり使ってこなかったので詳しくはありません。)

ドキュメントは最近ずっと参照しているここ。
参考: Window Functions

さて、改めてWINDOW関数の説明を見ると、WINDOW関数は次の形で呼び出すと書かれています。

function(args) OVER (
    [PARTITION BY expression]
    [ORDER BY expression [ASC|DESC]]
    [frame]
)

function は SUMとか ROW_NUMBER などのことですね。
大括弧書きの部分はオプションなので省略可能です。よく見ると、ORDER BY の次に [frame] というのがあります。これが今回の記事のテーマです。

WINDOW関数で集計する時は、まず各レコードを、PARTITION BY で指定した列の値によってグループ分け、ORDER BY で指定した列と、指定したソート順(ASC/DESC)によって整列します。
そして、 「frame で指定した範囲」で集計するのです。

前回の記事で見た通りframe を何も指定しないと、ORDER BY がなければPARTITION BY で区切ったグループ全体を集計し、ORDER BYがあれば、PARTITION BYで区切ったグループ内の最初の行から、その行と同順位の行までを集計します。

これが、frameを指定することで、 ORDER BYを指定した上でも、「PARTITION BY で区切ったグループ全体」とか、「その行からPARTITION BY で区切った最後の行まで」とか、「その行の3行前から5行後ろまで」など、さまざまなレンジで集計ができます。これによって「直近7レコードの移動平均」みたいなこともできるわけですね。

frameの指定方法ですが、ドキュメントには次のように記載されています。(次の2行のどちらか)

{RANGE|ROWS} frame_start
{RANGE|ROWS} BETWEEN frame_start AND frame_end

最初に、RANGEモードかROWSモードをを指定し、1番目の記法では開始行(frame_start)のみ指定、2番目の記法では開始行(frame_start)と終了行(frame_end)を指定します。1番目の記法を採用した場合は、終了行はCURRENT ROW を指定するのと同じ動きになります。

そして、frame_start/ frame_start は 次の5種類の表記が使えます。 (ROWSモードでしかサポートされていないのが2種類あるのでRANGEモードでは3種類)

UNBOUNDED PRECEDING
expression PRECEDING  -- only allowed in ROWS mode
CURRENT ROW
expression FOLLOWING  -- only allowed in ROWS mode
UNBOUNDED FOLLOWING

UNBOUNDED PRECEDING は一番最初の行からで、UNBOUNDED FOLLOWINGは一番最後の行までの意味です。CURRENT ROW は ROWSモードであればその行まで、RANGEモードであれば、その行と同じ順位の行までです。(前回の記事で試してるのは未指定なのでデフォルトのRANGEモードの方です。)

expression PRECEDING と expression FOLLOWING は ROWSモードでしか使えませんが、expression に数字を入れて、 3 PRECEDING とか、 5 FOLLOWING のように書きます。それぞれ3行前、5行後ろの意味です。

さて、だらだら説明を書いてきたのでこの辺で一つやってみましょう。設定できるオプションが多くておそらく用途も多いROWS モードの方をやってみます。
例が無駄に複雑になるので次の実行例ではPARTITION BY は使いませんでしたが、もちろん本当は使えます。集計関数は挙動を確認しやすいSUMを例にしていますがこちらももちろん他の関数も使えます。

WITH
    sample_table AS(
        SELECT 
            id,
            value
        FROM
            UNNEST(
                ARRAY[1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
                ARRAY[20, 10, 40, 30, 50, 30, 20, 60, 10, 30]
            ) AS t(
                id,
                value
            )
)
SELECT 
    id,
    value,
    SUM(value) OVER() AS sum_1,
    SUM(value) OVER(ROWS UNBOUNDED PRECEDING) AS sum_2,
    SUM(value) OVER(ORDER BY id ROWS BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING) AS sum_3,
    SUM(value) OVER(ORDER BY id ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS sum_4,
    SUM(value) OVER(ORDER BY id ROWS BETWEEN 3 PRECEDING AND 1 PRECEDING) AS sum_5
FROM
    sample_table

-- 以下出力結果
id, value, sum_1, sum_2, sum_3, sum_4, sum_5
1,  20,    300,   20,    300,   30,    NULL
2,  10,    300,   30,    280,   70,    20
3,  40,    300,   70,    270,   80,    30
4,  30,    300,   100,   230,   120,   70
5,  50,    300,   150,   200,   110,   80
6,  30,    300,   180,   150,   100,   120
7,  20,    300,   200,   120,   110,   110
8,  60,    300,   260,   100,   90,    100
9,  10,    300,   270,   40,    100,   110
10, 30,    300,   300,   30,    40,    90

色々試したので順番に見ていきましょう。
sum_1 列はただの比較用です。前回の記事でも見ましたが、ORDER BY含めて何も指定しなければ全体の和になります。これは、デフォルトのRANGEモードが作動しており、行同士の間に順序が定義されていないので、全行が同一順位として扱われ、同じ順位の行まで足すという挙動により全行が足されたからです。

sum_2 以降が ROWSモードになります。 まず sum_2 ではROWSモードへの変更のみ行いました。(RANGEモードも未指定の場合の挙動は RANGE UNBOUNDED PRECEDING なので、本当にモードだけ変えています。)
ご覧の通り、ROWS モードでは ORDER BYがなくてもvalue列の累積和になっています。
これはROWSモードでは、「先頭の行から、その行まで」集計するという挙動になったからです。前回の記事でソートに指定した列に同じ値ががあった場合に少し不思議な挙動をするという話を書きましたが、ROWSモードではそれも発生せず、普通に累積和になります。

さて、この後のsum_3以降がframe機能の本番で初期設定ではなく、frameの記法を使って範囲を指定しています。
sum_3 では、 BETWEEN CURRENT ROW AND UNBOUNDED FOLLOWING として、その行から(例ではPARTITION BY使ってないけど、使ってればそのグループ内の)最後の行まで、という指定になっています。逆順の累積和っぽい挙動になっているのが見て取れると思います。

sum_4 は、 BETWEEN 1 PRECEDING AND 1 FOLLOWING と指定しており、これは1行前から1行後ろまで、合計3行の和をとっています。
例えば id 5 の行を見ると、 30+50+30 = 110 になっていますね。

sum_5 は、 BETWEEN 3 PRECEDING AND 1 PRECEDING と指定していて、これは3行前から1行前までの合計3行の和を取るものです。1行目(id:1の行)は集計対象のデータがないのでNULLになってますね。
ここで言いたかったことは、BETWEENで指定するレンジの中にその行自身を含める必要はないということです。
これを使うと、例えば日別の時系列データで、BETWEEN 7 PRECEDING AND 1 PRECEDING とすることで、前日までの過去1週間の集計値(平均など)とその日の値を比べる、といったことも可能になります。

さて、ここまでの説明で、 frameのROWSモードの使い方は概ね網羅できたのではないかと思います。集計の幅がグンと広がると思うのでぜひ使ってみてください。

一方で、 RANGE モードについてはほぼ放置しています。

Presto 以外の エンジンの中には、 実はRANGEモードをよりしっかりサポートしているものもあるそうです。その場合は、ORDER BYで指定した列の値に従って、値の差分が何以下の行を集計対象にするとか、そういう指定ができるそうです。(expression PRECEDING/ expression FOLLOWING がRANGEモードでもサポートされているとか。)

ただ、Prestoではサポートされておらず、トレジャーデータに試しに打ち込んでみても
Window frame RANGE PRECEDING is only supported with UNBOUNDED
などと、エラーが返ってくるだけなので試せていません。
(ではなぜRANGEモードをデフォルトにしているのかとか言いたいことはあるのですが)動かないものは仕方ないので、今回の記事はROWSモード中心に紹介させていただきました。
実際、ROWSモードが動けば大抵の要件には対応可能だと思います。

Prestoでは集計関数をWINDOW関数として扱える

今回もPrestoのWINDOW関数の話です。例によってトレジャーデータで動作を確認しています。参照するドキュメントは直近の数記事と同じこちら。
参考: Window Functions

All Aggregate Functions can be used as window functions by adding the OVER clause. The aggregate function is computed for each row over the rows within the current row’s window frame.

とドキュメントにある通り、Prestoの全ての集約関数はその後ろにOVER()をつけることで、WINDOW関数として動作させることができます。この場合、GROUP BY は要らないくなるので注意してください。

ちなみに、 Aggregate Functions (集約関数) の一覧はこちらのページにあります。
参考: Aggregate Functions

OVER() の中には他のWINDOW 関数と同じように、PARTITION BY と ORDER BY を指定できます。PARTITION BY を指定する場合はもちろん PARTITION BYで指定した列の値でグループ分けして、そのグループ内で集計が走ります。そしてさらに、 ORDER BY の有無によっても、集計される範囲が変わるので注意が必要です。この辺の概念を正しく理解するには、WINDOW関数の Frame という概念(というかオプション)について説明する必要があるので、別記事で紹介しようと思うのですが、今記事Frameを指定しない場合の挙動で説明します。

ORDER BY を指定しない場合、集計範囲は、PARTITION BYで指定されたグループ内の全行になります。PARTITION BYもORDER BYもどちらもしてしてない場合は単純にSELECTされた前レコードが集計範囲です。

ORDER BY を指定する場合、集計範囲は、PARTITION BYで指定されたグループ内のレコードをORDER BYでソートしたものの、「先頭からその行と同じ順位の行まで」になります。

このように言葉で書いてもわかりにくいと思うので、一番基本的な集約関数であるSUM() を使って、実際にやってみます。

WITH sample_table AS(
    SELECT 
        id,
        category,
        value
    FROM
        UNNEST(
            ARRAY[1, 2, 3, 4, 5, 6],
            ARRAY['A', 'A', 'A', 'A', 'B', 'B'],
            ARRAY[20, 10, 20, 30, 15, 30]
        ) AS t(
            id,
            category,
            value
        )
)
SELECT 
    id,
    category,
    value,
    SUM(value) OVER() AS sum_over_1,
    SUM(value) OVER(PARTITION BY category) AS sum_over_2,
    SUM(value) OVER(PARTITION BY category ORDER BY id) AS sum_over_3,
    SUM(value) OVER(PARTITION BY category ORDER BY value) AS sum_over_4
FROM
    sample_table

-- 以下出力結果
id, category, value, sum_over_1, sum_over_2, sum_over_3
1,  'A',      20,    125,        80,         20 
2,  'A',      10,    125,        80,         30
3,  'A',      20,    125,        80,         50
4,  'A',      30,    125,        80,         80
5,  'B',      15,    125,        45,         15
6,  'B',      30,    125,        45,         45

OVERの中に何も指定していないsum_over_1 の列の値は全行等しく6行のvalue 列の値の合計である125になりましたね。
そして、PARTITON BY を指定した sum_over_2 列の値は、category列の値が ‘A’なのか ‘B’なのかによって変わり、それぞれ該当する行のvalue列の値になっています。

そして、注目するのは ORDER BY も指定した sum_over_3 列です。
これは、 id でソートした後のvalue列の累積和が返されています。

どちらの値にも用途はたくさんありそうですね。目的に応じて使い分けていきましょう。
例えば累積和の方は、日々の売り上げから累積売り上げを算出するといったことができますし、全体の集計値の方はもう少し工夫して、そのグループ内のシェアを計算するなどの用途で使えます。要するにSELECT句に次のように入れたりです。(1.0を最初に掛けるのは型をDOUBLEに変換するため。CASTしても良し)
1.0 * value / SUM(value) OVER(PARTITION BY category)
次のように、グループ内の平均との差分を求めるというのも良いです。
value – AVG(value) OVER(PARTITION BY category)

さて、ORDER BYした時の挙動について、もう一点補足です。
先ほど説明の中で、「先頭からその行と同じ順位の行まで」などというまどろこしい表現を使いました。「先頭からその行まで」ではないんですね。
先ほどの例では ORDER BY で id列を指定し、 id列には重複した値がなかったのであまり気にしなくてよかったのですが、値の重複があり、ソートした順位が同順になる行が複数あると少し気をつける必要があります。 ということで、 id ではなくvalue 列でソートしてみたのが次の結果です。

WITH sample_table AS(
    SELECT 
        id,
        category,
        value
    FROM
        UNNEST(
            ARRAY[1, 2, 3, 4, 5, 6],
            ARRAY['A', 'A', 'A', 'A', 'B', 'B'],
            ARRAY[20, 10, 20, 30, 15, 30]
        ) AS t(
            id,
            category,
            value
        )
)
SELECT 
    id,
    category,
    value,
    SUM(value) OVER(PARTITION BY category ORDER BY value) AS sum_over_4
FROM
    sample_table

-- 以下出力結果
id, category, value, sum_over_4
2,  'A',      10,    10
1,  'A',      20,    50
3,  'A',      20,    50
4,  'A',      30,    80
5,  'B',      15,    15
6,  'B',      30,    30

さて、sum_over_4 列を順番に見ていきましょう。 1行目の結果はいいですね。 valueと同じ10です。そして、4行目の結果は1〜4行目の和で、 10+20+20+30=80 となります。
注目すべきは2行目と3行目です。これがどちらも50になっています。これは、10+20+20の結果です。

3行目が50なのはいいとして、2行目も50なのは少し違和感ありますね。
これが先ほど説明した、「先頭からその行と同じ順位の行まで」を集計するということです。

このような挙動になる理由を正しく理解するには、この記事の最初の方にも書きましたがFrameという概念を理解する必要があります。

別の記事でちゃんと書こうと思うのですが、Frameには行を基準に集計範囲を決めるROWモードと、値を基準に集計範囲を決めるRANGEモードというのがあります。
そして、Frameを指定しなかった場合の挙動は、値を基準とするRANGEモードなのです。

もう少し書くと、Frameを指定しないと、
RANGE UNBOUNDED PRECEDING
と指定するのと同じになり、これが「先頭からその行と同じ順位の行まで」を意味します。
もっと正確にいうと、 UNBOUNDED PRECEDING は、「先頭から」という意味で、「どこまで」は省略されており、「どこまで」を省略した時のデフォルトの挙動が「CURRENT ROW」を指定した場合に等しく、RANGEモードにおけるCURRENT ROWはその行ではなく、「その行と同じ値の行まで」を意味します。ややこしいですね。

とりあえず、Frameについては何も指定しなかった場合にどんな挙動になるのか、をしっかり理解しておくことが重要だと思います。想定と違った動きをしていた場合、SUMの場合は比較的気付きやすいですが平均(AVG)や分散(VAR_SAMP)の場合は見落としがちです。

Prestoで前後のレコードの値を参照する

3記事連続ですが、今回もPrestoのWindow関数の話です。例によって、TreasureDataで動かすことを想定しています。
Window関数の話なのでドキュメントの参照するページもこちら。
参考: Window Functions
今回は、Value Functionsのセクションですね。

さて、SQLで集計業務を行なっていて、同じ行内の別の列の値ではなく、別の行の値を参照する必要が出てくる場面は結構あります。特に前後隣(上下隣と呼ぶ方が直感的かも)の値を参照する場面は多々あります。

例えば、Webサイトのアクセス分析において、ユーザーが1つ前に見ていたページを取得してページ遷移を確認したり、次のページのアクセス時刻との差分からそのページの滞在時間を測ったりする場合です。そのほかにも日毎に集計された時系列データにおいて前日からの変化量を見る、といった用途もあるでしょう。

そのように、前後の値を取得したい時は、LAG() と LEAD() という関数を使用することができます。前の値を取得したい場合が LAG()で、後の値を取得したい場合に使うのがLEAD()です。

LEADの方を例に使い方を説明すると、次のようになります。

LEAD({値を取得したい列名}, {何行後のレコードを取得したいか}, {デフォルト値})
OVER(PARTITION BY {グループ分けに使う列名} ORDER BY {ソートに使う列名} {ASC/DESC})

{何行後のレコードを取得したいか} は 0以上の値が指定でき、省略した場合は1になります。
{デフォルト値}は、PARTITION BY で指定したグループ内に指定した行数だけ後のデータがなかった場合に返す値です。省略した場合はNULLになります。

PARTITION BY、ORDER BY は他の Window関数と同じですね。PARTITION BYを使わなかった場合は全レコードまとめて処理されます。ORDER BY も省略できますが、意図せぬ順番で結果が返ってくることがあるので、原則何か指定しましょう。

説明が長くなってきたので実際にやってみます。次の例は、A/B/C 3人のユーザーの架空のWEBページのアクセスログを集計するものです。(1時間おきにアクセスするWebサイトって現実的にあり得ないとか色々ツッコミどころはありますがあくまでも関数の動きを見やすくするための例なのでご了承ください。)
LAGとLEADを使って、ユーザーごとに、そのページの前後にアクセスしたページを出力しています。

WITH
    sample_table AS (
        SELECT 
            access_at,
            user,
            page
        FROM
            UNNEST(
                ARRAY['00:00:00', '01:00:00', '02:00:00', '03:00:00', '04:00:00',
                      '05:00:00', '06:00:00', '07:00:00', '08:00:00', '09:00:00'],
                ARRAY['A', 'A', 'A', 'A', 'B', 'B', 'B', 'C', 'C', 'C'],
                ARRAY['page0', 'page1', 'page2', 'page3', 'page4',
                      'page5', 'page6', 'page7', 'page8', 'page9']
            ) AS t(
                access_at,
                user,
                page
            )
  )
SELECT
    access_at,
    user,
    page,
    LAG(page) OVER (PARTITION BY user ORDER BY access_at) AS rag_page,
    LEAD(page) OVER (PARTITION BY user ORDER BY access_at) AS lead_page
FROM
    sample_table
ORDER BY
    access_at

-- 以下出力結果
access_at,  user, page,    rag_page, lead_page
'00:00:00', 'A',  'page0', NULL,     'page1'
'01:00:00', 'A',  'page1', 'page0',  'page2'
'02:00:00', 'A',  'page2', 'page1',  'page3'
'03:00:00', 'A',  'page3', 'page2',  NULL
'04:00:00', 'B',  'page4', NULL,     'page5'
'05:00:00', 'B',  'page5', 'page4',  'page6'
'06:00:00', 'B',  'page6', 'page5',  NULL
'07:00:00', 'C',  'page7', NULL,     'page8'
'08:00:00', 'C',  'page8', 'page7',  'page9'
'09:00:00', 'C',  'page9', 'page8',  NULL

{何行後のレコードを取得したいか} (OFFSET) は省略したので、RAGとLEADは、それぞれ1行前と後のpage列の値を取得しています。 page, rag_page, lead_pageの各列の結果を見比べるとわかりやすいと思います。

rag_pageで1つ前にアクセスしたページ、lead_pageで1つ後にアクセスしたページが取れていますね。

userでグループを切っているので、各ユーザーごとの最初のアクセスのLAGと、最後のアクセスのLEADは対象のレコードがないのでNULLになっています。

もし、関数を呼び出すときに3つ目の引数としてデフォルト値を指定していたら、NULLではなくそのデフォルト値が入った状態で戻されます。

ORDER BY で指定した、access_atに対しては、 ASC/DESCを指定しなかったので、デフォルトのASC(昇順)で動作しています。もし、それぞれDESCを指定したらaccess_atについて降順になり、前後が入れ替わるのでLAGとLEADの結果が入れ替わります。

ORDER BYで自分が何順を指定しているかと、 LAGとLEADのどちらを使っているかはセットで確認し、想定した結果が得られているかどうかは慎重に検証しましょう。

LAGとLEAD、わざわざ2関数使わなくても、LAGのOFFSET に -1 を入れたら LEADの動きになるんじゃないの?と思って試したのですが、これはエラーになりました。OFFSETは0以上の値しか受け付けないようです。 ちなみにOFFSETに0を入れるとその行の値をそのまま返すので、LAGやLEADの意味はなくなります。

Prestoの行番号以外のRanking関数

前回の記事の続きです。
参考: Prestoで行番号や順位を振る関数まとめ

Presto(トレジャーデータ)で行番号を振る3つの関数を紹介したのですが、ドキュメントの同じページの同じセクションには、これら3つの関数以外にも複数の関数が紹介されています。
実は僕はこれらを使ったことがなくて、ドキュメントをみて初めて存在を知ったので、この機会に挙動を確認しておこうと思いました。
参考: Window Functions の Ranking Functionsのセクション

一つ目は CUME_DIST() です。
ドキュメントによると、値の累積分布を返す関数だとあります。各行について、その行より前の行か同じ値の行の数をグループ内の行数で割った値を返すそうです。前の行が小さい値を指すのか、大きい値を指すのかは、ORDER BY で指定した昇順(ASC)/降順(DESC)によって決まります。
例えば、値が 10, 20, 20, 40 ならば、 0.25, 0.75, 0.75, 1 を返してくるイメージですね。
一瞬、RANK()を行数で割った値のことか?と勘違いしたのですが、RANK()は1,2,2,4 なので、これを行数(4)で割ると0.25, 0.5, 0.5, 1 なので少し違います。

少し多めに8個の値で実行してみたのが次の例です。
比較用にRANK() を8で割った結果も一緒につけました。
結果をシンプルにするためにこの例では使いませんでしたが、もちろんOVERの中でPARTITION BY を使ってグループごとに算出することもできますよ。

WITH
    sample_table AS (
        SELECT 
            id,
            value
        FROM
            UNNEST(
                ARRAY[1, 2, 3, 4, 5, 6, 7, 8],
                ARRAY[20, 10, 20, 30, 15, 30, 25, 40]
            ) AS t(
                id,
                value
            )
  )
SELECT
    id,
    value,
    CUME_DIST() OVER (ORDER BY value) AS cume_dist_,
    CAST(RANK() OVER (ORDER BY value) AS DOUBLE)/8 AS rank_norm
FROM
    sample_table

-- 以下出力結果
id, value, cume_dist_, rank_norm
2,  10,    0.125,      0.125
5,  15,    0.25,       0.25
1,  20,    0.5,        0.375
3,  20,    0.5,        0.375
7,  25,    0.625,      0.625
4,  30,    0.875,      0.75
6,  30,    0.875,      0.75
8,  40,    1.0,        1.0

CUME_DIST と RANK(を行数で割った値)を比較すると、同じ値になる行もあれば、複数行で同じ値になっている行については結果が異なることも見て取れますね。

次の関数は、NTILE です。これは NTILE(n) OVER(〜) のように、バケット数n を指定して使います。これは、PARTITION BYで区切られた各グループの行たちを、さらにそれぞれn個のバケットに分割して、バケット番号を振るものです。8行のデータに対して、ntile(4)を実行したら、1,1,2,2,3,3,4,4 と番号を振る感じですね。元の行数がバケット数でちょうど割り切れない場合は、最初の方のバケットに優先的に割り当てられます。つまり、7行のデータを4バケットに割ったら、1〜3番目のバケットに2個、4番目のバケットに1個割り当てられて、1,1,2,2,3,3,4 となります。

実際にやってみます。(この例でもPARTITION BY使ってませんが、実際は使えます。)

WITH
    sample_table AS (
        SELECT 
            id,
            value
        FROM
            UNNEST(
                ARRAY[1, 2, 3, 4, 5, 6, 7],
                ARRAY[20, 10, 20, 30, 35, 30, 25]
            ) AS t(
                id,
                value
            )
  )
SELECT
    id,
    value,
    NTILE(4) OVER (ORDER BY value) AS ntile_
FROM
    sample_table

-- 以下出力結果
id, value, ntile_
2,  10,    1
3,  20,    1
1,  20,    2
7,  25,    2
4,  30,    3
6,  30,    3
5,  35,    4

想定通りの結果が出ましたね。今回の例では意図的に、バケットの境目になる2番目と3番目(id 1と3)のvalue を同じ20にしていたのですが、一方がバケット番号1でもう一方は2と振り分けられました。値が同じことよりも各バケットに規定の個数振り分けることが優先されている動きになりました。

この記事で最後に紹介するのは、 PERCENT_RANK() です。
その行のRANK() の結果を $r$、そのグループの行数を$n$とすると、$(r-1)/(n-1)$を返す、というドキュメント通りの説明が一番わかりやすいと思います。
そのままなので、やってみます。(先の2関数と同じように、PERCENT_RANKも本当はPARTITION BYが使えます。) 比較用にRANK() の結果もつけました。

WITH
    sample_table AS (
        SELECT 
            id,
            value
        FROM
            UNNEST(
                ARRAY[1, 2, 3, 4, 5, 6, 7, 8, 9],
                ARRAY[20, 10, 20, 30, 35, 30, 25, 15, 40]
            ) AS t(
                id,
                value
            )
  )
SELECT
    id,
    value,
    RANK() OVER (ORDER BY value) AS rank_,
    PERCENT_RANK() OVER (ORDER BY value) AS percent_rank_
FROM
    sample_table

-- 以下出力結果
id, value, rank_, percent_rank_
2,  10,    1,     0.0
8,  15,    2,     0.125
3,  20,    3,     0.25
1,  20,    3,     0.25
7,  25,    5,     0.5
4,  30,    6,     0.625
6,  30,    6,     0.625
5,  35,    8,     0.875
9,  40,    9,     1.0

0〜1 までの値を振ってくれるので個人的には結構好みの結果です。
さて、(そんな利用例は滅多にないと思いますが、)元々抽出されたレコードが1件だたり、PARTITION BY で区切られたグループ内のデータの行数が1件だけだった場合、
$r=1, n=1$ なので、$(r-1)/(n-1)=0/0$となり、$0$除算のエラーにより結果はNULLになるんじゃないか、と思ってたのですが、どうやらこの場合は$0$を返してくるようです。
NULL考慮をしなくていいので楽ですね。(ちなみにこれは、トレジャーデータで試した場合の挙動の話です。もしかしたら他のPresto環境やDBMSの種類によっては違う動きになるかもしれません。)

Prestoで行番号や順位を振る関数まとめ

タイトルにはPrestoって書いてますが、例によってTreasureDataで検証しています。

集計業務を行なっていて、レコードに番号や順位を振りたい場面は意外にあるものです。レコードを何らかのグループごとに分けて、そのグループ内で特定の列ごとにソートしてそのグループ内で何番目のレコードかな、という情報を取得します。

そのようなときに使える関数がPrestoには複数実装されており、主に次の3つが使えます。
– ROW_NUMBER()
– RANK()
– DENSE_RANK()

ROW_NUMBER は単純で、1,2,3,4,… と通し番号を振ります。
RANKは順位なので、もしソートキーになる列の値が等しければ、同じ番号を振ってくれます。例えば、2位と3位が同じ値だったら、1,2,2,4,…という具合です。
DENSE_RANKはRANKと似ていますが、同じ順位のレコードが発生したとき、その次のレコードの番号を振るときに値が飛びません。先程の例で言えば、1,2,2,3,…と番号を振ります。

これらの関数は全てWINDOW関数と呼ばれるカテゴリに属しており、ドキュメントも次のページにまとまっています。
参考: Window Functions

これらのWINDOW関数は利用する時は、必ず OVER() という関数をセットで使います。

ROW_NUMBER() OVER (PARTITION BY [グループ分けする列] ORDER BY [ソートする列])

のようなイメージです。

PARTITION BY は必須ではなく、指定しなければ全レコードを通した番号を振ってくれ、指定すれば、指定した列の値が等しいレコードでグループを作って、そのグループ内での番号を振ってくれます。

ORDER BY も必須ではなく、未指定であればROW_NUMBER()の場合は、DBがSELECTするときに抽出した順で番号を振ってくれますが、普通は何かしら欲しい順序があると思うので指定するようにしましょう。
RANK() と DENSE_RANK() については、ORDER BYを指定しないと、どの列で順位をつけたら良いか不明なので、全部1を返してしまいます。これらを使う時は必ず指定しましょう。
ソートする列名の後ろに、 ASC(昇順)/DESC(降順) を指定することもできます。省略した場合は、ASC(昇順)です。

PARTITION BY, ORDER BY ともに、カンマ区切りで複数の列を指定することもできます。

さて、文章で説明ばっかり書いちゃったので実際にやってみましょう。
使うデータは実データではなく、最初のWITH句で生成したダミーデータです。以下のレコードを生成してます。
id, category, value
1, ‘A’, 20
2, ‘A’, 10
3, ‘A’, 20
4, ‘A’, 30
5, ‘B’, 15
6, ‘B’, 30

WITH sample_table AS(
    SELECT 
        id,
        category,
        value
    FROM
        UNNEST(
            ARRAY[1, 2, 3, 4, 5, 6],
            ARRAY['A', 'A', 'A', 'A', 'B', 'B'],
            ARRAY[20, 10, 20, 30, 15, 30]
        ) AS t(
            id,
            category,
            value
        )
)
SELECT 
    id,
    category,
    value,
    ROW_NUMBER() OVER(PARTITION BY category ORDER BY value) AS row_number_,
    RANK() OVER(PARTITION BY category ORDER BY value) AS rank_,
    DENSE_RANK() OVER(PARTITION BY category ORDER BY value) AS dense_rank_
FROM
    sample_table

結果は次のようになります。

idcategoryvaluerow_number_rank_dense_rank_
2A10111
1A20222
3A20322
4A30443
5B15111
6B30222

想定通りの結果が得られましたね。

Pandasのデータを割合に変換する

業務で集計したデータを「実数だけではなく割合でも出して欲しい」というオーダーを受けることはよくあります。そんな時に、PandasのDataFrameのデータを列ごとや、行ごとの割合に変換する方法のまとめです。

DataFrameの話に入る前に、まずSeries型でやってみましょう。これは非常に簡単で、元のデータをその合計で割るだけです。

import pandas as pd


# 元のデータ
sr = pd.Series([30, 0, 40, 30, 10])
# 合計で割ると割合になる
print(sr/sr.sum())
"""
0    0.272727
1    0.000000
2    0.363636
3    0.272727
4    0.090909
dtype: float64
"""

続いて、DataFrame のデータを列ごとに、その列の値の和に占める割合に変換する方法を見ていきます。実はこれも簡単でDataFrameにたいしてsum()メソッドを実行すると列ごとの和が得られ、元のDataFrameをその和で割るといい感じにブロードキャストされて望む結果が得られます。
ブロードキャストについてはこちらも参照ください。今回の例で言えば、型が(5, 3) と (3,) なのでブロードキャストされます。
参考: NumPyのブロードキャストで変換できる型

# 元のデータを生成する
df = pd.DataFrame(
    {
        "col1": [0, 60, 80, 60, 0],
        "col2": [10, 80, None, 20, 40],
        "col3": [30, 0, 40, 30, 10],
    }
)
print(df)
"""
   col1  col2  col3
0     0  10.0    30
1    60  80.0     0
2    80   NaN    40
3    60  20.0    30
4     0  40.0    10
"""

# sum() すると列ごとの和が得られる
print(df.sum())
"""
col1    200.0
col2    150.0
col3    110.0
dtype: float64
"""

print(df/df.sum())
"""
   col1      col2      col3
0   0.0  0.066667  0.272727
1   0.3  0.533333  0.000000
2   0.4       NaN  0.363636
3   0.3  0.133333  0.272727
4   0.0  0.266667  0.090909
"""

ここからがこの記事の本題です。

列ごとに割合に変換するのは簡単でしたが、行ごとに割合に変換するのはこのようにはうまくいきません。sum(axis=1) で各行ごとの和は出せますが、それで元のデータフレームを割ろうとすると適切にブロードキャストされないからです。なんか変な結果が戻ってきます。

print(df/df.sum(axis=1))
"""
   col1  col2  col3   0   1   2   3   4
0   NaN   NaN   NaN NaN NaN NaN NaN NaN
1   NaN   NaN   NaN NaN NaN NaN NaN NaN
2   NaN   NaN   NaN NaN NaN NaN NaN NaN
3   NaN   NaN   NaN NaN NaN NaN NaN NaN
4   NaN   NaN   NaN NaN NaN NaN NaN NaN
"""

対処法はいくつかあると思います。一つは、「列ごとの処理は簡単で行ごとの処理が難しいなら行列入れ替えればいい」という発想に基づくものです。単純に転置して割合に変換した後もう一回転置します。

# 転置したDataFrameを作る
df_t = df.T
print((df_t/df_t.sum()).T)
"""
       col1      col2      col3
0  0.000000  0.250000  0.750000
1  0.428571  0.571429  0.000000
2  0.666667       NaN  0.333333
3  0.545455  0.181818  0.272727
4  0.000000  0.800000  0.200000
"""

もう一つ、applyメソッドをaxis=1を指定して使い行ごとのSeriesに対して、割合に変換する方法もあります。個人的にはこちらの方が若干スマートに思えます。
(ちなみに、axis=0 (デフォルト)で実行すると列ごとに割合に変換してくれます)

print(df.apply(lambda x: x/x.sum(), axis=1))
"""
       col1      col2      col3
0  0.000000  0.250000  0.750000
1  0.428571  0.571429  0.000000
2  0.666667       NaN  0.333333
3  0.545455  0.181818  0.272727
4  0.000000  0.800000  0.200000
"""

さて、 lambda 式の中で、 x.sum() ってメソッドが出てきますが、ここがちょっとしたコツです。ここで呼び出されるsum()はSeriesオブジェクトのメソッドのsum()なのですが、これはNaNを無視して和をとってくれます。そのため、index 2 の行(3行目)は、元の値が[80, NaN, 40] ですが、これの和を120として、元の値を割ってくれているわけです。

ここで、x.sum() とせずに、 sum(x)と、Python組み込みメソッドのsum()を呼び出すと結果が変わります。これはNaNを無視せず、NaNが含まれる和はNaNにしてしまうのです。そのため、sum(x)を使うと次のような結果になります。

print(df.apply(lambda x:x/sum(x), axis=1))
"""
       col1      col2      col3
0  0.000000  0.250000  0.750000
1  0.428571  0.571429  0.000000
2       NaN       NaN       NaN
3  0.545455  0.181818  0.272727
4  0.000000  0.800000  0.200000
"""

index 2 の行が全部 NaNになってしまいましたね。元のデータにNaNがなければ気にしなくて良い違いなのですが、うっかりしていると見落としがちな性質なので気をつけましょう。

当然ですが、Series型のデータに対しても、もし元のデータがNaNを含んでいたら、sum(sr)で割るのと、sr.sum()で割るのは結果が変わります。

sr = pd.Series([30, None, 40, 30, 10])
print(sr/sum(sr))
"""
0   NaN
1   NaN
2   NaN
3   NaN
4   NaN
dtype: float64
"""

print(sr/sr.sum())
"""
0    0.272727
1         NaN
2    0.363636
3    0.272727
4    0.090909
dtype: float64
"""

pandasの日付データを週単位で丸める(to_periodを使う方法)

前回の記事では、トレジャーデータで週単位の集計をする方法を紹介しましたが、今回はすでにDBから抽出が終わっているデータを週単位で集計する方法を紹介します。

日時の列をindexに設定してresampleするとか、方法はいろいろあるのですが、週単位の集計の場合、個人的にはto_period メソッドを使って丸めるのが一番気に入っているのでそれを紹介します。

ドキュメントはこちらです。
参考: pandas.Series.dt.to_period — pandas 1.3.2 documentation

とりあえずデータを作っておきます。

import pandas as pd


df = pd.DataFrame({
    'date': [
        '2021-08-01',
        '2021-08-02',
        '2021-08-03',
        '2021-08-04',
        '2021-08-05',
        '2021-08-06',
        '2021-08-07',
        '2021-08-08',
        '2021-08-09',
        '2021-08-10',
    ]
})
print(df)
"""
         date
0  2021-08-01
1  2021-08-02
2  2021-08-03
3  2021-08-04
4  2021-08-05
5  2021-08-06
6  2021-08-07
7  2021-08-08
8  2021-08-09
9  2021-08-10
"""

to_period メソッドは、 datetime系の型の列でなければ使えないので、pd.to_datetime() して型を変換します。

df["date"] = pd.to_datetime(df["date"])

さて、これで準備が整いました。1週間単位で日付を丸めたい場合は、最初の引数(freq)に”W”を指定して to_period メソッドを使えばOKです。

print(df["date"].dt.to_period("W"))
"""
0    2021-07-26/2021-08-01
1    2021-08-02/2021-08-08
2    2021-08-02/2021-08-08
3    2021-08-02/2021-08-08
4    2021-08-02/2021-08-08
5    2021-08-02/2021-08-08
6    2021-08-02/2021-08-08
7    2021-08-02/2021-08-08
8    2021-08-09/2021-08-15
9    2021-08-09/2021-08-15
Name: date, dtype: period[W-SUN]
"""

2021-08-01 は、 2021-07-26(月)〜2021-08-01(日)の週に丸められ、
2021-08-02〜2021-08-08は、2021-08-02(月)〜2021-08-08(日)の週に丸められましたね。
ちなみにこの結果は、Periodというデータ型になっています。
元と同じようにdatetime型で扱いたい場合や、丸めた週の頭の日付にしたいという場合はさらに変換する必要があります。自分はもっぱら次の形で使うことが多いです。

df["week"] = df["date"].dt.to_period("W").dt.to_timestamp()
print(df)
"""
        date       week
0 2021-08-01 2021-07-26
1 2021-08-02 2021-08-02
2 2021-08-03 2021-08-02
3 2021-08-04 2021-08-02
4 2021-08-05 2021-08-02
5 2021-08-06 2021-08-02
6 2021-08-07 2021-08-02
7 2021-08-08 2021-08-02
8 2021-08-09 2021-08-09
9 2021-08-10 2021-08-09
"""

あとは、この週単位に変換した列を使って groupbyして、sumなりcountなり望みの集計をすることで、Pandasのデータを週単位で集計することができます。

週単位以外の基準で集計したい場合、”W”以外の対応した文字を使えば実現可能です。
例えば月単位なら”M”、日単位なら”D”などです。
利用可能な集計基準と対応する文字は、こちらのページにまとまっています。
参考: Offset aliases

ただ、TD_TIME_TRUNCの記事でも似たような話を書きましたが、日単位や月単位で集計したいのであれば、to_periodして、timestumpに戻して、とやるよりも、strftimeなどを使った方が簡単だと思います。例えば月単位で集計したいなら次のように変換できます。

print(df["date"].dt.strftime("%Y-%m-01"))
"""
0    2021-08-01
1    2021-08-01
2    2021-08-01
3    2021-08-01
4    2021-08-01
5    2021-08-01
6    2021-08-01
7    2021-08-01
8    2021-08-01
9    2021-08-01
Name: date, dtype: object
"""

日単位の場合は、 “%Y-%m-%d”です。

ドキュメントになぜか記載がないのですが、週単位で丸める場合、週の始まりの曜日(実際にコードで指定するのは週の終わりの曜日)を指定することもできます。
方法は簡単で、”W”の代わりに、”W-WED”(水曜日終わり、木曜日始まり)、や、
“W-FRI”(金曜日終わり、土曜日始まり)などを指定します。
“W”は”W-SUN”(日曜日終わり、月曜日始まり)と同じ挙動になります。
基本的に”W”を使っていれば良いと思うのですが、開始日を変えたいこともあると思いますので覚えておくと役に立つ場面もあるかもしれません。

トレジャーデータで1週間単位の集計

トレジャーデータのUDFである、TD_DATE_TRUNCの紹介記事です。この記事ではHiveではなく、Prestoを利用することを前提としています。

色々な集計業務を行う中で、週単位の集計を行う必要が出てくる場面は結構あります。トレジャーデータでは、そのような場面に備えて、タイムスタンプを週単位の値に変換してくれるUDFが容易されています。それが、冒頭に挙げたTD_DATE_TRUNCです。

ドキュメント: TD_DATE_TRUNC

使い方は簡単で、以下の構文で利用します。

TD_DATE_TRUNC(
    '集計したい時間の単位を示す文字列',
    元のタイムスタンプ,
    'タイムゾーン文字列'
)

集計したい時間の単位を示す文字列は次の単語が使えます。
minute/ hour/ day/ week/ month/ quarter/ year

ただし、後述しますが、 日単位(day)や月単位(month)などで集計したい場合は、TD_TIME_FORMATを使った方が手軽に目当ての結果を得やすいので、 実際に使う機会があるのは、週単位(week) かクオーター(quarter)に限られると思います。
(そのため、この記事のタイトルも1週間単位の集計としました)

さて、実際に使ってみましょう。例えば元の時刻が ‘2021-08-11 13:24:16’ だったとします。
(日本時間の場合)この時刻のタイムスタンプは、1628655856 です。そこで、これを週単位で切り詰めると次のようになります。

SELECT
    TD_DATE_TRUNC(
        'week',
        1628655856,
        'JST'
    )

上のクエリの結果は、 1628434800 です。このタイムスタンプがいつを表すかというと、’2021-08-09 00:00:00′ となり、元の時刻と同じ週の月曜日のちょうど0時となります。

この例で分かる通り、 TD_DATE_TRUNC はLONG型のタイムスタンプを受け取って、LONG型のタイムスタンプを返します。
実用上はタイムスタンプで結果を得てもいつのことなのかわかりにくいので、TD_TIME_FORMATで整形することになるでしょう。
time 列を週ごとにまとめて集計するであれば次のような書き方になると思います。

SELECT
    TD_TIME_FORMAT(
        TD_DATE_TRUNC(
            'week',
            time,
            'JST'
        ),
        'yyyy-MM-dd',
        'JST'
    )
FROM
    table_name

元々の時刻として、time列ではなく、’2021-08-11 13:24:16’のような文字列で時刻が入っている列を使う場合、タイムスタンプに変換する必要があるので、事前に、TD_TIME_PARSE する必要があります。 例えば、 created_at という列を週単位に集計するのであれば次のようなクエリになります。

SELECT
    TD_TIME_FORMAT(
        TD_DATE_TRUNC(
            'week',
            TD_TIME_PARSE(created_at, 'JST'),
            'JST'
        ),
        'yyyy-MM-dd',
        'JST'
    )
FROM
    table_name

以上のようにして、週単位での集計ができるようになりました。
四半期単位での集計がしたい時は、 week を quarter に置き換えることで同様に実行できます。

さて、最初の方で少し述べましたが、 TD_DATE_TRUNC は day や month でも使うことができます。しかし、日単位や月単位で集計したい場合は、 TD_TIME_FORMAT だけで目的を果たすことができるので、TD_DATE_TRUNCはむしろ使わない方がクエリがスッキリすると思います。

例えば、次の二つのクエリ内の2列は結果が同じです。明らかに2列目の方が記述が少なくて楽ですね。

SELECT
    -- TD_DATE_TRUNCを使った月単位の集計
    TD_TIME_FORMAT(
        TD_DATE_TRUNC(
            'month',
            time,
            'JST'
        ),
        'yyyy-MM-dd',
        'JST'
    ),
    -- TD_TIME_FORMAT を使った月単位の集計
    TD_TIME_FORMAT(time, 'yyyy-MM-01', 'JST')
FROM
    table_name

sedコマンドの使い方メモ

テキストの編集は大抵vimでやっちゃうので滅多に使わないのですが、稀にコマンドラインでファイルの置換を完結させたり、一つのテンプレートファイルから中身を一部置換したファイルを複数生成する必要が発生し、それをコマンドラインで済ませたいことがあります。
そんなときに、sed (Stream Editor) コマンドを使うのですが、その度に使い方を調べているのでここメモしておきます。

基本的な使い方は、
sed -e {編集コマンド} {入力ファイル}
です。パイプラインで利用する場合は{入力ファイル}を省略できます。
ファイル中の aaa を xxx に置換する場合は次のように書きます。

$ cat input.txt
aaa,bbb,ccc
ddd,eee,fff

$ sed -e s/aaa/xxx/g input.txt
xxx,bbb,ccc
ddd,eee,fff

-e は省略可能で、省略した場合は最初の引数が編集コマンドとみなされます。
なので、大抵の場合は -e を省略して大丈夫です。

$ cat input.txt | sed s/aaa/xxx/g
xxx,bbb,ccc
ddd,eee,fff

-e オプションは複数指定することもできて、同時に複数の置換をかけることもできます。

$ sed -e s/aaa/xxx/g -e s/bbb/yyy/g input.txt
xxx,yyy,ccc
ddd,eee,fff

編集コマンドは、シングルクオーテーション、もしくはダブルクオーテーションで囲むこともできます。(置換前後の文字列のどちらかにスペースを含む場合は、確実に囲むようしましょう。そうしないとエラーになります。)

$ sed "s/aaa/X Y G/g" input.txt
X Y G,bbb,ccc
ddd,eee,fff

置換結果を別のファイルに出力する時は、他のシェルコマンド同様にリダイレクションしてあげれば大丈夫です。僕はもっぱらその形で使います。
例えば、 input.txt の aaa を xxx に置換した output.txt を生成するには次のようにします。

$ sed "s/aaa/xxx/g" input.txt > output.txt

元のファイルをそのまま書き換えることもでき、その場合は
-i {拡張子} オプションをつけます。
すると、{元のファイル名}{拡張子} というファイル名でバックアップを取った上で、入力ファイルを書き換えてくれます。
バックアップはいらないよという場合は、拡張子として長さ0の文字列を渡します。

$ sed -i '' -e 's/aaa/xxx/g' input.txt

この時、” をつけ忘れると、 input.txt-e という変な名前のファイルが残ってしまうので注意してください。

多くの編集コマンドを同時に実行したい場合などは、編集コマンドをまとめたファイルを用意しておき、それを -f オプションで渡すこともできます。

$ cat edit.sed
s/aaa/xxx/g
s/bbb/y y y/g
s/ccc/zzz/g

$ sed -f edit.sed input.txt
xxx,y y y,zzz
ddd,eee,fff

あとは、入力ファイルはスペース区切りで複数同時に渡すことも可能です。
-i オプションをつけている場合は、渡した入力ファイルたちがそれぞれ編集されます。
-i オプションがない場合は、それぞれのファイルを編集した結果が連結されて標準出力に返されます。

sed コマンドで文字列を置換することに関して主に知っておくべきことはこれくらいかなと思います。