Repository navigation
Expand file tree
/
Copy pathsilver.py
More file actions
96 lines (81 loc) · 3.31 KB
/
Copy pathsilver.py
File metadata and controls
96 lines (81 loc) · 3.31 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
from pyspark import pipelines as dp
from pyspark.sql.functions import *
from pyspark.sql.types import *
rides_schema = StructType([
StructField('ride_id', StringType(), True),
StructField('confirmation_number', StringType(), True),
StructField('passenger_id', StringType(), True),
StructField('driver_id', StringType(), True),
StructField('vehicle_id', StringType(), True),
StructField('pickup_location_id', StringType(), True),
StructField('dropoff_location_id', StringType(), True),
StructField('vehicle_type_id', LongType(), True),
StructField('vehicle_make_id', LongType(), True),
StructField('payment_method_id', LongType(), True),
StructField('ride_status_id', LongType(), True),
StructField('pickup_city_id', LongType(), True),
StructField('dropoff_city_id', LongType(), True),
StructField('cancellation_reason_id', LongType(), True),
StructField('passenger_name', StringType(), True),
StructField('passenger_email', StringType(), True),
StructField('passenger_phone', StringType(), True),
StructField('driver_name', StringType(), True),
StructField('driver_rating', DoubleType(), True),
StructField('driver_phone', StringType(), True),
StructField('driver_license', StringType(), True),
StructField('vehicle_model', StringType(), True),
StructField('vehicle_color', StringType(), True),
StructField('license_plate', StringType(), True),
StructField('pickup_address', StringType(), True),
StructField('pickup_latitude', DoubleType(), True),
StructField('pickup_longitude', DoubleType(), True),
StructField('dropoff_address', StringType(), True),
StructField('dropoff_latitude', DoubleType(), True),
StructField('dropoff_longitude', DoubleType(), True),
StructField('distance_miles', DoubleType(), True),
StructField('duration_minutes', LongType(), True),
StructField('booking_timestamp', TimestampType(), True),
StructField('pickup_timestamp', StringType(), True),
StructField('dropoff_timestamp', StringType(), True),
StructField('base_fare', DoubleType(), True),
StructField('distance_fare', DoubleType(), True),
StructField('time_fare', DoubleType(), True),
StructField('surge_multiplier', DoubleType(), True),
StructField('subtotal', DoubleType(), True),
StructField('tip_amount', DoubleType(), True),
StructField('total_fare', DoubleType(), True),
StructField('rating', DoubleType(), True)
])
# Empty target streaming table
dp.create_streaming_table("stg_rides")
# -------------------------------------------------
# BULK / INITIAL LOAD — RUN ONCE
# -------------------------------------------------
@dp.append_flow(
target="stg_rides",
once=True
)
def rides_bulk():
df = spark.read.table("bulk_rides")
df = df.withColumn(
"booking_timestamp",
to_timestamp(col("booking_timestamp"))
)
return df
# -------------------------------------------------
# REAL-TIME STREAMING LOAD — CONTINUOUS
# -------------------------------------------------
@dp.append_flow(
target="stg_rides"
)
def rides_stream():
df = spark.readStream.table("rides_raw")
df_parsed = (
df
.withColumn(
"parsed_rides",
from_json(col("rides"), rides_schema)
)
.select("parsed_rides.*")
)
return df_parsed