Module lunar.ml

Sub-modules

lunar.ml.aidp_service
lunar.ml.edd_service
lunar.ml.lunar_model
lunar.ml.model_registry

Classes

class AIDPService (config: Config)

Methods

def copy_table_to_s3(self, database_name: str, table_name: str, db_type: str, operation: str, target_name: str = None, s3_dt: str = None, bq_table_partition: str = None, write_type: str = 'json', max_num_files: int = 20) ‑> Dict[str, Any]

Copy a table from AIDP BigQuery to BAP S3 and Database.

Args

  • database_name: (str) Database name of Bigquery
  • table_name: (str) Name of a table on Bigquery
  • db_type: (str) Type of database to be loaded on BAP ('dynamodb' | 'docdb' | 'es' | 's3')
  • operation: (str) Type of operation ('put' (db_type 's3' is only available on 'put') | 'update' | 'upsert' (only for db_type 'dynamodb') | 'delete')
  • target_name: (optional) (str) Target name on database to be loaded (Default: Value of table_name)
  • s3_dt: (optional) (str) Set a dt partition on BAP S3. If db_type is docdb, it is necessary
  • bq_table_partition: (optional) (str) Partition value of table on BigQuery
  • write_type: (optional) Write type ((default) 'json' | 'parquet')
  • max_num_files: (optional) (int) Maximum number of files to be saved on AWS S3 (default: 10)

Returns

dict

Example

response = copy_table_to_s3(
    database_name="apollo",
    table_name="test_table",
    db_type="dynamodb",
    operation="put",
    target_name="test_data",
    s3_dt="20211101",
    write_type="json"
    bq_table_partition="2021-09-01",
)
class EDDService (config: Config)

Methods

def copy_model_to_s3(self, model_path: str, model_name: str, model_version: str) ‑> Dict[str, Any]

Copy model binaries and meta files from EDD to BAP model registry.

Args

  • model_path: (str) Relative path of model binaries and meta (Caution: Must not start with slash and must end with slash)
  • model_name: (str) Name of a model
  • model_version: (str) Version of a model

Returns

dict

Example

response = copy_model_to_s3(model_path="myfiles/models/test_model/v1/", model_name="test_model", version="v1")
def copy_table_to_s3(self, database_name: str, table_name: str, db_type: str, operation: str, target_name: str = None, s3_dt: str = None, edd_table_partition: str = None) ‑> Dict[str, Any]

Copy a table from EDD S3 to BAP S3 and Database(dynamodb or docdb).

Args

  • database_name: (str) Database name of EDD S3 ('Datalake' is not available)
  • table_name: (str) Name of a table on EDD S3
  • db_type: (str) Type of database to be loaded on BAP ('dynamodb' | 'docdb' | 'es' | 's3')
  • operation: (str) Type of operation ('put' (db_type 's3' is only available on 'put') | 'update' | 'upsert' (only for db_type 'dynamodb') | 'delete')
  • target_name: (optional) (str) Target name on database to be loaded (Default: Value of table_name)
  • s3_dt: (optional) (str) Set a dt partition on BAP S3. If db_type is docdb, it is necessary
  • edd_table_partition: (optional) (str) Partition value of table on EDD S3

Returns

dict

Example

response = copy_table_to_s3(
    database_name="apollo",
    table_name="test_table",
    db_type="docdb",
    operation="put",
    target_name="test_data",
    s3_dt="20211101",
    edd_table_partition="dt=202109/type=put",
)
class LunarModel (name: str, version: str, predict_fn: Callable, models: Optional[List[Any]] = [])

Methods

def add_attr(self, key: str, value: Any) ‑> None
def clear_attrs(self) ‑> None
async def predict(self, id: str, channel_id: str, **kwargs) ‑> List[Dict[str, Any]]
def remove_attr(self, key: str) ‑> None
class ModelRegistry (config: Config)

Methods

def save(self, lunar_model: LunarModel, force: bool = False, to_local: bool = False) ‑> None