aligned.data_source.batch_data_source.BatchDataSource

class documentation

class BatchDataSource: (source)

Known subclasses: aligned.data_source.batch_data_source.CodableBatchDataSource, aligned.data_source.model_predictor.PredictModelSource

View In Hierarchy

Undocumented

Class Method	`multi_source_features_for`	Undocumented
Method	`__hash__`	Undocumented
Method	`all`	Undocumented
Method	`all_between_dates`	Undocumented
Method	`all_columns`	Undocumented
Method	`all_data`	Undocumented
Method	`depends_on`	Undocumented
Async Method	`feature_view_code`	Setup the code needed to represent the data source as a feature view
Method	`features_for`	Undocumented
Method	`filter`	Undocumented
Async Method	`freshness`	.table("my_table") .freshness()
Method	`job_group_key`	A key defining which sources can be grouped together in one request.
Method	`location_id`	Undocumented
Async Method	`schema`	Returns the schema for the data source
Method	`source_id`	An id that identifies a source from others.
Method	`tags`	Undocumented
Method	`transform_with_polars`	Undocumented
Method	`with_loaded_at`	Undocumented
Method	`with_view`	Undocumented

@classmethod
def multi_source_features_for(cls: type[T], facts: RetrivalJob, requests: list[tuple[T, RetrivalRequest]]) -> RetrivalJob: (source) ¶

overridden in aligned.CustomMethodDataSource, aligned.data_source.batch_data_source.FilteredDataSource, aligned.data_source.batch_data_source.LoadedAtSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.in_mem_source.InMemorySource, aligned.sources.lancedb.LanceDbTable, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.random_source.RandomDataSource, aligned.sources.redshift.RedshiftSQLDataSource

Undocumented

def __hash__(self) -> int: (source) ¶

overridden in aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.redshift.RedshiftSQLDataSource

Undocumented

def all(self, result: RequestResult, limit: int | None = None) -> RetrivalJob: (source) ¶

Undocumented

def all_between_dates(self, request: RetrivalRequest, start_date: datetime, end_date: datetime) -> RetrivalJob: (source) ¶

overridden in aligned.CustomMethodDataSource, aligned.data_source.batch_data_source.FilteredDataSource, aligned.data_source.batch_data_source.JoinAsofDataSource, aligned.data_source.batch_data_source.JoinDataSource, aligned.data_source.batch_data_source.LoadedAtSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.random_source.RandomDataSource, aligned.sources.redshift.RedshiftSQLDataSource, aligned.sources.s3.AwsS3CsvDataSource

Undocumented

def all_columns(self, limit: int | None = None) -> RetrivalJob: (source) ¶

Undocumented

def all_data(self, request: RetrivalRequest, limit: int | None) -> RetrivalJob: (source) ¶

overridden in aligned.CustomMethodDataSource, aligned.data_source.batch_data_source.FilteredDataSource, aligned.data_source.batch_data_source.JoinAsofDataSource, aligned.data_source.batch_data_source.JoinDataSource, aligned.data_source.batch_data_source.LoadedAtSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.lancedb.LanceDbTable, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.random_source.RandomDataSource, aligned.sources.redshift.RedshiftSQLDataSource, aligned.sources.s3.AwsS3CsvDataSource

Undocumented

def depends_on(self) -> set[FeatureLocation]: (source) ¶

Undocumented

async def feature_view_code(self, view_name: str) -> str: (source) ¶

overridden in aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource

Setup the code needed to represent the data source as a feature view

```python FileSource.parquet("my_path.parquet").feature_view_code(view_name="my_view")

>>> """from aligned import FeatureView, String, Int64, Float

class MyView(FeatureView):

metadata = FeatureView.metadata_with(

name="Embarked", description="some description", batch_source=FileSource.parquest("my_path.parquet") stream_source=None,

)

Passenger_id = Int64() Survived = Int64() Pclass = Int64() Name = String() Sex = String() Age = Float() Sibsp = Int64() Parch = Int64() Ticket = String() Fare = Float() Cabin = String() Embarked = String()"""

```

Returns:: str: The code needed to setup a basic feature view

def features_for(self, facts: RetrivalJob, request: RetrivalRequest) -> RetrivalJob: (source) ¶

overridden in aligned.CustomMethodDataSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource

Undocumented

def filter(self, condition: DerivedFeature | Feature) -> CodableBatchDataSource: (source) ¶

Undocumented

async def freshness(self, feature: Feature) -> datetime | None: (source) ¶

my_table_freshenss = await (PostgreSQLConfig("DB_URL"): .table("my_table") .freshness()

)

def job_group_key(self) -> str: (source) ¶

overridden in aligned.CustomMethodDataSource, aligned.data_source.batch_data_source.FilteredDataSource, aligned.data_source.batch_data_source.JoinAsofDataSource, aligned.data_source.batch_data_source.JoinDataSource, aligned.data_source.batch_data_source.LoadedAtSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.in_mem_source.InMemorySource, aligned.sources.lancedb.LanceDbTable, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.random_source.RandomDataSource, aligned.sources.redshift.RedshiftSQLDataSource, aligned.sources.s3.AwsS3CsvDataSource, aligned.sources.s3.AwsS3ParquetDataSource

A key defining which sources can be grouped together in one request.

def location_id(self) -> set[FeatureLocation]: (source) ¶

overridden in aligned.data_source.model_predictor.PredictModelSource

Undocumented

async def schema(self) -> dict[str, FeatureType]: (source) ¶

overridden in aligned.data_source.batch_data_source.FilteredDataSource, aligned.data_source.batch_data_source.JoinAsofDataSource, aligned.data_source.batch_data_source.JoinDataSource, aligned.data_source.batch_data_source.LoadedAtSource, aligned.data_source.batch_data_source.StackSource, aligned.data_source.model_predictor.PredictModelSource, aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobDeltaDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.local.CsvFileSource, aligned.sources.local.DeltaFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource, aligned.sources.psql.PostgreSQLDataSource, aligned.sources.random_source.RandomDataSource

Returns the schema for the data source

`python source = FileSource.parquet_at('test_data/titanic.parquet') schema = await source.schema() >>> {'passenger_id': FeatureType(name='int64'), ...} `

Returns:: dict[str, FeatureType]: A dictionary containing the column name and the feature type

def source_id(self) -> str: (source) ¶

overridden in aligned.sources.psql.PostgreSQLDataSource

An id that identifies a source from others.

def tags(self) -> list[str]: (source) ¶

Undocumented

def transform_with_polars(self, method: Callable[[pl.LazyFrame], Awaitable[pl.LazyFrame]] | Callable[[pl.LazyFrame], pl.LazyFrame]) -> CodableBatchDataSource: (source) ¶

Undocumented

def with_loaded_at(self) -> CodableBatchDataSource: (source) ¶

Undocumented

def with_view(self: T, view: CompiledFeatureView) -> T: (source) ¶

overridden in aligned.sources.azure_blob_storage.AzureBlobCsvDataSource, aligned.sources.azure_blob_storage.AzureBlobParquetDataSource, aligned.sources.azure_blob_storage.AzureBlobPartitionedParquetDataSource, aligned.sources.in_mem_source.InMemorySource, aligned.sources.local.CsvFileSource, aligned.sources.local.ParquetFileSource, aligned.sources.local.PartitionedParquetFileSource

Undocumented