Spark SQL ウィンドウ関数を使用して Became_Active 日付をユーザー ログイン データに効率的に割り当てるにはどうすればよいですか?

ウィンドウ関数を使用した Spark SQL での Became_Active 日付割り当ての最適化

この例では、特定の時間枠を考慮して、ユーザーのログイン データに became_active の日付を割り当てる方法を示します。 単純なウィンドウ関数のアプローチで十分に思えるかもしれませんが、特に古い Spark バージョンの場合、より堅牢なソリューションを以下に示します。

Spark 3.2 以降

Spark 3.2 以降のバージョンではセッション ウィンドウ (SPARK-10816、SPARK-34893) が提供され、このタスクが大幅に簡素化されています。 これらの組み込み関数は、セッションの識別と日付の割り当てに直接対応します。 セッションウィンドウの利用の詳細については、Spark のドキュメントを参照してください。

3.2 より前の Spark バージョン

3.2 より前の Spark バージョンの場合は、複数ステップのアプローチが必要です。

1. 必要な関数をインポート:

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{coalesce, datediff, lag, lit, min, sum}

2. ウィンドウの定義:

val userWindow = Window.partitionBy("user_name").orderBy("login_date")
val userSessionWindow = Window.partitionBy("user_name", "session")

3. セッション識別:

このステップでは、ログイン日の 5 日間のギャップに基づいて、新しいユーザー セッションの開始を決定します。

val newSession = (coalesce(
  datediff($"login_date", lag($"login_date", 1).over(userWindow)),
  lit(0)
) > 5).cast("bigint")

val sessionized = df.withColumn("session", sum(newSession).over(userWindow))

4. セッションごとの最も早いログイン日:

最後に、各セッション内の最も早いログイン日が became_active 日付として割り当てられます。

val result = sessionized
  .withColumn("became_active", min($"login_date").over(userSessionWindow))
  .drop("session")

このメソッドは、定義された時間枠を遵守して各ユーザーの became_active 列に効果的にデータを入力し、3.2 より前の Spark バージョンの再帰的アプローチよりもクリーンなソリューションを提供します。 session 列は仲介として使用され、その後削除されます。

以上がSpark SQL ウィンドウ関数を使用して Became_Active 日付をユーザー ログイン データに効率的に割り当てるにはどうすればよいですか?の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。