1 of 36

The magic behind your Lyft ride prices

A case study on machine learning and streaming

Strata Data, San Francisco, March 27th 2019

Rakesh Kumar | Engineer, Pricing

Thomas Weise | @thweise | Engineer, Streaming Platform

go.lyft.com/dynamic-pricing-strata-sf-2019

2 of 36

Agenda

Introduction to dynamic pricing
Legacy pricing infrastructure
Streaming use case
Streaming based infrastructure
Beam & multiple languages
Beam Flink runner
Lessons learned

2

3 of 36

Dynamic Pricing

Supply/Demand curve

ETA

Pricing

Notifications

Detect Delays

Coupons

User Delight

Fraud

Behaviour Fingerprinting

Monetary Impact

Imperative to act fast

Top Destinations

Core Experience

3

Lyft is a ride sharing company, our goal is to match passengers with drivers. Our goal is to ensure our users - both drivers and passengers are happy. This means short ETA, fair pricing, fair earnings for drivers and timely notifications.

To achieve these a lot of decisions are made in realtime. For example, to ensure fair pricing and fair pay, supply/demand curves are taken into account. This changes with traffic and other conditions so it is imperative our models adapt to that in realtime.

We want to make sure our users are getting notified in a timely manner about delays, route changes, coupons etc.

We also want to combat Fraud on the platform, since there can be severe monetary impact. For example: we look at anomalies in user behaviour, for example if a user enters several different credit cards one after another thats a red flag. And once again it is important to detect these patterns as quickly as possible.

4 of 36

Introduction to Dynamic Pricing

4

5 of 36

What is prime time?

Location + time specific multiplier on the base fare for a ride

e.g. "in downtown SF at 5:00pm, prime time is 2.0"

Means we double the normal fare in that place at that time

Location: geohash6 (e.g. ‘9q8yyq’)

Time: calendar minute

5

6 of 36

Why do we need prime time?

6

Balance supply and demand to maintain service level�
State of marketplace is constantly changing �
"Surge pricing solves the wild goose chase" (paper)

Some people assume it's a way for us to make money but it's not... we would prefer not to have to do it.

But we have to in order to balance supply and demand

example-- bad weather or concert gets out, suddenly we have lots of people requesting and not enough drivers

if we don't PT all drivers will quickly be used up and waits for rides will increase unacceptably;

also more and more drivers will spend their time en route to pick up passengers rather than giving rides to passengers (utilization goes down)

“surge” pricing only at high demand times, brings demand back under control and avoids these catastrophic failures.

https://papers.ssrn.com/sol3/papers.cfm?abstract_id=2890666

"optimum level of throughput maximizing demand" is what we're trying to find

Abstract

Ride-hailing apps usually match more efficiently than taxis, but they can enter a failure mode anticipated by Arnott (1996) that we call wild goose chases. High demand depletes the platform of idle drivers, so cars must be sent to pick up distant customers. Time wasted on pick-ups decreases drivers’ earnings, leading to exit and exacerbating the problem. Raising prices, either by keeping them consistently high or “surge” pricing only at high demand times, brings demand back under control and avoids these catastrophic failures. Banning surge pricing would thus likely result in always-high prices. Alternative solutions would undermine ride-hailing’s brand promise.

7 of 36

Legacy Pricing Infrastructure

7

8 of 36

Legacy architecture: A series of cron jobs

Ingest high volume of client app events (Kinesis, KCL)�
Compute features (e.g. demand, conversation rate, supply) from events�
Run ML models on features to compute primetime for all regions (per min, per gh6)

SFO, calendar_min_1: {gh6: 1.0, gh6: 2.0, ...}

NYC: calendar_min_1: {gh6, 2.0, gh6: 1.0, ...}

8

gigantic optimization problem to find PT that gives "optimum level of throughput maximizing demand" subject to constraints. talk about inputs? (how many drivers do we need to hold in reserve to maintain some level of eta, what effect does PT have on conversions, what level of conversions are we seeing, what level of demand are we seeing...) And also intricacies like mode specific PT

So that's what PT is. How do we calculate it? At a high level-- we ingest client events (app opens, ride requests, etc) that come in constantly at high volume. We process these events to generate features that are input to machine learning models that output the necessary PT to achieve our desired ETA/driver reserve levels (or whatever objective we decide upon),

At each stage, essentially we have a bunch of worker machines running cron jobs. We distribute work across workers using a 'region' abstraction -- i.e. San Francisco, New York. We use redis to store intermediate results and to handle region-level locks so two workers don't try to process the same region.

region-level distribution of work, with the help of redis lock.

�Requirements:

1. Low latency, high reliability, scalability -- We need to process all events per min and respond quickly to changes in market... market conditions can change quickly (e.g. weather). Crucial to business. We're growing by leaps and bounds so our event volume is growing accordingly

2. Allow quick iteration to support new ML models and features -- our data scientists are constantly cooking up new models and we need to be able to implement them with minimal hassle

~100 features per geohash, ~3 million geohashes �total, per minute = ~400 billion features per day

9 of 36

Problems

Latency�
Code complexity (LOC)�
Hard to add new features involving windowing/join (i.e. arbitrary demand windows, subregional computation)�
No dynamic / smart triggers

9

10 of 36

Can we use Flink?

10

11 of 36

Streaming Stack

11

Streaming

Application

(SQL, Java)

Stream / Schema

Registry

Deployment

Tooling

Metrics & Dashboards

Alerts

Logging

Amazon

EC2

Amazon S3

Wavefront

Salt

(Config / Orca)

Docker

Source

Sink

11

12 of 36

Streaming and Python

Flink and many other big data ecosystem projects are Java / JVM based

Team wants to adopt streaming, but doesn’t have the Java skills
Jython != Python

Use cases for different language environments

Python primary option for Machine Learning

Cost of many API styles and runtime environments

12

13 of 36

Solution with Beam

Streaming

Application

(Python/Beam)

Source

Sink

13

14 of 36

Streaming based Pricing Infrastructure

14

15 of 36

Pipeline (conceptual outline)

15

kinesis events (source)

aggregate and window

filter events

run models to generate features (culminating in PT)

internal services

redis

ride_requested, app_open, ...

unique_users_per_min,

unique_requests_per_5_min, ...

conversion learner,

eta learner, ...

Lyft apps (phones)

valid sessions, dedupe, ...

16 of 36

Details of implementation

Filtering (with internal service calls)
Aggregation with Beam windowing: 1min, 5min (by event time)
Triggers: watermark or stateful processing
Machine learning models invoked using stateful Beam transforms
Final gh6:pt output from pipeline stored to Redis

16

17 of 36

Gains

60% reduction in latency
Reuse of model code
10K => 4K LOC
300 => 120 AWS instances

17

18 of 36

Beam and multiple languages

18

19 of 36

The Beam Vision

End users: who want to write pipelines in a language that’s familiar.�
SDK writers: who want to make Beam concepts available in new languages. Includes IOs: connectors to data stores.�
Runner writers: who have a distributed processing environment and want to support Beam pipelines

Beam Model: Fn Runners

Apache Flink

Apache Spark

Beam Model: Pipeline Construction

Other

Languages

Beam Java

Beam Python

Execution

Cloud Dataflow

Execution

https://s.apache.org/apache-beam-project-overview�

19

20 of 36

Multi-Language Support

Initially Java SDK and Java Runners
2016: Start of cross-language support effort
2017: Python SDK on Dataflow
2018: Go SDK (for portable runners)
2018: Python on Flink MVP
Next: Cross-language pipelines, more portable runners�

20

21 of 36

Python Example

p = beam.Pipeline(runner=runner, options=pipeline_options)

(p

| ReadFromText("/path/to/text*") | Map(lambda line: ...)

| WindowInto(FixedWindows(120)

trigger=AfterWatermark(� early=AfterProcessingTime(60),

late=AfterCount(1))� accumulation_mode=ACCUMULATING)

| CombinePerKey(sum))

| WriteToText("/path/to/outputs")

)

result = p.run()

( What, Where, When, How )

21

22 of 36

Portability (originally)

⋮

input | Sum.PerKey()

Python

input.apply(

Sum.integersPerKey())

Java

SELECT key, SUM(value) FROM input GROUP BY key

SQL (via Java)

⋮

Cloud Dataflow

Apache Spark

Apache Flink

Apache Apex

Gearpump

Apache Samza

Apache Nemo (incubating)

IBM Streams

Sum Per Key

Java objects

Sum Per Key

Dataflow JSON API

https://s.apache.org/state-of-beam-sfo-2018

22

23 of 36

Portability (current)

⋮

input | Sum.PerKey()

Python

stats.Sum(s, input)

Go

SELECT key, SUM(value) FROM input GROUP BY key

SQL (via Java)

⋮

input.apply(

Sum.integersPerKey())

Java

Apache Spark

Apache Flink

Apache Apex

Gearpump

Cloud Dataflow

Apache Samza

Apache Nemo (incubating)

IBM Streams

Sum Per Key

Java objects

Sum Per Key

Portable protos

https://s.apache.org/state-of-beam-sfo-2018

23

24 of 36

Beam Flink Runner

24

25 of 36

Portability Framework w/ Flink Runner

SDK

(Python)

Job Service

Artifact�Staging

Job Manager

Fn Services

(Beam Flink Task)

Task Manager

Executor / Fn API

Provision

Control

Data

Artifact�Retrieval

State

Logging

gRPC

Pipeline (protobuf)

Cluster

Runner

Dependencies

(optional)

python -m apache_beam.examples.wordcount \� --input=/etc/profile \� --output=/tmp/py-wordcount-direct \� --runner=PortableRunner \

--job_endpoint=localhost:8099 \� --streaming

Staging Location

(DFS, S3, …)

SDK Worker (UDFs)

SDK Worker (Python)

Flink Job

25

26 of 36

Portable Runner

Provide Job Service endpoint (Job Management API)
Translate portable pipeline representation to native (Flink) API
Provide gRPC endpoints for control/data/logging/state plane
Manage SDK worker processes that execute user code
Manage bundle execution (with arbitrary user code) via Fn API
Manage state for side inputs, user state and timers

Common implementation for JVM based runners (/runners/java-fn-execution) and portable “Validate Runner” integration test suite in Python!

26

27 of 36

Fn API - Bundle Processing

Bundle size matters!

Amortize overhead over many elements
Watermark hold effect on latency

https://s.apache.org/beam-fn-api-processing-a-bundle

27

28 of 36

Lyft Flink Runner Customizations

Translator extension for streaming sources

Kinesis, Kafka consumers that we also use in Java Flink jobs
Message decoding, watermarks

Python execution environment for SDK workers

Tailored to internal deployment tooling
Docker-free, frozen virtual envs

https://github.com/lyft/beam/tree/release-2.11.0-lyft

28

29 of 36

How slow is this ?

Fn API Overhead 15% ?
Fused stages
Bundle size
Parallel SDK workers
TODO: Cython, protobuf C++ bindings

Fn API

decode, …, window

count

(messages

| 'reshuffle' >> beam.Reshuffle()

| 'decode' >> beam.Map(lambda x: (__import__('random').randint(0, 511), 1))

| 'noop1' >> beam.Map(lambda x : x)

| 'noop2' >> beam.Map(lambda x : x)

| 'noop3' >> beam.Map(lambda x : x)

| 'window' >> beam.WindowInto(window.GlobalWindows(),

trigger=Repeatedly(AfterProcessingTime(5 * 1000)),

accumulation_mode= AccumulationMode.DISCARDING)

| 'group' >> beam.GroupByKey()

| 'count' >> beam.Map(count)

)

29

30 of 36

Fast enough for real Python work !

c5.4xlarge machines (16 vCPU, 32 GB)
16 SDK workers / machine
1000 ms or 1000 records / bundle
280,000 transforms / second / machine (~ 17,500 per worker)
Python user code will be gating factor

30

31 of 36

Beam Portability Recap

Pipelines written in non-JVM languages on JVM runners

Python, Go on Flink (and others)

Full isolation of user code

Native CPython execution w/o library restrictions

Configurable SDK worker execution

Docker, Process, Embedded, ...

Multiple languages in a single pipeline (future)

Use Java Beam IO with Python
Use TFX with Java
<your use case here>

31

32 of 36

Feature Support Matrix (Beam 2.11.0)

https://s.apache.org/apache-beam-portability-support-table

32

33 of 36

Lessons Learned

33

34 of 36

Lessons Learned

Python Beam SDK and portable Flink runner evolving
Keep pipeline simple - Flink tasks / shuffles are not free
Stateful processing is essential for complex logic
Model execution latency matters
Instrument everything for monitoring
Approach for pipeline upgrade and restart
Mind your dependencies - rate limit API calls
Testing story (integration, staging)

34

Beam is New

Notes on 'work in progress': (alpha)� Issues may exist-- e.g. reshuffling losing events, memory leaks, cancel causes exception. But they are usually resolved quickly!

Shuffle is not free

shuffle = group by key or cogroupbykey or combine or reshuffle

each taskmanager has a fixed amount of network buffer memory (% of total jvm) and the usage of network buffer mem scales quadratically w/ number of shuffles

anything that distributes work and requires sharding and sending over the wire

shuffles not only add latency because of network calls, they also take up network buffer space which is a % of total memory

We had to rewrite a lot when we realized GroupByKey and other reshuffles ops incurred significant cost in terms of latency and memory

There were memory leaks on both beam side and our side (via pandas dataframes for instance)

event time vs processing time - ??

staging env -

need to figure out how it should work, rollout is all or nothing....

We had some stuff out of the box -- iterator age for kinesis, memory and cpu stats-- but it was crucial to have detailed statsd logging on intermediate features at all stages-- segmented by region etc

This is obvious and yet we still had to go back and put in some stats that were overlooked. e.g. early on we noticed demand output from old system declining over time relative to new. diagnosing and debugging got us to a point where we weren't even doing any transforms yet and still seeing problem. finally we added event counts per event type per region as a stat within our old system, and comparing to kinesis showed events were getting dropped. So instrument at every stage and make sure you have symmetrical data from old vs new system.

Similarly you want to count output at each intermediate stage...

Finally at the end of the day we still ended up using a lot of log stmts. You create a do method that just outputs the input and insert it into your pipeline. Can't get around log statements

Ratelimiter

It is really important to use the ratelimiter for your api calls for otherwise you can DDOS your services. Generally, it is not recommended to have service call in the api because it can cause bursty traffic and your service may not be able to handle the load. Having a ratelimiter in front of them can help to handle bursty traffic.

Testing Story

Load testing helped us to discover many performance issues before going to production. Beam community were very helpful and active to quickly fix these issues. it could have been a major problem if we have rolled out in production and noticed such issues.

Flink deploy is all or nothing, different from other web services

No canary

How do you handle downtime during deploy? (checkpointing? discard last window?)

35 of 36

We’re Hiring! Apply at www.lyft.com/careers

or email data-recruiting@lyft.com

Data Engineering

Engineering Manager

San Francisco

Software Engineer

San Francisco, Seattle, & New York City

Data Infrastructure

Engineering Manager

San Francisco

Software Engineer

San Francisco & Seattle

Experimentation

Software Engineer

San Francisco

Streaming

Software Engineer

San Francisco

Observability

Software Engineer

San Francisco

36 of 36

Please ask questions!

This presentation:

http://go.lyft.com/dynamic-pricing-strata-sf-2019