Como calcular time-to-event (survival) em Apache Spark: passo a passo

João Barros 10 de November de 2026 5 min de leitura

Este tutorial mostra como calcular métricas de time-to-event (survival) em Apache Spark para analisar quanto tempo até um evento (por exemplo churn ou falha) e como lidar com censura. Saber calcular time-to-event em Apache Spark é útil para monitorizar retenção, manutenção preditiva e comparar grupos em grande escala.

Pré-requisitos

Python 3.x com PySpark instalado (Spark 3.x).
Acesso a um ambiente local ou cluster com recursos suficientes.
Conhecimentos básicos de DataFrame em PySpark (select, groupBy, withColumn).

Passo 1: Entender os conceitos e montar um exemplo simples

Time-to-event exige três coisas: um identificador de entidade, tempo de início/fim (ou duração) e um indicador de evento (1=evento ocorreu, 0=censurado). Vamos criar um exemplo pequeno para simular utilizadores com datas de início e data do evento (ou null se censurado).

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, when, datediff, to_date

spark = SparkSession.builder.appName("time_to_event_example").getOrCreate()

data = [
    (1, '2022-01-01', '2022-03-15'),
    (2, '2022-02-10', None),
    (3, '2022-01-20', '2022-01-30'),
    (4, '2022-03-01', None)
]

df = spark.createDataFrame(data, ['id','start_date','event_date'])

Passo 2: Calcular duração e variável de censura

Convertendo strings para datas, calculamos a duração em dias entre start_date e event_date. Para linhas censuradas (event_date null) usamos uma data de corte (ex: data de análise) e marcamos censura=0.

from pyspark.sql.functions import lit

analysis_date = to_date(lit('2022-04-01'))

df2 = (df
    .withColumn('start_date', to_date(col('start_date')))
    .withColumn('event_date', to_date(col('event_date')))
    .withColumn('obs_end', when(col('event_date').isNotNull(), col('event_date')).otherwise(analysis_date))
    .withColumn('duration_days', datediff(col('obs_end'), col('start_date')))
    .withColumn('event', when(col('event_date').isNotNull(), lit(1)).otherwise(lit(0)))
)

df2.select('id','start_date','event_date','obs_end','duration_days','event').show()

Passo 3: Agregar para obter funções de sobrevivência por grupo

Uma forma simples de estimar a curva de sobrevivência é calcular, para cada duração t, quantas entidades estão em risco e quantas tiveram evento. Vamos obter tabela de contagens por duração (life-table style).

from pyspark.sql.functions import count, sum as _sum

# tabela por duração em dias
life_table = (df2
    .groupBy('duration_days')
    .agg(
        count('*').alias('n_at_time'),
        _sum(col('event')).alias('events')
    )
    .orderBy('duration_days')
)

life_table.show()

Passo 4: Calcular sobrevivência acumulada (Kaplan-Meier simplificado)

Kaplan-Meier calcula sobrevivência como produto de (1 - d_i / n_i) onde d_i são eventos e n_i número em risco naquele tempo. Em Spark vamos usar operações de janela para calcular o produto cumulativo transformando em log para estabilidade numérica.

from pyspark.sql.window import Window
from pyspark.sql.functions import log, exp, coalesce

lt2 = (life_table
    .withColumn('at_risk', col('n_at_time'))  # em exemplos simples assumimos n_at_time como risco naquele t
    .withColumn('q', col('events') / col('at_risk'))
    .withColumn('log_one_minus_q', log(1 - col('q')))
)

w = Window.orderBy('duration_days').rowsBetween(Window.unboundedPreceding, 0)

km = (lt2
    .withColumn('cum_log', coalesce(_sum(col('log_one_minus_q')).over(w), lit(0)))
    .withColumn('survival', exp(col('cum_log')))
    .select('duration_days','n_at_time','events','survival')
)

km.show()

Passo 5: Exemplo prático por grupo (ex.: plano A vs plano B)

Se tiver um campo de grupo (por exemplo plano), repetimos o processo por grupo usando window particionado. Mostrarei como agregar por grupo e calcular survival por grupo.

# acrescentar grupo de exemplo
from pyspark.sql.functions import when

df3 = df2.withColumn('plan', when(col('id') % 2 == 0, 'B').otherwise('A'))

lt_group = (df3
    .groupBy('plan','duration_days')
    .agg(count('*').alias('n_at_time'), _sum(col('event')).alias('events'))
    .orderBy('plan','duration_days')
)

w_grp = Window.partitionBy('plan').orderBy('duration_days').rowsBetween(Window.unboundedPreceding, 0)

ltg2 = (lt_group
    .withColumn('q', col('events')/col('n_at_time'))
    .withColumn('log_one_minus_q', log(1 - col('q')))
    .withColumn('cum_log', coalesce(_sum(col('log_one_minus_q')).over(w_grp), lit(0)))
    .withColumn('survival', exp(col('cum_log')))
    .select('plan','duration_days','n_at_time','events','survival')
)

ltg2.show()

Verificar o resultado

Confirme que existem colunas duration_days e event em df2 e que life_table/lt_group mostram contagens não nulas. A coluna survival deve começar em 1 (ou próximo) e decrescer após eventos. Verifique com .show() e com contagens por grupo: df3.groupBy('plan').count().

Conclusão

Agora tem um fluxo básico para calcular time-to-event em Apache Spark: preparar datas, definir censura, agregar por tempo e calcular uma estimativa de sobrevivência (Kaplan-Meier simplificada). Próximos passos: ajustar a definição de "n_at_risk" corretamente (considerando riscos acumulados), lidar com largos intervalos de tempo, e visualizar curvas em Python (matplotlib) após recolher resultados. Dica: teste com dados reais e verifique casos de zero eventos para evitar divisões por zero.