Read from SQL database - dispatched via @ray.remote to cluster workers. Accepts connection_url (str) instead of a connection_factory callable so the argument is serialisable across the Ray object store boundary.
(self, sql: str, connection_url: str, **kwargs)
| 530 | return RemoteDatasetProxy(remote_fn.remote(uri, database, collection, kwargs)) |
| 531 | |
| 532 | def read_sql(self, sql: str, connection_url: str, **kwargs) -> Any: |
| 533 | """Read from SQL database - dispatched via @ray.remote to cluster workers. |
| 534 | |
| 535 | Accepts connection_url (str) instead of a connection_factory callable so |
| 536 | the argument is serialisable across the Ray object store boundary. |
| 537 | """ |
| 538 | from feast.infra.ray_shared_utils import RemoteDatasetProxy |
| 539 | |
| 540 | @ray.remote |
| 541 | def _remote(sql_query, conn_url, read_kwargs): |
| 542 | import ray |
| 543 | import sqlalchemy |
| 544 | |
| 545 | def _factory(): |
| 546 | return sqlalchemy.create_engine(conn_url).raw_connection() |
| 547 | |
| 548 | return ray.data.read_sql(sql_query, _factory, **read_kwargs) |
| 549 | |
| 550 | opts = self._get_task_options() |
| 551 | remote_fn = _remote.options(**opts) if opts else _remote |
| 552 | return RemoteDatasetProxy(remote_fn.remote(sql, connection_url, kwargs)) |
| 553 | |
| 554 | def from_huggingface( |
| 555 | self, dataset_name: str, split: str = "train", **kwargs |
nothing calls this directly
no test coverage detected