Skip to content

Commit ae38473

Browse files
ntkatholentkathole
committed
feat: Added Iceberg REST Catalog data source support
Signed-off-by: ntkathole <[email protected]>
1 parent 14e0a83 commit ae38473

24 files changed

Lines changed: 2793 additions & 18 deletions

File tree

‎docs/reference/data-sources/README.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,10 @@ Please see [Data Source](../../getting-started/concepts/data-ingestion.md) for a
4242
[spark.md](spark.md)
4343
{% endcontent-ref %}
4444

45+
{% content-ref url="iceberg.md" %}
46+
[iceberg.md](iceberg.md)
47+
{% endcontent-ref %}
48+
4549
{% content-ref url="postgres.md" %}
4650
[postgres.md](postgres.md)
4751
{% endcontent-ref %}

‎docs/reference/data-sources/iceberg.md‎

Lines changed: 161 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
# Iceberg source (contrib)
2+
3+
## Description
4+
5+
Iceberg data sources are tables managed by any supported Iceberg catalog. The `IcebergSource` class provides a unified interface with a configurable `catalog_type` parameter:
6+
7+
- **`"rest"`** (default): [Apache Iceberg REST Catalog specification](https://iceberg.apache.org/concepts/catalog/#decoupling-using-the-rest-catalog) — Unity Catalog, Apache Polaris, Nessie, Snowflake Open Catalog
8+
- **`"hive"`**: Hive Metastore catalog
9+
- **`"glue"`**: AWS Glue catalog
10+
- **`"sql"`**: SQL-based (JDBC) catalog
11+
- **`"dynamodb"`**: DynamoDB-based catalog
12+
13+
The data source carries catalog connection details (catalog_type, endpoint, warehouse, namespace, table, authentication). When the offline store (DuckDB, Spark) encounters this source, it resolves table metadata and credentials via the configured catalog at query time.
14+
15+
## Examples
16+
17+
### IcebergSource (REST catalog)
18+
19+
Works with any Iceberg REST Catalog:
20+
21+
```python
22+
from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
23+
24+
my_source = IcebergSource(
25+
catalog_type="rest", # default
26+
endpoint="http://localhost:8081/api/2.1/unity-catalog/iceberg",
27+
warehouse="unity",
28+
namespace="default",
29+
table="driver_features",
30+
timestamp_field="event_timestamp",
31+
token_env_var="UC_TOKEN",
32+
)
33+
```
34+
35+
### IcebergSource (Hive Metastore)
36+
37+
```python
38+
from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
39+
40+
my_source = IcebergSource(
41+
catalog_type="hive",
42+
catalog_properties={"uri": "thrift://metastore:9083"},
43+
warehouse="my_warehouse",
44+
namespace="default",
45+
table="driver_features",
46+
timestamp_field="event_timestamp",
47+
)
48+
```
49+
50+
### IcebergSource (AWS Glue)
51+
52+
```python
53+
from feast.infra.data_sources.contrib.iceberg_catalog import IcebergSource
54+
55+
my_source = IcebergSource(
56+
catalog_type="glue",
57+
catalog_properties={"region_name": "us-east-1"},
58+
warehouse="my_account",
59+
namespace="my_database",
60+
table="driver_features",
61+
timestamp_field="event_timestamp",
62+
)
63+
```
64+
65+
### UnityCatalogSource (with governance) {#unity-catalog-source}
66+
67+
Extends `IcebergSource` with Unity Catalog governance:
68+
69+
```python
70+
from feast.infra.data_sources.contrib.iceberg_catalog import (
71+
UnityCatalogSource,
72+
)
73+
74+
my_uc_source = UnityCatalogSource(
75+
warehouse="production",
76+
namespace="ml_features",
77+
table="driver_stats",
78+
timestamp_field="event_timestamp",
79+
register_as_feature_table=True, # Register in UC on feast apply
80+
sync_lineage=True, # Record lineage in UC
81+
)
82+
```
83+
84+
When `endpoint` is omitted, it defaults to `{DATABRICKS_HOST}/api/2.1/unity-catalog/iceberg`.
85+
When `token_env_var` is omitted, it defaults to `DATABRICKS_TOKEN`.
86+
87+
### Full Feature View Example
88+
89+
```python
90+
from datetime import timedelta
91+
92+
from feast import Entity, FeatureView, Field
93+
from feast.types import Float64, Int64
94+
95+
from feast.infra.data_sources.contrib.iceberg_catalog import (
96+
UnityCatalogSource,
97+
)
98+
99+
driver = Entity(name="driver_id", join_keys=["driver_id"])
100+
101+
driver_stats_source = UnityCatalogSource(
102+
warehouse="production",
103+
namespace="ml_features",
104+
table="driver_hourly_stats",
105+
timestamp_field="event_timestamp",
106+
created_timestamp_column="created",
107+
)
108+
109+
driver_stats_fv = FeatureView(
110+
name="driver_hourly_stats",
111+
entities=[driver],
112+
source=driver_stats_source,
113+
schema=[
114+
Field(name="conv_rate", dtype=Float64),
115+
Field(name="acc_rate", dtype=Float64),
116+
Field(name="avg_daily_trips", dtype=Int64),
117+
],
118+
ttl=timedelta(days=1),
119+
online=True,
120+
)
121+
```
122+
123+
## Configuration Reference
124+
125+
### IcebergSource
126+
127+
| Parameter | Type | Description |
128+
| :--- | :--- | :--- |
129+
| `catalog_type` | `str` | Catalog backend: `"rest"` (default), `"hive"`, `"glue"`, `"sql"`, `"dynamodb"` |
130+
| `endpoint` | `str` | Catalog endpoint URL (required for `"rest"`, optional for others) |
131+
| `warehouse` | `str` | Catalog/warehouse name |
132+
| `namespace` | `str` | Schema/namespace within the catalog |
133+
| `table` | `str` | Table name |
134+
| `catalog_properties` | `dict` | Additional catalog-specific properties passed to PyIceberg |
135+
| `timestamp_field` | `str` | Event timestamp column for point-in-time joins |
136+
| `created_timestamp_column` | `str` | Optional column indicating row creation time |
137+
| `token_env_var` | `str` | Environment variable name holding the auth token |
138+
| `credential_vending` | `bool` | Whether to request scoped credentials (default: `True`) |
139+
| `field_mapping` | `dict` | Column name mapping from source to feature names |
140+
141+
### UnityCatalogSource (additional parameters)
142+
143+
| Parameter | Type | Description |
144+
| :--- | :--- | :--- |
145+
| `register_as_feature_table` | `bool` | Register as UC feature table on `feast apply` (default: `True`) |
146+
| `sync_lineage` | `bool` | Sync lineage metadata to Unity Catalog (default: `True`) |
147+
148+
## Supported Types
149+
150+
| Iceberg Type | Feast Type |
151+
| :--- | :--- |
152+
| `boolean` | `BOOL` |
153+
| `int` | `INT32` |
154+
| `long` | `INT64` |
155+
| `float` | `FLOAT` |
156+
| `double` | `DOUBLE` |
157+
| `string` | `STRING` |
158+
| `binary` | `BYTES` |
159+
| `timestamp` / `timestamptz` | `INT64` |
160+
| `decimal` | `DOUBLE` |
161+
| `uuid` | `STRING` |

‎protos/feast/core/DataSource.proto‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,7 @@ message DataSource {
3636
reserved 6 to 10;
3737

3838
// Type of Data Source.
39-
// Next available id: 13
39+
// Next available id: 14
4040
enum SourceType {
4141
INVALID = 0;
4242
BATCH_FILE = 1;
@@ -51,6 +51,7 @@ message DataSource {
5151
BATCH_TRINO = 10;
5252
BATCH_SPARK = 11;
5353
BATCH_ATHENA = 12;
54+
BATCH_ICEBERG = 13;
5455
}
5556

5657
// Unique name of data source within the project

‎pyproject.toml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -112,6 +112,7 @@ mysql = ["pymysql", "types-PyMySQL"]
112112
openlineage = ["openlineage-python>=1.40.0"]
113113
opentelemetry = ["prometheus_client", "psutil"]
114114
spark = ["pyspark>=4.0.0"]
115+
iceberg = ["pyiceberg>=0.7.0"]
115116
trino = ["trino>=0.305.0,<0.400.0", "regex"]
116117
postgres = ["psycopg[binary,pool]>=3.2.5"]
117118
# psycopg[c] install requires a system with a C compiler, python dev headers, & postgresql client dev headers

‎sdk/python/feast/data_source.py‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -157,6 +157,7 @@ def to_proto(self) -> DataSourceProto.KinesisOptions:
157157
DataSourceProto.SourceType.BATCH_TRINO: "feast.infra.offline_stores.contrib.trino_offline_store.trino_source.TrinoSource",
158158
DataSourceProto.SourceType.BATCH_SPARK: "feast.infra.offline_stores.contrib.spark_offline_store.spark_source.SparkSource",
159159
DataSourceProto.SourceType.BATCH_ATHENA: "feast.infra.offline_stores.contrib.athena_offline_store.athena_source.AthenaSource",
160+
DataSourceProto.SourceType.BATCH_ICEBERG: "feast.infra.data_sources.contrib.iceberg_catalog.iceberg_source.IcebergSource",
160161
DataSourceProto.SourceType.STREAM_KAFKA: "feast.data_source.KafkaSource",
161162
DataSourceProto.SourceType.STREAM_KINESIS: "feast.data_source.KinesisSource",
162163
DataSourceProto.SourceType.REQUEST_SOURCE: "feast.data_source.RequestSource",

‎sdk/python/feast/infra/data_sources/__init__.py‎

Whitespace-only changes.

‎sdk/python/feast/infra/data_sources/contrib/__init__.py‎

Whitespace-only changes.

‎sdk/python/feast/infra/data_sources/contrib/iceberg_catalog/__init__.py‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
from feast.infra.data_sources.contrib.iceberg_catalog.iceberg_rest_client import (
2+
CatalogAuthError,
3+
IcebergCatalogError,
4+
IcebergRestClient,
5+
TableMetadata,
6+
TableNotFoundError,
7+
TempCredential,
8+
)
9+
from feast.infra.data_sources.contrib.iceberg_catalog.iceberg_source import (
10+
IcebergSource,
11+
)
12+
from feast.infra.data_sources.contrib.iceberg_catalog.uc_client import UCClient
13+
from feast.infra.data_sources.contrib.iceberg_catalog.unity_catalog_source import (
14+
UnityCatalogSource,
15+
)
16+
17+
__all__ = [
18+
"CatalogAuthError",
19+
"IcebergCatalogError",
20+
"IcebergRestClient",
21+
"IcebergSource",
22+
"TableMetadata",
23+
"TableNotFoundError",
24+
"TempCredential",
25+
"UCClient",
26+
"UnityCatalogSource",
27+
]

0 commit comments

Comments
 (0)