This repository has no description
add experiment_duckdb.py
+124
-4
+1
Pipfile
+1
Pipfile
+32
-1
Pipfile.lock
+32
-1
Pipfile.lock
···
1
1
{
2
2
"_meta": {
3
3
"hash": {
4
-
"sha256": "c8a61b88d883b430ecd6ff769383b207307f14afbf8aea68aa4da7396efd3aa5"
4
+
"sha256": "d626bf77b7fff9a73f1f188eccbf8b212341cb065716fd380b3f05baf5464249"
5
5
},
6
6
"pipfile-spec": 6,
7
7
"requires": {
···
431
431
"sha256:52d171a6a2b031d8a5d1de6efe451cf4f5baff1a2819aabc3741c8406539ba04"
432
432
],
433
433
"version": "==0.10.9.5"
434
+
},
435
+
"pyarrow": {
436
+
"hashes": [
437
+
"sha256:1cbcfcbb0e74b4d94f0b7dde447b835a01bc1d16510edb8bb7d6224b9bf5bafc",
438
+
"sha256:25aa11c443b934078bfd60ed63e4e2d42461682b5ac10f67275ea21e60e6042c",
439
+
"sha256:2d53ba72917fdb71e3584ffc23ee4fcc487218f8ff29dd6df3a34c5c48fe8c06",
440
+
"sha256:2d942c690ff24a08b07cb3df818f542a90e4d359381fbff71b8f2aea5bf58841",
441
+
"sha256:2f51dc7ca940fdf17893227edb46b6784d37522ce08d21afc56466898cb213b2",
442
+
"sha256:362a7c881b32dc6b0eccf83411a97acba2774c10edcec715ccaab5ebf3bb0835",
443
+
"sha256:3e99be85973592051e46412accea31828da324531a060bd4585046a74ba45854",
444
+
"sha256:40bb42afa1053c35c749befbe72f6429b7b5f45710e85059cdd534553ebcf4f2",
445
+
"sha256:410624da0708c37e6a27eba321a72f29d277091c8f8d23f72c92bada4092eb5e",
446
+
"sha256:41a1451dd895c0b2964b83d91019e46f15b5564c7ecd5dcb812dadd3f05acc97",
447
+
"sha256:5461c57dbdb211a632a48facb9b39bbeb8a7905ec95d768078525283caef5f6d",
448
+
"sha256:69309be84dcc36422574d19c7d3a30a7ea43804f12552356d1ab2a82a713c418",
449
+
"sha256:7c28b5f248e08dea3b3e0c828b91945f431f4202f1a9fe84d1012a761324e1ba",
450
+
"sha256:8f40be0d7381112a398b93c45a7e69f60261e7b0269cc324e9f739ce272f4f70",
451
+
"sha256:a37bc81f6c9435da3c9c1e767324ac3064ffbe110c4e460660c43e144be4ed85",
452
+
"sha256:aaee8f79d2a120bf3e032d6d64ad20b3af6f56241b0ffc38d201aebfee879d00",
453
+
"sha256:ad42bb24fc44c48f74f0d8c72a9af16ba9a01a2ccda5739a517aa860fa7e3d56",
454
+
"sha256:ad7c53def8dbbc810282ad308cc46a523ec81e653e60a91c609c2233ae407689",
455
+
"sha256:becc2344be80e5dce4e1b80b7c650d2fc2061b9eb339045035a1baa34d5b8f1c",
456
+
"sha256:caad867121f182d0d3e1a0d36f197df604655d0b466f1bc9bafa903aa95083e4",
457
+
"sha256:ccbf29a0dadfcdd97632b4f7cca20a966bb552853ba254e874c66934931b9841",
458
+
"sha256:da93340fbf6f4e2a62815064383605b7ffa3e9eeb320ec839995b1660d69f89b",
459
+
"sha256:e217d001e6389b20a6759392a5ec49d670757af80101ee6b5f2c8ff0172e02ca",
460
+
"sha256:f010ce497ca1b0f17a8243df3048055c0d18dcadbcc70895d5baf8921f753de5",
461
+
"sha256:f12932e5a6feb5c58192209af1d2607d488cb1d404fbc038ac12ada60327fa34"
462
+
],
463
+
"index": "pypi",
464
+
"version": "==11.0.0"
434
465
},
435
466
"pyparsing": {
436
467
"hashes": [
+89
src/experiment_duckdb.py
+89
src/experiment_duckdb.py
···
1
+
import argparse
2
+
import json
3
+
import logging as log
4
+
import time
5
+
import uuid
6
+
7
+
import duckdb
8
+
import psutil
9
+
10
+
11
+
log.basicConfig(format="%(asctime)s - [%(levelname)s] %(message)s", level=log.INFO)
12
+
13
+
################## metadata ##################
14
+
RUN_ID = str(uuid.uuid4())
15
+
ENGINE = "duckdb"
16
+
MODE = "default"
17
+
18
+
log.info(f"start run - {ENGINE}-{MODE}: {RUN_ID}")
19
+
20
+
################## init ##################
21
+
### get input params ###
22
+
parser = argparse.ArgumentParser()
23
+
parser.add_argument("--trial_rows")
24
+
args = parser.parse_args()
25
+
trial_rows = int(args.trial_rows)
26
+
27
+
log.info(f"trial_rows: {trial_rows}")
28
+
29
+
start_time = time.time() # start timer
30
+
31
+
################## main ##################
32
+
path = "data/prep/nyc-trip-data/part-00003-f6a895f0-23e3-4aaa-a731-dff7e37be1a2-c000.snappy.parquet" # large
33
+
# path = 'data/prep/nyc-trip-data/part-00003-fa42da38-f364-4099-9725-a119a1f65de8-c000.snappy.parquet' # small
34
+
35
+
df = duckdb.sql(
36
+
f"""
37
+
SELECT VendorID,
38
+
payment_type,
39
+
--- passenger_count ---
40
+
min(passenger_count) AS min_passenger_count,
41
+
max(passenger_count) AS max_passenger_count,
42
+
avg(passenger_count) AS avg_passenger_count,
43
+
--- trip_distance ---
44
+
min(trip_distance) AS min_trip_distance,
45
+
max(trip_distance) AS max_trip_distance,
46
+
avg(trip_distance) AS avg_trip_distance,
47
+
--- total_amount ---
48
+
min(total_amount) AS min_total_amount,
49
+
max(total_amount) AS max_total_amount,
50
+
avg(total_amount) AS avg_total_amount
51
+
52
+
FROM (
53
+
SELECT 'dummy' AS partition,
54
+
--- filter ---
55
+
(epoch(tpep_dropoff_datetime)-epoch(tpep_pickup_datetime))/60 AS trip_length_minute,
56
+
percent_rank() OVER (PARTITION BY partition ORDER BY trip_length_minute) AS trip_length_minute_percentile,
57
+
--- groupby cols ---
58
+
VendorID,
59
+
payment_type,
60
+
--- agg cols ---
61
+
passenger_count,
62
+
trip_distance,
63
+
total_amount
64
+
FROM read_parquet('{path}')
65
+
)
66
+
WHERE trip_length_minute_percentile BETWEEN 0.2 AND 0.8
67
+
GROUP BY VendorID, payment_type
68
+
"""
69
+
).pl()
70
+
71
+
log.info(f"output rows: {len(df)}") # trigger actions
72
+
end_time = time.time() # end timer
73
+
74
+
elapsed_time = end_time - start_time
75
+
log.info(f"Elapsed time was {elapsed_time} seconds")
76
+
77
+
################## logging ##################
78
+
with open("data/runs.json", "a") as f:
79
+
r = {
80
+
"uuid": RUN_ID,
81
+
"engine": ENGINE,
82
+
"mode": MODE,
83
+
"processed_rows": trial_rows,
84
+
"duration": elapsed_time,
85
+
"swap_usage": psutil.swap_memory().total,
86
+
}
87
+
88
+
f.write(json.dumps(r))
89
+
f.write("\n")
+1
-2
src/experiment_polars.py
+1
-2
src/experiment_polars.py
···
87
87
)
88
88
)
89
89
90
-
log.info(f"output rows: {len(df_out.collect())}") # trigger spark actions
90
+
log.info(f"output rows: {len(df_out.collect())}") # trigger actions
91
91
end_time = time.time() # end timer
92
92
93
93
elapsed_time = end_time - start_time
94
94
log.info(f"Elapsed time was {elapsed_time} seconds")
95
95
96
96
################## logging ##################
97
-
98
97
with open("data/runs.json", "a") as f:
99
98
r = {
100
99
"uuid": RUN_ID,