stream

stream.R

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")