Skip to content

Online Store API

The online module provides abstract base classes and implementations for real-time feature serving.

Classes

OnlineStore (Abstract Base Class)

mlforge.online.OnlineStore

Bases: ABC

Abstract base class for online feature storage backends.

Online stores provide low-latency read/write access to feature values for real-time ML inference. Unlike offline stores (which store full feature history), online stores typically only hold the latest values.

Key design
  • Simple key-value model: entity keys -> feature values
  • JSON serialization for human-readable debugging
  • Batch operations for efficient bulk access
Source code in src/mlforge/online.py 
 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
class OnlineStore(ABC):
    """
    Abstract base class for online feature storage backends.

    Online stores provide low-latency read/write access to feature values
    for real-time ML inference. Unlike offline stores (which store full
    feature history), online stores typically only hold the latest values.

    Key design:
        - Simple key-value model: entity keys -> feature values
        - JSON serialization for human-readable debugging
        - Batch operations for efficient bulk access
    """

    @abstractmethod
    def write(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
        values: dict[str, Any],
    ) -> None:
        """
        Write feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values (e.g., {"user_id": "123"})
            values: Feature column values (e.g., {"amount__sum__7d": 1500.0})
        """
        ...

    @abstractmethod
    def write_batch(
        self,
        feature_name: str,
        records: list[dict[str, Any]],
        entity_key_columns: list[str],
    ) -> int:
        """
        Write feature values for multiple entities.

        Args:
            feature_name: Name of the feature
            records: List of records, each containing entity keys and feature values
            entity_key_columns: Column names that form the entity key

        Returns:
            Number of records written
        """
        ...

    @abstractmethod
    def read(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> dict[str, Any] | None:
        """
        Read feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            Feature values dict, or None if not found
        """
        ...

    @abstractmethod
    def read_batch(
        self,
        feature_name: str,
        entity_keys: list[dict[str, str]],
    ) -> list[dict[str, Any] | None]:
        """
        Read feature values for multiple entities.

        Args:
            feature_name: Name of the feature
            entity_keys: List of entity key dicts

        Returns:
            List of feature value dicts (None for missing entities)
        """
        ...

    @abstractmethod
    def delete(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> bool:
        """
        Delete feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            True if deleted, False if not found
        """
        ...

    @abstractmethod
    def exists(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> bool:
        """
        Check if feature values exist for an entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            True if exists, False otherwise
        """
        ...

delete abstractmethod 

delete(
    feature_name: str, entity_keys: dict[str, str]
) -> bool

Delete feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
bool

True if deleted, False if not found
Source code in src/mlforge/online.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@abstractmethod
def delete(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> bool:
    """
    Delete feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        True if deleted, False if not found
    """
    ...

exists abstractmethod 

exists(
    feature_name: str, entity_keys: dict[str, str]
) -> bool

Check if feature values exist for an entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
bool

True if exists, False otherwise
Source code in src/mlforge/online.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
@abstractmethod
def exists(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> bool:
    """
    Check if feature values exist for an entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        True if exists, False otherwise
    """
    ...

read abstractmethod 

read(
    feature_name: str, entity_keys: dict[str, str]
) -> dict[str, Any] | None

Read feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
dict[str, Any] | None

Feature values dict, or None if not found
Source code in src/mlforge/online.py 
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@abstractmethod
def read(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> dict[str, Any] | None:
    """
    Read feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        Feature values dict, or None if not found
    """
    ...

read_batch abstractmethod 

read_batch(
    feature_name: str, entity_keys: list[dict[str, str]]
) -> list[dict[str, Any] | None]

Read feature values for multiple entities.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys list[dict[str, str]]

List of entity key dicts

 required

Returns:

Type Description
list[dict[str, Any] | None]

List of feature value dicts (None for missing entities)
Source code in src/mlforge/online.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@abstractmethod
def read_batch(
    self,
    feature_name: str,
    entity_keys: list[dict[str, str]],
) -> list[dict[str, Any] | None]:
    """
    Read feature values for multiple entities.

    Args:
        feature_name: Name of the feature
        entity_keys: List of entity key dicts

    Returns:
        List of feature value dicts (None for missing entities)
    """
    ...

write abstractmethod 

write(
    feature_name: str,
    entity_keys: dict[str, str],
    values: dict[str, Any],
) -> None

Write feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values (e.g., {"user_id": "123"})

 required
values dict[str, Any]

Feature column values (e.g., {"amount__sum__7d": 1500.0})

 required
Source code in src/mlforge/online.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@abstractmethod
def write(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
    values: dict[str, Any],
) -> None:
    """
    Write feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values (e.g., {"user_id": "123"})
        values: Feature column values (e.g., {"amount__sum__7d": 1500.0})
    """
    ...

write_batch abstractmethod 

write_batch(
    feature_name: str,
    records: list[dict[str, Any]],
    entity_key_columns: list[str],
) -> int

Write feature values for multiple entities.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
records list[dict[str, Any]]

List of records, each containing entity keys and feature values

 required
entity_key_columns list[str]

Column names that form the entity key

 required

Returns:

Type Description
int

Number of records written
Source code in src/mlforge/online.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
@abstractmethod
def write_batch(
    self,
    feature_name: str,
    records: list[dict[str, Any]],
    entity_key_columns: list[str],
) -> int:
    """
    Write feature values for multiple entities.

    Args:
        feature_name: Name of the feature
        records: List of records, each containing entity keys and feature values
        entity_key_columns: Column names that form the entity key

    Returns:
        Number of records written
    """
    ...

RedisStore

mlforge.online.RedisStore

Bases: OnlineStore

Redis-backed online feature store.

Stores feature values as JSON in Redis with optional TTL. Uses Redis pipelines for efficient batch operations.

Key format: mlforge:{feature_name}:{entity_key_hash} Value format: JSON serialized feature values

Attributes:

Name Type Description
host

Redis server hostname
port

Redis server port
db

Redis database number
password

Redis password (optional)
ttl

Time-to-live in seconds (optional, None = no expiry)
prefix

Key prefix (default: "mlforge")
Example

store = RedisStore(host="localhost", port=6379, ttl=3600)

Write single entity

store.write( feature_name="user_spend", entity_keys={"user_id": "user_123"}, values={"amount__sum__7d": 1500.0}, )

Read single entity

features = store.read( feature_name="user_spend", entity_keys={"user_id": "user_123"}, )

Returns:

Source code in src/mlforge/online.py 
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
class RedisStore(OnlineStore):
    """
    Redis-backed online feature store.

    Stores feature values as JSON in Redis with optional TTL.
    Uses Redis pipelines for efficient batch operations.

    Key format: mlforge:{feature_name}:{entity_key_hash}
    Value format: JSON serialized feature values

    Attributes:
        host: Redis server hostname
        port: Redis server port
        db: Redis database number
        password: Redis password (optional)
        ttl: Time-to-live in seconds (optional, None = no expiry)
        prefix: Key prefix (default: "mlforge")

    Example:
        store = RedisStore(host="localhost", port=6379, ttl=3600)

        # Write single entity
        store.write(
            feature_name="user_spend",
            entity_keys={"user_id": "user_123"},
            values={"amount__sum__7d": 1500.0},
        )

        # Read single entity
        features = store.read(
            feature_name="user_spend",
            entity_keys={"user_id": "user_123"},
        )
        # Returns: {"amount__sum__7d": 1500.0}
    """

    def __init__(
        self,
        host: str = "localhost",
        port: int = 6379,
        db: int = 0,
        password: str | None = None,
        ttl: int | None = None,
        prefix: str = "mlforge",
    ) -> None:
        """
        Initialize Redis online store.

        Args:
            host: Redis server hostname. Defaults to "localhost".
            port: Redis server port. Defaults to 6379.
            db: Redis database number. Defaults to 0.
            password: Redis password. Defaults to None.
            ttl: Time-to-live in seconds. Defaults to None (no expiry).
            prefix: Key prefix for all keys. Defaults to "mlforge".

        Raises:
            ImportError: If redis package is not installed
        """
        try:
            import redis
        except ImportError as e:
            raise ImportError(
                "Redis package not installed. "
                "Install with: pip install mlforge[redis]"
            ) from e

        self.host = host
        self.port = port
        self.db = db
        self.password = password
        self.ttl = ttl
        self.prefix = prefix

        self._client = redis.Redis(
            host=host,
            port=port,
            db=db,
            password=password,
            decode_responses=True,
        )

    def _build_key(self, feature_name: str, entity_keys: dict[str, str]) -> str:
        """Build Redis key with configured prefix."""
        entity_hash = _compute_entity_hash(entity_keys)
        return f"{self.prefix}:{feature_name}:{entity_hash}"

    @override
    def write(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
        values: dict[str, Any],
    ) -> None:
        """
        Write feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values
            values: Feature column values to store
        """
        key = self._build_key(feature_name, entity_keys)
        value_json = json.dumps(values)

        if self.ttl:
            self._client.setex(key, self.ttl, value_json)
        else:
            self._client.set(key, value_json)

    @override
    def write_batch(
        self,
        feature_name: str,
        records: list[dict[str, Any]],
        entity_key_columns: list[str],
    ) -> int:
        """
        Write feature values for multiple entities using Redis pipeline.

        Args:
            feature_name: Name of the feature
            records: List of records with entity keys and feature values
            entity_key_columns: Column names that form the entity key

        Returns:
            Number of records written
        """
        if not records:
            return 0

        pipe = self._client.pipeline()

        for record in records:
            # Extract entity keys
            entity_keys = {col: str(record[col]) for col in entity_key_columns}

            # Extract feature values (all columns except entity keys)
            values = {
                k: v for k, v in record.items() if k not in entity_key_columns
            }

            key = self._build_key(feature_name, entity_keys)
            value_json = json.dumps(values)

            if self.ttl:
                pipe.setex(key, self.ttl, value_json)
            else:
                pipe.set(key, value_json)

        pipe.execute()
        return len(records)

    @override
    def read(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> dict[str, Any] | None:
        """
        Read feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            Feature values dict, or None if not found
        """
        key = self._build_key(feature_name, entity_keys)
        value = cast(str | None, self._client.get(key))

        if value is None:
            return None

        return json.loads(value)

    @override
    def read_batch(
        self,
        feature_name: str,
        entity_keys: list[dict[str, str]],
    ) -> list[dict[str, Any] | None]:
        """
        Read feature values for multiple entities using Redis pipeline.

        Args:
            feature_name: Name of the feature
            entity_keys: List of entity key dicts

        Returns:
            List of feature value dicts (None for missing entities)
        """
        if not entity_keys:
            return []

        pipe = self._client.pipeline()

        for keys in entity_keys:
            key = self._build_key(feature_name, keys)
            pipe.get(key)

        results = pipe.execute()

        return [
            json.loads(value) if value is not None else None
            for value in results
        ]

    @override
    def delete(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> bool:
        """
        Delete feature values for a single entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            True if deleted, False if not found
        """
        key = self._build_key(feature_name, entity_keys)
        deleted = cast(int, self._client.delete(key))
        return deleted > 0

    @override
    def exists(
        self,
        feature_name: str,
        entity_keys: dict[str, str],
    ) -> bool:
        """
        Check if feature values exist for an entity.

        Args:
            feature_name: Name of the feature
            entity_keys: Entity key columns and values

        Returns:
            True if exists, False otherwise
        """
        key = self._build_key(feature_name, entity_keys)
        count = cast(int, self._client.exists(key))
        return count > 0

    def ping(self) -> bool:
        """
        Check Redis connection.

        Returns:
            True if connected, False otherwise
        """
        try:
            return bool(self._client.ping())
        except Exception:
            return False

__init__ 

__init__(
    host: str = "localhost",
    port: int = 6379,
    db: int = 0,
    password: str | None = None,
    ttl: int | None = None,
    prefix: str = "mlforge",
) -> None

Initialize Redis online store.

Parameters:

Name Type Description Default
host str

Redis server hostname. Defaults to "localhost".

 'localhost'
port int

Redis server port. Defaults to 6379.

 6379
db int

Redis database number. Defaults to 0.

 0
password str | None

Redis password. Defaults to None.

 None
ttl int | None

Time-to-live in seconds. Defaults to None (no expiry).

 None
prefix str

Key prefix for all keys. Defaults to "mlforge".

 'mlforge'

Raises:

Type Description
ImportError

If redis package is not installed
Source code in src/mlforge/online.py 
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
def __init__(
    self,
    host: str = "localhost",
    port: int = 6379,
    db: int = 0,
    password: str | None = None,
    ttl: int | None = None,
    prefix: str = "mlforge",
) -> None:
    """
    Initialize Redis online store.

    Args:
        host: Redis server hostname. Defaults to "localhost".
        port: Redis server port. Defaults to 6379.
        db: Redis database number. Defaults to 0.
        password: Redis password. Defaults to None.
        ttl: Time-to-live in seconds. Defaults to None (no expiry).
        prefix: Key prefix for all keys. Defaults to "mlforge".

    Raises:
        ImportError: If redis package is not installed
    """
    try:
        import redis
    except ImportError as e:
        raise ImportError(
            "Redis package not installed. "
            "Install with: pip install mlforge[redis]"
        ) from e

    self.host = host
    self.port = port
    self.db = db
    self.password = password
    self.ttl = ttl
    self.prefix = prefix

    self._client = redis.Redis(
        host=host,
        port=port,
        db=db,
        password=password,
        decode_responses=True,
    )

delete 

delete(
    feature_name: str, entity_keys: dict[str, str]
) -> bool

Delete feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
bool

True if deleted, False if not found
Source code in src/mlforge/online.py 
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
@override
def delete(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> bool:
    """
    Delete feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        True if deleted, False if not found
    """
    key = self._build_key(feature_name, entity_keys)
    deleted = cast(int, self._client.delete(key))
    return deleted > 0

exists 

exists(
    feature_name: str, entity_keys: dict[str, str]
) -> bool

Check if feature values exist for an entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
bool

True if exists, False otherwise
Source code in src/mlforge/online.py 
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
@override
def exists(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> bool:
    """
    Check if feature values exist for an entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        True if exists, False otherwise
    """
    key = self._build_key(feature_name, entity_keys)
    count = cast(int, self._client.exists(key))
    return count > 0

ping 

ping() -> bool

Check Redis connection.

Returns:

Type Description
bool

True if connected, False otherwise
Source code in src/mlforge/online.py 
422
423
424
425
426
427
428
429
430
431
432
def ping(self) -> bool:
    """
    Check Redis connection.

    Returns:
        True if connected, False otherwise
    """
    try:
        return bool(self._client.ping())
    except Exception:
        return False

read 

read(
    feature_name: str, entity_keys: dict[str, str]
) -> dict[str, Any] | None

Read feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required

Returns:

Type Description
dict[str, Any] | None

Feature values dict, or None if not found
Source code in src/mlforge/online.py 
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
@override
def read(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
) -> dict[str, Any] | None:
    """
    Read feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values

    Returns:
        Feature values dict, or None if not found
    """
    key = self._build_key(feature_name, entity_keys)
    value = cast(str | None, self._client.get(key))

    if value is None:
        return None

    return json.loads(value)

read_batch 

read_batch(
    feature_name: str, entity_keys: list[dict[str, str]]
) -> list[dict[str, Any] | None]

Read feature values for multiple entities using Redis pipeline.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys list[dict[str, str]]

List of entity key dicts

 required

Returns:

Type Description
list[dict[str, Any] | None]

List of feature value dicts (None for missing entities)
Source code in src/mlforge/online.py 
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
@override
def read_batch(
    self,
    feature_name: str,
    entity_keys: list[dict[str, str]],
) -> list[dict[str, Any] | None]:
    """
    Read feature values for multiple entities using Redis pipeline.

    Args:
        feature_name: Name of the feature
        entity_keys: List of entity key dicts

    Returns:
        List of feature value dicts (None for missing entities)
    """
    if not entity_keys:
        return []

    pipe = self._client.pipeline()

    for keys in entity_keys:
        key = self._build_key(feature_name, keys)
        pipe.get(key)

    results = pipe.execute()

    return [
        json.loads(value) if value is not None else None
        for value in results
    ]

write 

write(
    feature_name: str,
    entity_keys: dict[str, str],
    values: dict[str, Any],
) -> None

Write feature values for a single entity.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
entity_keys dict[str, str]

Entity key columns and values

 required
values dict[str, Any]

Feature column values to store

 required
Source code in src/mlforge/online.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
@override
def write(
    self,
    feature_name: str,
    entity_keys: dict[str, str],
    values: dict[str, Any],
) -> None:
    """
    Write feature values for a single entity.

    Args:
        feature_name: Name of the feature
        entity_keys: Entity key columns and values
        values: Feature column values to store
    """
    key = self._build_key(feature_name, entity_keys)
    value_json = json.dumps(values)

    if self.ttl:
        self._client.setex(key, self.ttl, value_json)
    else:
        self._client.set(key, value_json)

write_batch 

write_batch(
    feature_name: str,
    records: list[dict[str, Any]],
    entity_key_columns: list[str],
) -> int

Write feature values for multiple entities using Redis pipeline.

Parameters:

Name Type Description Default
feature_name str

Name of the feature

 required
records list[dict[str, Any]]

List of records with entity keys and feature values

 required
entity_key_columns list[str]

Column names that form the entity key

 required

Returns:

Type Description
int

Number of records written
Source code in src/mlforge/online.py 
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
@override
def write_batch(
    self,
    feature_name: str,
    records: list[dict[str, Any]],
    entity_key_columns: list[str],
) -> int:
    """
    Write feature values for multiple entities using Redis pipeline.

    Args:
        feature_name: Name of the feature
        records: List of records with entity keys and feature values
        entity_key_columns: Column names that form the entity key

    Returns:
        Number of records written
    """
    if not records:
        return 0

    pipe = self._client.pipeline()

    for record in records:
        # Extract entity keys
        entity_keys = {col: str(record[col]) for col in entity_key_columns}

        # Extract feature values (all columns except entity keys)
        values = {
            k: v for k, v in record.items() if k not in entity_key_columns
        }

        key = self._build_key(feature_name, entity_keys)
        value_json = json.dumps(values)

        if self.ttl:
            pipe.setex(key, self.ttl, value_json)
        else:
            pipe.set(key, value_json)

    pipe.execute()
    return len(records)