FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
feast/sdk/python/feast/dbt/mapper.py at master · feast-dev/feast · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
feast-dev
/
feast
Public
Notifications
You must be signed in to change notification settings
Fork
1.4k
Star
7.2k
Code
Issues
218
Pull requests
191
Discussions
Actions
Security and quality
1
Insights
Additional navigation options
Code
Issues
Pull requests
Discussions
Actions
Security and quality
Insights
Expand file tree
Breadcrumbs
feast
/
sdk
/
python
/
feast
/
dbt
/
mapper.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
480 lines (417 loc) · 15.3 KB
Breadcrumbs
feast
/
sdk
/
python
/
feast
/
dbt
/
mapper.py
Copy path
File metadata and controls
480 lines (417 loc) · 15.3 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
"""
dbt to Feast type and object mapper.
This module provides functionality to map dbt model metadata to Feast objects
including DataSource, Entity, and FeatureView.
"""
from
datetime
import
timedelta
from
typing
import
Any
,
Dict
,
List
,
Optional
,
Union
from
feast
.
dbt
.
parser
import
DbtModel
from
feast
.
entity
import
Entity
from
feast
.
feature_view
import
FeatureView
from
feast
.
field
import
Field
from
feast
.
types
import
(
Array
,
Bool
,
Bytes
,
FeastType
,
Float32
,
Float64
,
Int32
,
Int64
,
String
,
UnixTimestamp
,
)
from
feast
.
value_type
import
ValueType
# Mapping from FeastType to ValueType for entity value inference
FEAST_TYPE_TO_VALUE_TYPE
:
Dict
[
FeastType
,
ValueType
]
=
{
String
:
ValueType
.
STRING
,
Int32
:
ValueType
.
INT32
,
Int64
:
ValueType
.
INT64
,
Float32
:
ValueType
.
FLOAT
,
Float64
:
ValueType
.
DOUBLE
,
Bool
:
ValueType
.
BOOL
,
Bytes
:
ValueType
.
BYTES
,
UnixTimestamp
:
ValueType
.
UNIX_TIMESTAMP
,
}
def
feast_type_to_value_type
(
feast_type
:
FeastType
)
->
ValueType
:
"""Convert a FeastType to its corresponding ValueType for entities."""
return
FEAST_TYPE_TO_VALUE_TYPE
.
get
(
feast_type
,
ValueType
.
STRING
)
# Comprehensive mapping from dbt/warehouse types to Feast types
# Covers BigQuery, Snowflake, Redshift, PostgreSQL, and common SQL types
DBT_TO_FEAST_TYPE_MAP
:
Dict
[
str
,
FeastType
]
=
{
# String types
"STRING"
:
String
,
"TEXT"
:
String
,
"VARCHAR"
:
String
,
"CHAR"
:
String
,
"CHARACTER"
:
String
,
"NVARCHAR"
:
String
,
"NCHAR"
:
String
,
"CHARACTER VARYING"
:
String
,
# Integer types
"INT"
:
Int64
,
"INT32"
:
Int32
,
"INT64"
:
Int64
,
"INTEGER"
:
Int64
,
"BIGINT"
:
Int64
,
"SMALLINT"
:
Int32
,
"TINYINT"
:
Int32
,
"BYTEINT"
:
Int32
,
"NUMBER"
:
Int64
,
# Snowflake - default to Int64, precision handling below
"NUMERIC"
:
Int64
,
"DECIMAL"
:
Int64
,
# Float types
"FLOAT"
:
Float32
,
"FLOAT32"
:
Float32
,
"FLOAT64"
:
Float64
,
"DOUBLE"
:
Float64
,
"DOUBLE PRECISION"
:
Float64
,
"REAL"
:
Float32
,
# Boolean types
"BOOL"
:
Bool
,
"BOOLEAN"
:
Bool
,
# Timestamp types
"TIMESTAMP"
:
UnixTimestamp
,
"TIMESTAMP_NTZ"
:
UnixTimestamp
,
"TIMESTAMP_LTZ"
:
UnixTimestamp
,
"TIMESTAMP_TZ"
:
UnixTimestamp
,
"DATETIME"
:
UnixTimestamp
,
"DATE"
:
UnixTimestamp
,
"TIME"
:
UnixTimestamp
,
# Binary types
"BYTES"
:
Bytes
,
"BINARY"
:
Bytes
,
"VARBINARY"
:
Bytes
,
"BLOB"
:
Bytes
,
}
def
map_dbt_type_to_feast_type
(
dbt_type
:
str
)
->
FeastType
:
"""
Map a dbt data type to a Feast type.
Handles various database type formats including:
- Simple types: STRING, INT64, FLOAT
- Parameterized types: VARCHAR(255), NUMBER(10,2), DECIMAL(18,0)
- Array types: ARRAY<STRING>, ARRAY<INT64>
Args:
dbt_type: The dbt/database data type string
Returns:
The corresponding Feast type
"""
if
not
dbt_type
:
return
String
# Normalize the type string
normalized
=
dbt_type
.
upper
().
strip
()
# Handle ARRAY types: ARRAY<element_type>
if
normalized
.
startswith
(
"ARRAY<"
)
and
normalized
.
endswith
(
">"
):
element_type_str
=
normalized
[
6
:
-
1
].
strip
()
element_type
=
map_dbt_type_to_feast_type
(
element_type_str
)
# Array only supports primitive types
valid_array_types
=
{
String
,
Int32
,
Int64
,
Float32
,
Float64
,
Bool
,
Bytes
,
UnixTimestamp
,
}
if
element_type
in
valid_array_types
:
return
Array
(
element_type
)
return
Array
(
String
)
# Fallback for complex nested types
# Handle parameterized types: VARCHAR(255), NUMBER(10,2), etc.
# Extract base type by removing parentheses and parameters
base_type
=
normalized
.
split
(
"("
)[
0
].
strip
()
# Handle Snowflake NUMBER with precision
if
base_type
==
"NUMBER"
and
"("
in
normalized
:
try
:
# Parse precision and scale: NUMBER(precision, scale)
params
=
normalized
.
split
(
"("
)[
1
].
rstrip
(
")"
).
split
(
","
)
precision
=
int
(
params
[
0
].
strip
())
scale
=
int
(
params
[
1
].
strip
())
if
len
(
params
)
>
1
else
0
if
scale
>
0
:
# Has decimal places, use Float64
return
Float64
elif
precision
<=
9
:
return
Int32
elif
precision
<=
18
:
return
Int64
else
:
# Precision > 18, may exceed Int64 range
return
Float64
except
(
ValueError
,
IndexError
):
return
Int64
# Look up in mapping table
if
base_type
in
DBT_TO_FEAST_TYPE_MAP
:
return
DBT_TO_FEAST_TYPE_MAP
[
base_type
]
# Default to String for unknown types
return
String
class
DbtToFeastMapper
:
"""
Maps dbt models to Feast objects.
Supports creating DataSource, Entity, and FeatureView objects from
dbt model metadata.
Example::
mapper = DbtToFeastMapper(data_source_type="bigquery")
data_source = mapper.create_data_source(model)
feature_view = mapper.create_feature_view(
model, data_source, entity_column="driver_id"
)
Args:
data_source_type: Type of data source ('bigquery', 'snowflake', 'file')
timestamp_field: Default timestamp field name
ttl_days: Default TTL in days for feature views
"""
def
__init__
(
self
,
data_source_type
:
str
=
"bigquery"
,
timestamp_field
:
str
=
"event_timestamp"
,
ttl_days
:
int
=
1
,
):
self
.
data_source_type
=
data_source_type
.
lower
()
self
.
timestamp_field
=
timestamp_field
self
.
ttl_days
=
ttl_days
def
_infer_entity_value_type
(
self
,
model
:
DbtModel
,
entity_col
:
str
)
->
ValueType
:
"""Infer entity ValueType from dbt model column type."""
for
column
in
model
.
columns
:
if
column
.
name
==
entity_col
:
feast_type
=
map_dbt_type_to_feast_type
(
column
.
data_type
)
return
feast_type_to_value_type
(
feast_type
)
return
ValueType
.
UNKNOWN
def
create_data_source
(
self
,
model
:
DbtModel
,
timestamp_field
:
Optional
[
str
]
=
None
,
created_timestamp_column
:
Optional
[
str
]
=
None
,
)
->
Any
:
"""
Create a Feast DataSource from a dbt model.
Args:
model: The DbtModel to create a DataSource from
timestamp_field: Override the default timestamp field
created_timestamp_column: Column for created timestamp (dedup)
Returns:
A Feast DataSource (BigQuerySource, SnowflakeSource, or FileSource)
Raises:
ValueError: If data_source_type is not supported
"""
ts_field
=
timestamp_field
or
self
.
timestamp_field
# Build tags from dbt metadata
tags
=
{
"dbt.model"
:
model
.
name
}
for
tag
in
model
.
tags
:
tags
[
f"dbt.tag.
{
tag
}
"
]
=
"true"
if
self
.
data_source_type
==
"bigquery"
:
from
feast
.
infra
.
offline_stores
.
bigquery_source
import
BigQuerySource
return
BigQuerySource
(
name
=
f"
{
model
.
name
}
_source"
,
table
=
model
.
full_table_name
,
timestamp_field
=
ts_field
,
created_timestamp_column
=
created_timestamp_column
or
""
,
description
=
model
.
description
,
tags
=
tags
,
)
elif
self
.
data_source_type
==
"snowflake"
:
from
feast
.
infra
.
offline_stores
.
snowflake_source
import
SnowflakeSource
return
SnowflakeSource
(
name
=
f"
{
model
.
name
}
_source"
,
database
=
model
.
database
,
schema
=
model
.
schema
,
table
=
model
.
alias
,
timestamp_field
=
ts_field
,
created_timestamp_column
=
created_timestamp_column
or
""
,
description
=
model
.
description
,
tags
=
tags
,
)
elif
self
.
data_source_type
==
"file"
:
from
feast
.
infra
.
offline_stores
.
file_source
import
FileSource
# For file sources, use the model name as a placeholder path
return
FileSource
(
name
=
f"
{
model
.
name
}
_source"
,
path
=
f"/data/
{
model
.
name
}
.parquet"
,
timestamp_field
=
ts_field
,
created_timestamp_column
=
created_timestamp_column
or
""
,
description
=
model
.
description
,
tags
=
tags
,
)
else
:
raise
ValueError
(
f"Unsupported data_source_type:
{
self
.
data_source_type
}
. "
f"Supported types: bigquery, snowflake, file"
)
def
create_entity
(
self
,
name
:
str
,
join_keys
:
Optional
[
List
[
str
]]
=
None
,
description
:
str
=
""
,
tags
:
Optional
[
Dict
[
str
,
str
]]
=
None
,
value_type
:
ValueType
=
ValueType
.
STRING
,
)
->
Entity
:
"""
Create a Feast Entity.
Args:
name: Entity name
join_keys: List of join key column names (defaults to [name])
description: Entity description
tags: Optional tags
value_type: Value type for the entity (default: STRING)
Returns:
A Feast Entity
"""
return
Entity
(
name
=
name
,
join_keys
=
join_keys
or
[
name
],
value_type
=
value_type
,
description
=
description
,
tags
=
tags
or
{},
)
def
create_feature_view
(
self
,
model
:
DbtModel
,
source
:
Any
,
entity_columns
:
Union
[
str
,
List
[
str
]],
entities
:
Optional
[
Union
[
Entity
,
List
[
Entity
]]]
=
None
,
timestamp_field
:
Optional
[
str
]
=
None
,
ttl_days
:
Optional
[
int
]
=
None
,
exclude_columns
:
Optional
[
List
[
str
]]
=
None
,
online
:
bool
=
True
,
)
->
FeatureView
:
"""
Create a Feast FeatureView from a dbt model.
Args:
model: The DbtModel to create a FeatureView from
source: The DataSource for this FeatureView
entity_columns: Entity column name(s) - single string or list of strings
entities: Optional pre-created Entity or list of Entities
timestamp_field: Override the default timestamp field
ttl_days: Override the default TTL in days
exclude_columns: Additional columns to exclude from features
online: Whether to enable online serving
Returns:
A Feast FeatureView
"""
# Normalize to lists
entity_cols
:
List
[
str
]
=
(
[
entity_columns
]
if
isinstance
(
entity_columns
,
str
)
else
list
(
entity_columns
)
)
entity_objs
:
List
[
Entity
]
=
[]
if
entities
is
not
None
:
entity_objs
=
[
entities
]
if
isinstance
(
entities
,
Entity
)
else
list
(
entities
)
# Validate
if
not
entity_cols
:
raise
ValueError
(
"At least one entity column must be specified"
)
if
entity_objs
and
len
(
entity_cols
)
!=
len
(
entity_objs
):
raise
ValueError
(
f"Number of entity_columns (
{
len
(
entity_cols
)
}
) must match "
f"number of entities (
{
len
(
entity_objs
)
}
)"
)
ts_field
=
timestamp_field
or
self
.
timestamp_field
ttl
=
timedelta
(
days
=
ttl_days
if
ttl_days
is
not
None
else
self
.
ttl_days
)
# Columns to exclude from schema (timestamp + any explicitly excluded)
# Note: entity columns should NOT be excluded - FeatureView.__init__
# expects entity columns to be in the schema and will extract them
excluded
=
{
ts_field
}
if
exclude_columns
:
excluded
.
update
(
exclude_columns
)
# Create schema from model columns (includes entity columns)
schema
:
List
[
Field
]
=
[]
for
column
in
model
.
columns
:
if
column
.
name
not
in
excluded
:
feast_type
=
map_dbt_type_to_feast_type
(
column
.
data_type
)
schema
.
append
(
Field
(
name
=
column
.
name
,
dtype
=
feast_type
,
description
=
column
.
description
,
)
)
# Create entities if not provided
if
not
entity_objs
:
entity_objs
=
[]
for
entity_col
in
entity_cols
:
# Infer entity value type from model column
entity_value_type
=
self
.
_infer_entity_value_type
(
model
,
entity_col
)
ent
=
self
.
create_entity
(
name
=
entity_col
,
description
=
f"Entity for
{
model
.
name
}
"
,
value_type
=
entity_value_type
,
)
entity_objs
.
append
(
ent
)
# Build tags from dbt metadata
tags
=
{
"dbt.model"
:
model
.
name
,
"dbt.unique_id"
:
model
.
unique_id
,
}
for
tag
in
model
.
tags
:
tags
[
f"dbt.tag.
{
tag
}
"
]
=
"true"
return
FeatureView
(
name
=
model
.
name
,
source
=
source
,
schema
=
schema
,
entities
=
entity_objs
,
ttl
=
ttl
,
online
=
online
,
description
=
model
.
description
,
tags
=
tags
,
)
def
create_all_from_model
(
self
,
model
:
DbtModel
,
entity_columns
:
Union
[
str
,
List
[
str
]],
timestamp_field
:
Optional
[
str
]
=
None
,
ttl_days
:
Optional
[
int
]
=
None
,
exclude_columns
:
Optional
[
List
[
str
]]
=
None
,
online
:
bool
=
True
,
)
->
Dict
[
str
,
Union
[
List
[
Entity
],
Any
,
FeatureView
]]:
"""
Create all Feast objects (DataSource, Entity, FeatureView) from a dbt model.
This is a convenience method that creates all necessary Feast objects
in one call.
Args:
model: The DbtModel to create objects from
entity_columns: Entity column name(s) - single string or list of strings
timestamp_field: Override the default timestamp field
ttl_days: Override the default TTL in days
exclude_columns: Additional columns to exclude from features
online: Whether to enable online serving
Returns:
Dict with keys 'entities', 'data_source', 'feature_view'
"""
# Normalize to list
entity_cols
:
List
[
str
]
=
(
[
entity_columns
]
if
isinstance
(
entity_columns
,
str
)
else
list
(
entity_columns
)
)
# Create entities (plural)
entities_list
=
[]
for
entity_col
in
entity_cols
:
entity_value_type
=
self
.
_infer_entity_value_type
(
model
,
entity_col
)
entity
=
self
.
create_entity
(
name
=
entity_col
,
description
=
f"Entity for
{
model
.
name
}
"
,
tags
=
{
"dbt.model"
:
model
.
name
},
value_type
=
entity_value_type
,
)
entities_list
.
append
(
entity
)
# Create data source
data_source
=
self
.
create_data_source
(
model
=
model
,
timestamp_field
=
timestamp_field
,
)
# Create feature view
feature_view
=
self
.
create_feature_view
(
model
=
model
,
source
=
data_source
,
entity_columns
=
entity_cols
,
entities
=
entities_list
,
timestamp_field
=
timestamp_field
,
ttl_days
=
ttl_days
,
exclude_columns
=
exclude_columns
,
online
=
online
,
)
return
{
"entities"
:
entities_list
,
"data_source"
:
data_source
,
"feature_view"
:
feature_view
,
}
Back
|
FazBrowse Home
|
New Git URL