はじめに
ショップは閲覧のあったセッション/商品ペアと、イベントの繰り返しを区別できるようになりました。次は、有効なカート追加のない閲覧セッションの割合が高い商品と、それぞれの比較を支える観察量を知りたいと考えています。
閲覧セッションの分母、完了数、未完了数、未完了割合を持つ完全な表を作ります。有効な追加がゼロの商品と同率の商品をすべて残します。結果は追加調査への観察の手掛かりであり、商品やページが訪問を中断させたという証明ではありません。
この独立ラボには前のガイド付きラボで完成した経路モジュールと新しい生ファイルが用意されています。最初の閲覧、セッション/商品、時刻のルールに加え、Spark のグループ化と Parquet を理解している必要があります。Terminal を使います。バッチは有限で終了済み、識別子と秒単位 UTC 時刻は有効、重複イベント ID は完全なコピーです。著者見積もりは35～45分です。ストリーミング、一意な人数の推定、因果推論は扱いません。
完全な商品比較を作る
このステップでは、閲覧経路ごとの 1 行を、商品ごとの 1 行に集約します。Terminal で始めます。提供プログラムのうち analysis.py だけを編集してください。生ファイルと提供された run_analysis.py、read_result.py、support.py は変更せず、派生結果は output/ 以下に置きます。
cd /home/labex/project
cat data/batch.csv
cat analysis.py
モジュールには前のラボから独立に準備した read_events、build_views、build_journeys があります。モジュールは再利用可能な関数を保持します。ドライバーが現在の SparkSession と入力パスを渡し、返された経路 DataFrame を新しい build_report(journeys) に渡します。前の VM や端末セッションの何かに依存しません。
契約を思い出してください。完全に同じイベントコピーは除去し、閲覧のある各 (session_id, product_id) は最初の閲覧を使います。有効な追加は両キーが同じで、その閲覧と同時刻以降です。有効な追加が複数でも完了経路は1件です。ドライバーは local[2]、UTC 整形、ドライバー/ワーカー共通の Python を使い、関数が戻った後で出力を書きます。
閲覧ペア3件、完了ペア2件の商品を手計算します。閲覧セッション3、カートセッション2、カートなし1、カートなし割合 1/3 です。分母のない割合では、1訪問が大きな標本と同程度に説得的に見える恐れがあります。
count("*") は今度は生イベントではなく経路行を数えます。Boolean の completed を long にキャストすると true は1、false は0です。Spark の sum がそれを加算します。完了数を引くと未完了数、正の閲覧セッション数で割ると0～1の割合になります。各グループには少なくとも1閲覧があります。追加しかない商品にはグループも分母もなく、ゼロ分母の割合を捏造しません。
商品集約だけを書く
提供された read_events、build_views、build_journeys は残してください。前のラボで完成した処理なので、入力し直しません。F は既存の Spark 関数の別名です。同じモジュールを nano で開きます。
nano analysis.py
矢印キーで build_journeys の末尾へ移動し、空行を入れます。練習したとおり、Ctrl+W で関数名を検索できます。新しい宣言は左端から始め、本体はスペース 4 個でインデントします。用意された解答テンプレートを埋めるのではなく、build_report 全体を書きます。
まず経路の DataFrame を商品でグループ化します。同じ集約内で 2 つの件数を求めます。閲覧の分母にはすべての行を数え、次に完了を表す Boolean を整数へキャストして合計します。ドライバーが使うフィールド名を保ってください。
def build_report(journeys):
counts = journeys.groupBy("product_id").agg(
F.count("*").alias("view_sessions"),
F.sum(F.col("completed").cast("long")).alias("cart_sessions"),
)
同じ関数内を続けます。withColumn は集約済み件数から名前付きの列を導きます。閲覧数から完了数を引き、同じ正の閲覧分母で割ります。先に完了だけに絞らないでください。有効な追加がゼロの商品が消えてしまいます。割合が最大の商品を 1 つだけ選ばないでください。同率の商品はすべてレポートに必要です。
return counts.withColumn(
"no_cart_sessions", F.col("view_sessions") - F.col("cart_sessions")
).withColumn(
"no_cart_ratio", F.col("no_cart_sessions") / F.col("view_sessions")
)
Ctrl+O で保存し、Enter で analysis.py を確認してから Ctrl+X で終了します。再度開き、保存した関数と、それ以前の関数が残っていることを確認します。
nano analysis.py
Ctrl+W で build_report を探します。誤りがあれば、矢印と Backspace/Delete で影響する行だけを置き換えます。2 つ目の定義は追加しません。継続行のインデントと閉じ括弧を保ってください。変更を保存して終了し、ファイルを読みます。
cat analysis.py
次の実行で SyntaxError や IndentationError が出たら、nano で指定行を探し、周囲の関数、括弧、スペース 4 個のインデントを確認して保存・再実行します。計算が成功しても件数や割合が違う場合は、グループ化、キャスト、分母を確認します。レポートの修正のために VM をリセットしたり、動作する G1 の関数を書き直したりしないでください。
入力と出力のパスを明示してレポート段階を実行します。
bigdata-python run_analysis.py data/batch.csv output/main --stage report
次の完全な論理表を期待します。浮動小数点の表示はさらに桁数が多い場合があります。
product_id view_sessions cart_sessions no_cart_sessions no_cart_ratio
card 1 0 1 1.0
pin 1 0 1 1.0
tea 3 1 2 0.6666666666666666
book 2 1 1 0.5
mug 3 2 1 0.3333333333333333
ドライバーは実際の保存済み Parquet を読み直して表示します。no_cart_ratio 降順、次に商品識別子で並べるのは同率を読みやすくするためです。card と pin の両方を残します。同率だから片方を捨ててよいわけではありません。追加しかない poster は閲覧ペアがないので除外します。
結果は output/main/journeys と output/main/report の型付きデータセットです。後者は文字列の商品 ID、整数の件数、数値の割合を持ちます。Parquet パートの名前や数は実装詳細であり、レポートの本質ではありません。同等の Spark DataFrame/SQL 式は有効です。提供入力からのパイプラインを保ち、変換を Spark 内で行ってください。生の経路を Python に収集したり、RDD/pandas に切り替えたり、チェックポイント後に作り直したりしないでください。
ファイルを読み、スナップショットを再構築する
小さな CSV を読み、元データを変更せず派生スナップショットを置き換えます。Terminal の /home/labex/project で続けます。
準備済みドライバーは同僚向けに output/main/report.csv を1つの小さな UTF-8 CSV として保存します。build_report が戻った後、集約済みの最終商品表だけをシリアライズのため収集します。この最後の表示処理は Spark の変換を置き換えません。実ファイルを確認します。
cat output/main/report.csv
ヘッダーは Parquet レポートと同じ5フィールドです。5商品すべて、card と pin のカート数ゼロ、両者の同率1.0を確認します。CSV は行とフィールドとして読めますが、行数やファイル名だけではこれらの関係は証明できません。
付属リーダーで本物の Parquet 2データセットと CSV を読みます。read_result.py は渡した出力パスを読み、Spark 内で時刻を UTC に整形し、CSV の内容と保存済み表を比較します。欠けた結果を作成・補完しません。
bigdata-python read_result.py output/main
経路10行、商品5行、CSV matches stored report を期待します。欠けたファイルがあれば、正しい入力と出力先で分析を再実行します。リーダーは答えを捏造せず意図的に失敗します。
sha256sum は生ファイルごとの内容の指紋を表示します。*.csv パターンは data/ の CSV を選びます。再構築の前後でこれらの行を比較します。
sha256sum data/*.csv
bigdata-python run_analysis.py data/batch.csv output/main --stage report
bigdata-python read_result.py output/main
sha256sum data/*.csv
同じ入力からは商品や経路の重複がない同じ論理表が得られます。生ファイルの指紋は不変でなければなりません。Parquet の新しいファイル名やバイト列は実行ごとに変わることがあります。再現性の契約は論理内容です。ライターはこの既知の派生先だけを置き換え、生ファイルや別の結果ディレクトリは変えません。
保存済みモジュールを通常サンプルと別の出力先で使います。
bigdata-python run_analysis.py data/normal.csv output/normal --stage report
bigdata-python read_result.py output/normal
商品2行があり、それぞれ閲覧セッション1、カートセッション1、カートなし0、割合0を期待します。主結果はそのまま残ります。同じパラメーター化した関数が両バッチを扱います。
この小さなバッチで card と pin は注目に値しますが、各1観察セッションしかありません。割合が高くても、訪問をやめた理由、一意な人数、今後も傾向が続くことは確定しません。件数とすべての同率を示すと観察を検討でき、因果の主張には追加の証拠が必要です。
採点は実際の結果を再構築せず読みます。また未変更のモジュールを読み込み、識別子と行順が異なる使い捨てバッチで関数を呼び、実際の Spark 入力の系譜と一時出力を確認します。正確なコマンドの綴り、自作の計画、アプリ名マーカーを採点せず、実際の結果を補修しません。関数に書き込みを入れず、シリアライズはドライバーに任せてください。
まとめ
完全な Parquet と CSV の比較表を作成・読み込み、カート追加ゼロの商品と同率を保ち、生入力を維持して派生スナップショットを再構築しました。分母と件数列が各割合を説明します。終了済みセッションのパターンは追加観察を支えるもので、因果の主張ではありません。
