data_stream <- sc %>% stream_read_json(s3_folder)      %>%
    filter(event=="Added to Cart")                 %>%
    filter(properties.serviceType == 'Food')       %>%
    sdf_select(event,
        cart_id     = properties.CartId,
        recommender = properties.experimentDetails.id,
        variant     = properties.experimentDetails.variant) %>%
    filter(!is.na(recommender))                    %>%
    group_by(recommender)                          %>%
    summarize(n = n())                             %>%
    stream_write_memory("data_stream", mode = "complete")