Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

bug; Queuing budget no longer exists. Quality of service is not yet integrated #133

Merged
merged 3 commits into from
Jul 12, 2020
Merged

bug; Queuing budget no longer exists. Quality of service is not yet integrated #133

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f6932cb
bug; Queuing budget no longer exists. Quality of service is not yet i…
Jul 12, 2020
ab167dd
FlyteKit new release
Jul 12, 2020
0fa1f98
removed trailing comma
Jul 12, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
338 changes: 338 additions & 0 deletions .demo/Fee Waiver demo.ipynb
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,338 @@
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Connected to flyte.lyft.net\n"
]
}
],
"source": [
"from flytekit.configuration import set_flyte_config_file, platform\n",
"set_flyte_config_file(\"/Users/kumare/.ssh/notebook-production.config\")\n",
"#set_flyte_config_file(\"notebook.config\")\n",
"\n",
"print(\"Connected to {}\".format(platform.URL.get()))\n",
"\n",
"def print_console_url(exc):\n",
" print(\"http://{}/console/projects/{}/domains/{}/executions/{}\".format(platform.URL.get(), exc.id.project, exc.id.domain, exc.id.name))"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {},
"outputs": [],
"source": [
"query=\"\"\"WITH eme AS (\n",
" SELECT\n",
" ride_id,\n",
" feature_driver_distance_at_arrival_meters,\n",
" feature_driver_distance_at_cancellation_meters,\n",
" feature_dvr_cancellation_rate,\n",
" feature_dvr_no_show_rate,\n",
" feature_dvr_num_voice_calls_to_pax,\n",
" feature_dvr_rides_28d,\n",
" feature_dvr_sum_call_duration,\n",
" feature_dvr_total_rides,\n",
" feature_fixed_fare_amount,\n",
" feature_gh6_total_rides,\n",
" feature_has_waypoint,\n",
" feature_hour_local,\n",
" feature_hour_of_week_local,\n",
" feature_hour_of_week_shifted_local,\n",
" feature_hour_shifted_local,\n",
" feature_is_scheduled_ride,\n",
" feature_num_average_daily_rides_canceled,\n",
" feature_num_rides_taken,\n",
" feature_pax_avg_pickup_time_seconds,\n",
" feature_pax_no_show_rate,\n",
" feature_pax_num_voice_calls_to_driver,\n",
" feature_pax_sms,\n",
" feature_pax_sms_char_len,\n",
" feature_pax_sum_call_duration,\n",
" feature_pax_total_rides,\n",
" feature_pax_unsuccessful_voice,\n",
" feature_request_started_at_to_arrived_at_seconds,\n",
" feature_seconds_since_arrival,\n",
" feature_upfront_fare_amount\n",
" FROM event_model_executed\n",
" WHERE ds >= '{{.inputs.start_date}}'\n",
" AND ds < '{{.inputs.end_date}}'\n",
" AND model = 'dummyfeatureloggingnoshowmodel'\n",
"),\n",
"\n",
"dsi AS (\n",
" SELECT\n",
" ride_id,\n",
" MAX(CAST(is_a1k AS INT)) AS pax_a1k\n",
" FROM dimension_support_issues\n",
" WHERE issue_started_at >= CAST('{{.inputs.start_date}}' AS TIMESTAMP)\n",
" AND issue_started_at < CAST('{{.inputs.end_date}}' AS TIMESTAMP) + INTERVAL '7' DAY\n",
" AND impacted_user = 'passenger'\n",
" GROUP BY ride_id\n",
")\n",
"\n",
"SELECT\n",
" erc.ride_id,\n",
" feature_driver_distance_at_arrival_meters,\n",
" feature_driver_distance_at_cancellation_meters,\n",
" feature_dvr_cancellation_rate,\n",
" feature_dvr_no_show_rate,\n",
" feature_dvr_num_voice_calls_to_pax,\n",
" feature_dvr_rides_28d, \n",
" feature_dvr_sum_call_duration,\n",
" feature_dvr_total_rides,\n",
" feature_fixed_fare_amount,\n",
" feature_gh6_total_rides,\n",
" feature_has_waypoint,\n",
" feature_hour_local,\n",
" feature_hour_of_week_local,\n",
" feature_hour_of_week_shifted_local,\n",
" feature_hour_shifted_local,\n",
" feature_is_scheduled_ride,\n",
" feature_num_average_daily_rides_canceled,\n",
" feature_num_rides_taken,\n",
" feature_pax_avg_pickup_time_seconds,\n",
" feature_pax_no_show_rate,\n",
" feature_pax_num_voice_calls_to_driver,\n",
" feature_pax_sms,\n",
" feature_pax_sms_char_len,\n",
" feature_pax_sum_call_duration,\n",
" feature_pax_total_rides,\n",
" feature_pax_unsuccessful_voice,\n",
" feature_request_started_at_to_arrived_at_seconds,\n",
" feature_seconds_since_arrival,\n",
" feature_upfront_fare_amount,\n",
" CASE WHEN dsi.pax_a1k = 1 THEN TRUE ELSE FALSE END AS should_waive_fee\n",
"\n",
"FROM event_cancels_process_canceled_ride erc\n",
"JOIN experimentation.latest_exposure le\n",
" ON erc.passenger_lyft_id = le.user_lyft_id\n",
" AND erc.ds >= '{{.inputs.start_date}}'\n",
" AND erc.ds < '{{.inputs.end_date}}'\n",
" AND erc.after_arrived = TRUE\n",
" AND (erc.due_to_no_show = TRUE OR erc.canceling_party = 'passenger')\n",
" AND erc.cancel_penalty > 0\n",
" AND le.experiment = 'CP_SXP_PAC_NS_JointHoldout_2019Q4'\n",
" AND erc.occurred_at > le.first_exposed_at\n",
" AND le.variant = 'holdout'\n",
"JOIN eme \n",
" ON erc.ride_id = eme.ride_id\n",
"LEFT JOIN dsi\n",
" ON erc.ride_id = dsi.ride_id\n",
"WHERE erc.ds >= '{{.inputs.start_date}}'\n",
" AND erc.ds < '{{.inputs.end_date}}'\"\"\""
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {},
"outputs": [],
"source": [
"from flytekit.sdk.tasks import inputs\n",
"from flytekit.sdk.types import Types\n",
"from flytekit.common.tasks.presto_task import SdkPrestoTask\n",
"\n",
"schema = Types.Schema([\n",
"('feature_driver_distance_at_arrival_meters', Types.Integer),\n",
"('feature_driver_distance_at_cancellation_meters', Types.Integer),\n",
"('feature_dvr_cancellation_rate', Types.Integer),\n",
"('feature_dvr_no_show_rate', Types.Integer),\n",
"('feature_dvr_num_voice_calls_to_pax', Types.Integer),\n",
"('feature_dvr_rides_28d', Types.Integer),\n",
"('feature_dvr_sum_call_duration', Types.Integer),\n",
"('feature_dvr_total_rides', Types.Integer),\n",
"('feature_fixed_fare_amount', Types.Integer),\n",
"('feature_gh6_total_rides', Types.Integer),\n",
"('feature_has_waypoint', Types.Integer),\n",
"('feature_hour_local', Types.Integer),\n",
"('feature_hour_of_week_local', Types.Integer),\n",
"('feature_hour_of_week_shifted_local', Types.Integer),\n",
"('feature_hour_shifted_local', Types.Integer),\n",
"('feature_is_scheduled_ride', Types.Integer),\n",
"('feature_num_average_daily_rides_canceled', Types.Integer),\n",
"('feature_num_rides_taken', Types.Integer),\n",
"('feature_pax_avg_pickup_time_seconds', Types.Integer),\n",
"('feature_pax_no_show_rate', Types.Integer),\n",
"('feature_pax_num_voice_calls_to_driver', Types.Integer),\n",
"('feature_pax_sms', Types.Integer),\n",
"('feature_pax_sms_char_len', Types.Integer),\n",
"('feature_pax_sum_call_duration', Types.Integer),\n",
"('feature_pax_total_rides', Types.Integer),\n",
"('feature_pax_unsuccessful_voice', Types.Integer),\n",
"('feature_request_started_at_to_arrived_at_seconds', Types.Integer),\n",
"('feature_seconds_since_arrival', Types.Integer),\n",
"('feature_upfront_fare_amount', Types.Integer),\n",
"])\n",
"\n",
"schema = Types.Schema()\n",
"\n",
"presto = SdkPrestoTask(\n",
" task_inputs=inputs(start_date=Types.String, end_date=Types.String),\n",
" statement=query,\n",
" output_schema=schema,\n",
" catalog=\"hive\",\n",
" schema=\"default\",\n",
" discoverable=True,\n",
" discovery_version=\"1\",\n",
")"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"http://flyte.lyft.net/console/projects/flyteexamples/domains/development/executions/d42y9db6qz\n"
]
}
],
"source": [
"exc = presto.register_and_launch(\"flyteexamples\", \"development\", inputs={\"start_date\":\"2020-04-07\", \"end_date\":\"2020-04-01\"})\n",
"print_console_url(exc)"
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {},
"outputs": [],
"source": [
"exc.wait_for_completion()"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {},
"outputs": [],
"source": [
"exc.sync()\n",
"results = exc.outputs[\"results\"]\n",
"results.download(\"/tmp/data\", overwrite=True)\n",
"dfs = []\n",
"with results as reader:\n",
" for df in reader.iter_chunks():\n",
" dfs.append(df)"
]
},
{
"cell_type": "code",
"execution_count": 16,
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"[]"
]
},
"execution_count": 16,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"dfs"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {},
"outputs": [
{
"ename": "IndentationError",
"evalue": "unexpected indent (<ipython-input-5-3bbda51f0e3c>, line 3)",
"output_type": "error",
"traceback": [
"\u001b[0;36m File \u001b[0;32m\"<ipython-input-5-3bbda51f0e3c>\"\u001b[0;36m, line \u001b[0;32m3\u001b[0m\n\u001b[0;31m train_features = Index(['feature_driver_distance_at_arrival_meters',\u001b[0m\n\u001b[0m ^\u001b[0m\n\u001b[0;31mIndentationError\u001b[0m\u001b[0;31m:\u001b[0m unexpected indent\n"
]
}
],
"source": [
"from sklearn.model_selection import train_test_split\n",
"train_dataset, test_dataset = train_test_split(df, test_size=0.33, random_state=42)\n",
" train_features = Index(['feature_driver_distance_at_arrival_meters',\n",
" 'feature_driver_distance_at_cancellation_meters',\n",
" 'feature_dvr_cancellation_rate',\n",
" 'feature_dvr_no_show_rate',\n",
" 'feature_dvr_num_voice_calls_to_pax',\n",
" 'feature_dvr_rides_28d', \n",
" 'feature_dvr_sum_call_duration',\n",
" 'feature_dvr_total_rides',\n",
" 'feature_fixed_fare_amount',\n",
" 'feature_gh6_total_rides',\n",
" 'feature_has_waypoint',\n",
" 'feature_hour_local',\n",
" 'feature_hour_of_week_local',\n",
" 'feature_hour_of_week_shifted_local',\n",
" 'feature_hour_shifted_local',\n",
" 'feature_is_scheduled_ride',\n",
" 'feature_num_average_daily_rides_canceled',\n",
" 'feature_num_rides_taken',\n",
" 'feature_pax_avg_pickup_time_seconds',\n",
" 'feature_pax_no_show_rate',\n",
" 'feature_pax_num_voice_calls_to_driver',\n",
" 'feature_pax_sms',\n",
" 'feature_pax_sms_char_len',\n",
" 'feature_pax_sum_call_duration',\n",
" 'feature_pax_total_rides',\n",
" 'feature_pax_unsuccessful_voice',\n",
" 'feature_request_started_at_to_arrived_at_seconds',\n",
" 'feature_seconds_since_arrival',\n",
" 'feature_upfront_fare_amount'], dtype='object')\n",
" \n",
"labels = Index(['should_waive_fee'])\n",
"\n",
"x_train = train_dataset[train_features]\n",
"y_train = train_dataset[labels]\n",
"\n",
"x_test = test_dataset[train_features]\n",
"y_test = test_dataset[labels]"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"from flytekit.sdk.workflow import workflow_class, Input, Output"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3.7.4 64-bit ('flytekit': virtualenv)",
"language": "python",
"name": "python37464bitflytekitvirtualenv72cbb5e9968e4a299c6026c09cce8d4c"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.7.4"
}
},
"nbformat": 4,
"nbformat_minor": 4
}
Loading