Comment puis-je attribuer efficacement des dates Became_Active aux données de connexion utilisateur à l'aide des fonctions de fenêtre Spark SQL ?

Optimisation de l'attribution de date Became_Active dans Spark SQL à l'aide des fonctions de fenêtre

Cet exemple montre l'attribution d'une became_active date aux données de connexion de l'utilisateur, en considérant une fenêtre horaire spécifique. Bien qu'une simple approche de fonction de fenêtre puisse sembler suffisante, une solution plus robuste, en particulier pour les anciennes versions de Spark, est présentée ci-dessous.

Spark 3.2 et versions ultérieures

Spark 3.2 et les versions ultérieures proposent des fenêtres de session (SPARK-10816, SPARK-34893), simplifiant considérablement cette tâche. Ces fonctions intégrées traitent directement de l'identification de session et de l'attribution de dates. Reportez-vous à la documentation Spark pour plus de détails sur l'utilisation des fenêtres de session.

Versions Spark avant 3.2

Pour les versions Spark antérieures à 3.2, une approche en plusieurs étapes est nécessaire :

1. Importer les fonctions nécessaires :

<code class="language-scala">import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{coalesce, datediff, lag, lit, min, sum}</code>

2. Définir Windows :

<code class="language-scala">val userWindow = Window.partitionBy("user_name").orderBy("login_date")
val userSessionWindow = Window.partitionBy("user_name", "session")</code>

3. Identification de la séance :

Cette étape détermine le début des nouvelles sessions utilisateur en fonction d'un intervalle de 5 jours dans les dates de connexion.

<code class="language-scala">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))</code>

4. Date de connexion la plus précoce par session :

Enfin, la première date de connexion au sein de chaque session est attribuée comme became_active date.

<code class="language-scala">val result = sessionized
  .withColumn("became_active", min($"login_date").over(userSessionWindow))
  .drop("session")</code>

Cette méthode remplit efficacement la colonne became_active pour chaque utilisateur, en respectant le délai défini, offrant une solution plus propre qu'une approche récursive pour les versions Spark antérieures à 3.2. La colonne session, utilisée comme intermédiaire, est ensuite supprimée.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!