Skip to main content
Version: 1.0.6

Contextual-Bandits using Vowpal Wabbit

In the contextual bandit problem, a learner repeatedly observes a context, chooses an action, and observes a loss/cost/reward for the chosen action only. Contextual bandit algorithms use additional side information (or context) to aid real world decision-making. They work well for choosing actions in dynamic environments where options change rapidly, and the set of available actions is limited.

An in-depth tutorial can be found here

Azure Personalizer emits logs in DSJSON-format. This example demonstrates how to perform off-policy evaluation.

Step1: Read the dataset​

import pyspark.sql.types as T
from pyspark.sql import functions as F

schema = T.StructType(
    [
        T.StructField("input", T.StringType(), False),
    ]
)

df = (
    spark.read.format("text")
    .schema(schema)
    .load("wasbs://publicwasb@mmlspark.blob.core.windows.net/decisionservice.json")
)
# print dataset basic info
print("records read: " + str(df.count()))
print("Schema: ")
df.printSchema()
display(df)

Step 2: Use VowpalWabbitFeaturizer to convert data features into vector​

from synapse.ml.vw import VowpalWabbitDSJsonTransformer

df_train = (
    VowpalWabbitDSJsonTransformer()
    .setDsJsonColumn("input")
    .transform(df)
    .withColumn("splitId", F.lit(0))
    .repartition(2)
)

# Show structured nature of rewards
df_train.printSchema()

# exclude JSON to avoid overflow
display(df_train.drop("input"))

Step 3: Train model​

VowpalWabbitGeneric performs these steps:

  • trains a model for each split (=group)
  • synchronizes accross partitions after every split
  • store the 1-step ahead predictions in the model
from synapse.ml.vw import VowpalWabbitGeneric

model = (
    VowpalWabbitGeneric()
    .setPassThroughArgs(
        "--cb_adf --cb_type mtr --clip_p 0.1 -q GT -q MS -q GR -q OT -q MT -q OS --dsjson --preserve_performance_counters"
    )
    .setInputCol("input")
    .setSplitCol("splitId")
    .setPredictionIdCol("EventId")
    .fit(df_train)
)

Step 4: Predict and evaluate​

df_predictions = model.getOneStepAheadPredictions()  # .show(5, False)
df_headers = df_train.drop("input")

df_headers_predictions = df_headers.join(df_predictions, "EventId")
display(df_headers_predictions)
from synapse.ml.vw import VowpalWabbitCSETransformer

metrics = VowpalWabbitCSETransformer().transform(df_headers_predictions)

display(metrics)

For each field of the reward column the metrics are calculated

per_reward_metrics = metrics.select("reward.*")

display(per_reward_metrics)