nayrïr thakhinak lurapxta, mä pipeline MLOps tukuyat tukuykama lurañataki, Databricks ukat Spark uka apnaqasa, Databricks ukan arquitectura de referencia ukamp irpata. Akax mä recapitulación ukawa, kuna jach’a amtawinaktix uñakipt’awayktan ukanakxa: Aka yatichäw serien nayrïr t’aqapanx : Jiwasan yatiyawinakasax bronce, qullqi ukat quri capas ukanakaruw Catálogo de Unidad ukan wakicht’apxta, mä sistema de gestión de datos estructurado ukat eficiente ukham utt’ayasa. Catálogo de Unidad ukax Arquitectura de Medallón ukatakiw utt’ayata : Jiwasax uñacht’ayapxtwa kunjams crudo datos ukax sistema ukar apkatasispa, ukhamat mä consistencia ukat calidad ukax qhipa etapas de procesamiento ukataki. Ingesting Data into Unity Catalogue : Databricks apnaqasa, mä modelo de aprendizaje automático ukar yatichapxta, ukax jiwasan conjunto de datos ukarjam wakicht’atawa, ukax suma lurawinakampiw modelo escalable ukat askinjam lurañataki. Modelo yatichaña : Modelo ukan lurawip jach’anchayañatakixa, HyperOpt ukampiwa irnaqapxta, ukhamata hiperparámetros óptimos ukanaka thaqhañataki, ukhamata chiqapa ukhamaraki suma lurañataki. Sintonización de hiperparámetros con HyperOpt : MLflow ukampiw yant’äwinakas qillqt’añataki ukat uñjañatakis apnaqapxta, mä jach’a qillqt’äwiw modelo ukan versión, métricas ukat parámetros ukanakat jasak chiqanchañataki ukat wasitat uñstayañataki. Databricks MLflow ukampiw yant’äwinak arknaqañax Aka fundamental lurawinakampi tukuyatatxa, modelo ukax jichhax primed ukhamawa despliegue ukataki. Aka payïr t’aqapanx pä componentes criticos ukanakaw sistema ukar mayacht’añatak uñakipt’atäni: : Procesamiento de lote uka phuqhaña jach’a conjuntos de datos ukan yatiyawinak lurañataki, ukax aplicaciones ukanakatakix askiwa kunjamakitix puntuación a granel ukhamarak informe periódico. Inferencia de Lotes : Modelo de tiempo real ukan utt’ayaña servir ukhamat jank’ak yatiyawinak churañataki, ukax wali wakiskiriwa aplicaciones interactivas ukat servicios ukanakataki. Inferencia en Línea (Servicio Modelo) ukhamata modelos desplegados ukanakaxa suma lurawi ukhamaraki confiabilidad uka pachana utjañapataki. Modelo Monitoreo: ¡Ukar mantañäni! Modelo de Despliegue ukax mä juk’a pachanakanwa Qhipa blog ukan saräwipax modelo de evaluación ukawa. Jichhax amuyt’añäni, uñakipt’awix lurawaytanwa ukatx jikxatawayktanwa jiwasan modelo ukax mä jach’a lurawi uñacht’ayi aka modelo de producción ukar uñtasita. Kunjamatix jiwasax (asumi) modelo ukax producción ukan apnaqañ munktanxa, taqi datos ukanak aprovechañ munapxtan kunatix utjkistu ukanakxa. Jutïr amtawix modelo ukar yatichañawa ukat yant’añawa, ukax taqpach conjunto de datos ukampiw lurasi. Ukatx jiwasan modelo ukax qhipa apnaqañatakiw ch’amanchasi, ukax jiwasan modelo campeón ukham uñstayasa. Niyakixay akax qhipa modelo ukax inferencia ukatak apnaqañ munstanxa, ukax cliente de Ingeniería de Características ukampiw modelo ukar yatichañatak apnaqapxta. Ukhamatx janiw linaje modelo ukak jasak arknaqktanti, jan ukasti esquema validación ukat características transformación (ukax utjchi ukhax) cliente ukaruw descargar. with mlflow.start_run(run_name="ALS_best_model") as run: als = ALS() # Now we set the parameters for the method als.setMaxIter(MAX_ITER)\ .setSeed(SEED)\ .setRegParam(best_params["REG_PARAM"])\ .setUserCol(COL_USER)\ .setItemCol(COL_ITEM)\ .setRatingCol(COL_LABEL)\ .setRank(best_params["RANK"]) mlflow.log_param("MAX_ITER", MAX_ITER) mlflow.log_param("RANK", best_params["RANK"]) mlflow.log_param("REG_PARAM", best_params["REG_PARAM"]) # Create the model with these parameters. model = als.fit(df_full_data) #drop predictions where users and products from the test test and didn't make it into the training set. in this case, the prediction is NaN model.setColdStartStrategy('drop') predictions = model.transform(df_full_data) signature = infer_signature(model_input = df_full_data, model_output = predictions.select(COL_LABEL)) #log the model mlflow.spark.log_model(model, model_name, sample_input=df_full_data.limit(3), signature = signature, conda_env=mlflow.spark.get_default_conda_env(), registered_model_name=f"{catalog_name}.{model_schema}.{model_name}") evaluator = RegressionEvaluator(predictionCol=COL_PRED, labelCol=COL_LABEL) rmse = evaluator.setMetricName("rmse").evaluate(predictions) mlflow.log_metric('rmse', rmse) ukhamarakiw apnaqaraksna, modelos ukanakar yatichañataki ukhamarak qillqt’añataki Feature Store jan ukax Feature Engineering APIs ukanak model_info = fe.log_model(model=model, artifact_path = model_name, flavor=mlflow.spark, training_set=fe_full_data, conda_env=mlflow.spark.get_default_conda_env(), registered_model_name= f"{catalog_name}.{model_schema}.{model_name}" ) kunapachatix API ingeniería de características uka apnaqktan ukhax modelo ukan linaje ukar Catálogo Explorer ukan uñakipt’araksnawa Jichhax modelo ukan qhanañchäwip machaqar tukuyañäni ukat mä etiqueta Campeón ukar churañäni. import time from mlflow.tracking.client import MlflowClient from mlflow.entities.model_registry.model_version_status import ModelVersionStatus client = MlflowClient() #find the latest model version model_name_path = f"{catalog_name}.{model_schema}.{model_name}" model_version_infos = client.search_model_versions(f"name ='{model_name_path}'") new_model_version = max([int(model_version_info.version) for model_version_info in model_version_infos]) #add the model and model version descirption client.update_registered_model( name=model_name_path, description="collaborative filtering using Spark mllib ALS. This model use rating table" ) client.update_model_version( name=model_name_path, version=new_model_version, description="this model is optimized Rank and REG_PARAM with Hyperopt and rmse as a loss function. trained on the full dataset" ) # assign alias client.set_registered_model_alias(model_name_path, "Champion", new_model_version) Jichhax nayrar sartañamawa ukat esquema uñakipt’añamawa kunatix modelo ukar qillqantatawa. taqi machaq lurawinakamx akham uñjañamawa : Ukax registro de modelo ukatakix irnaqäw chiqawj apnaqañax modelos ukanakar apnaqañatakix etapas ukanakaw wakisi. Alias ukanakamp apnaqañax janiw irnaqkaniti. uñakipt’añäni kunjams irnaqaski uk uñjañataki Modelo etapas Aka chiqan Inferencia Modelo ukax mä juk’a pachanakanwa Lote Puntuación ukax mä juk’a pachanakanwa Jichhax amuyt’añäni, jiwasan modelo ukar producción ukan inferencia ukar apnaqañ munapxta. Aka thakhinx modelo campeón ukaruw cargapxta ukatx ukampiw 20 películas ukan iwxt’awinakap sapa apnaqirinakatak uñstayañataki. from mlflow.spark import load_model as spark_load_model from mlflow.tracking.client import MlflowClient from create_training_set import split_data #-- set UC as model registray mlflow.set_registry_uri("databricks-uc") #-- initate mlflow client client = MlflowClient() # -- read the config file with open('config.json') as config_file: config = json.load(config_file) catalog_name = config["catalog_name"] gold_layer = config["gold_layer_name"] silver_layer = config["silver_layer_name"] user_item_table_name = config["user_item_table_name"] ft_user_item_name = config["ft_user_item_name"] model_name = config["model_name"] model_schema = config["model_schema"] #-- create the model uri model_path = f"{catalog_name}.{model_schema}.{model_name}" # --create the model_uri: there are two ways to do this # 1: using the alias (we use this*) model_version_uri = f"models:/{model_uri}@champion" # 2: using model version #champion_version = client.get_model_version_by_alias(model_uri, "champion") #model_version_uri = f"models:/{model_uri}/{champion_version.version}" # -- load the model pipline and exctract the model model_pipeline = spark_load_model(model_version_uri) model = model_pipeline.stages[0] # -- batch scoring using the the model fe_full_data, df_full_data, df_train, df_test = split_data() df_batch_input = df_full_data.drop("rating") df_scores = model.transform(df_batch_input) # --- in case you used Feature Engineering to train and register model #from databricks.feature_engineering import FeatureEngineeringClient #fe = FeatureEngineeringClient() # fe.score_batch(model_uri=f"{model_version_uri}",df = df_batch_input) ukat uñjapxasmawa pachpa yatichäw yatiyawinak apnaqapxta lote puntuación ukataki. Sistemas recomendadores ukanx mä amuyunikïkchisa, jilpach aplicación ukanx modelo uka apnaqañ munapxta yaqhip jan uñjat yatiyawinak puntuación ukar puriñapataki. Amuyt’añataki, Imaging your are Netflix ukat apnaqirin iwxt’awinakaparux uru tukuyarux machaq uñjat lista ukarjam machaqar tukuyañ munapxi. Jiwasax irnaqäw wakicht’araksnawa, ukax lote puntuación ukax mä hora específico ukanw uru tukuyanx apnaqasi. Jichhax nayrar sartañasawa ukat sapa apnaqiritak iwxt’awinak uñstayañasawa. Ukatakix sapa apnaqirinakarux 20 nayrïr yänak jikxatapxta from pyspark.sql.window import Window from pyspark.sql.functions import col, split, row_number, collect_list from pyspark.sql.functions import col, collect_list, expr, lit, min, row_number, desc windowSpec = Window.partitionBy("user_id").orderBy(col("prediction").desc()) df_top_20_items = df_scores.withColumn("rank", row_number().over(windowSpec)).filter(col("rank") <= 20) df_user_recs = df_top_20_items.groupBy("user_id") \ .agg(collect_list(col("item_id").cast("double")).alias("top_item_ids")) akax kunjams uka resultadox uñstaski ukawa Tukuyañatakix predicción ukax mä etiqueta delta ukhamaw jiwasan UC ukan imasispa jan ukax sistemas descendentes Mongo DB jan ukax Azure Cosmos DB ukar uñt’ayaraksnawa. Jiwasax firs opción ukampiw sarapxta df_user_recs.write.mode("overwrite").saveAsTable(f"{catalog_name}.{output_schema}.top20_item_recommendations") Streaming/Inferencia en línea ukax mä juk’a pachanakanwa Jichhax mä caso amuyt’añäni, kawkhantix chiqpach pachan apnaqirinakan mayacht’asiwiparjamax jiwasan iwxt’awinakas machaqar tukuyañ munapxta. Aka tuqitx modelo serving ukampiw apnaqasispa. Kunawsatix maynix modelo ukar apnaqañ munki ukhax servidor ukarux datos ukanakaw apayanispa. Ukatx servidor ukax uka datos ukarux modelo desplegado ukarux manq’ayaraki, ukax lurawiruw sararaki, datos ukarux uñakipi, ukatx mä predicción ukaw lurasi. Ukanakax aplicaciones web, apps móviles jan ukax sistemas embebidos ukanakan apnaqasispawa. Aka amtawin mä apnaqawipax A/B yant’awinakatakix ruteo de tráfico ukar yanapt’añawa. Algoritmo ALS ukax janiw chiqak inferencia en línea ukatakix apnaqatäkaspati kunatix modelo ukar wasitat yatichañax wakisiwa taqpach datos (nayra + machaq) apnaqasa, iwxt’awinak machaqar tukuyañataki. Algoritmos de aprendizaje de Gradiente Descenso ukax modelo ukan uñacht’ayatawa, ukax internet tuqin machaq uñstawinakatakiw apnaqasispa. Inas jutïr post ukanx yaqhip uka algoritmos ukanakax uñakipt’araksna. Ukampirus, kunjams ukham modelox irnaqaspa uk uñacht’ayañatakik, mä (inútil) modelo lurapxta, ukax punto final ukar yanapt’i, ukax películas ukan calificación ukarjam uñt’ayi, kunapachatix mä usuariox mä películas ukar uñt’ayi! import requests model_path = f"{catalog_name}.{model_schema}.{model_name}" champion_version = client.get_model_version_by_alias(model_path, "champion") # Set the name of the MLflow endpoint endpoint_name = config["model_serving_endpoint_name"] # Name of the registered MLflow model model_name = model_path # Specify the type of compute (CPU, GPU_SMALL, GPU_MEDIUM, etc.) workload_type = "CPU" # Specify the scale-out size of compute (Small, Medium, Large, etc.) workload_size = "Small" # Get the latest version of the MLflow model model_version = int(champion_version.version) # Specify Scale to Zero(only supported for CPU endpoints) scale_to_zero = False # Get the API endpoint and token for the current notebook context API_ROOT = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiUrl().get() API_TOKEN = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get() data = { "name": endpoint_name, "config": { "served_models": [ { "model_name": model_name, "model_version": int(model_version), "workload_size": workload_size, "scale_to_zero_enabled": scale_to_zero, "workload_type": workload_type, } ] }, } headers = {"Context-Type": "text/json", "Authorization": f"Bearer {API_TOKEN}"} response = requests.post( url=f"{API_ROOT}/api/2.0/serving-endpoints", json=data, headers=headers ) Akax lurani ukat chika uru modelo serving cluster jiwasataki ukhamat mä juk’a pachaw munasispa. Jichhax ventana jist’arañax tukuyañ chiqawj uñjañama. Serving mä tukuyañ chiqaw walja modelo ukar yanapt’añatak apnaqaraksna. Ukatx ruteo de tráfico ukampiw escenarios ukanakatakix apnaqaraksna, kunjamatix A/B yant’awinakataki jan ukax modelos de diferencia ukan lurawip producción ukan uñakipt’araksnawa. Tabla de Inferencia ukaxa mä juk’a pachanakwa lurasi Tablas de inferencias ukanakax Databricks Model Serving ukanx mä registro automático ukhamaw jiwasan modelos desplegados ukanakatakix irnaqapxi. Kunawsatix ch’amanchatäki ukhax mantanir mayiwinak (datos ukanakax yatiyañatak apayata), modelo correspondiente ukan mistuwinakap (predicciones) ukat yaqhip yaqha metadatos ukanakax mä tabla Delta ukham Catálogo de Unidad ukan katjapxi. Jiwasax tabla de inferencias ukanak apnaqaraksnawa, , , ukatx mä procedimiento de recogida de datos ukampiw modelos ukanakar jan ukax . uñakipañataki ukhamarak depuración linaje ukar arknaqañataki wasitat yatichañataki suma askichañataki Jiwasax ukax jiwasan punto final de servicio ukanx modelo ukar uñjañatakis ch’amancharaksnawa. Jiwasax luraraksnawa, ukan propiedades ukanakax payload ukan uñacht’ayasa, kunapachatix nayrïr kutix tukuyañ chimpu lurasktan ukhaxa. Jan ukax tukuyañ chiqarux qhipat machaqar tukuytan kamachimpi ukat tukuyañ URL ukampiw akham (juk’amp inference table auto_capture_config put config ) aka chiqan data = { "served_models": [ { "model_name": model_name, "model_version": int(model_version), "workload_size": workload_size, "scale_to_zero_enabled": scale_to_zero, "workload_type": workload_type, } ], "auto_capture_config":{ "catalog_name": catalog_name, "schema_name": model_schema, "table_name_prefix": payload_table, } } headers = {"Context-Type": "application/json", "Authorization": f"Bearer {API_TOKEN}"} response = requests.put(url=f"{API_ROOT}/api/2.0/serving-endpoints/{endpoint_name}/config", json=data, headers=headers) print(json.dumps(response.json(), indent=4)) jichhax mä qawqha simulacro usuario interacción datos ukanakamp tukuyañ chiqar manq’ayañäni import random import time all_items = df_full_data.select(col("item_id")).distinct() for user_id in range(50,54): items_not_rated_by_user = df_full_data.where(col("user_id")==user_id).select(col("item_id")).distinct()#collect()[0][0] no_rated_items = [item.item_id for item in all_items.subtract(items_not_rated_by_user).limit(4).collect()] data = { "dataframe_records": [ {"user_id":user_id, "item_id":no_rated_items[0], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[1], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[2], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[2], "rating": random.randint(1, 5)}, ] } response = requests.post( url=f"{API_ROOT}/serving-endpoints/{endpoint_name}/invocations", json=data, headers=headers ) # generate the data within the timespan of 1 to 8 minutes time.sleep(random.randint(60*1, 60*8)) Jiwasax tukuyañ qillqatanakax tabla ukan uñakipt’araksnawa. Ukax niya 10 minutos ukjaw tabla ukan datos ukanakax uñjasi. <catalog>.<schema>.<payload_table> table_path = f"{catalog_name}.{model_schema}.{payload_table}" # Read data from the inference table df_inf_table = spark.read.table(table_path) display(df_inf_table ) jumax uñjañamawa kunatix aka payload tabla Aka inferencia tabla ukan esquema uñakipañatakixa, “Unidad catálogo inferencia tabla esquema==” uñakipt’aña .== aka tuqina Modelo Uñakipañataki Modelo ukat datos uñakipaña mä complejo tema ukax walja pachaw munasispa masterización ukataki. Databricks Lakehouse Monitoring (DLM) ukaxa jisk’acharakiw mä sistema de monitoreo aski lurañatakixa, plantillas estándar ukhamaraki personalizables ukanakampi, casos de uso común ukanakataki. Ukampirus DLM ukat modelo de monitoreo general ukar masterización ukax walja experimentaciones ukanakaw wakisi. Janiw nayax mä jach’a uñakipäw modelo de monitoreo uka tuqit aka chiqan churañ munkti jan ukasti mä qalltañ chiqawj churañ munsma. Nayax jutïrinx mä blog uka tuqitw dedicañ munta. Mä jisk’a qhanañcht’awi DLM lurawinakata ukhamaraki lurawinakata Jichhax jiwasan modelo ukax utt’ayatawa, ukatx tabla de inferencias generadas por nuestro punto final de servicio ukampiw métricas claves ukanakar uñjañatakix kunjamatix modelo ukan lurawipa ukhamarak deriva ukax kuna desviaciones jan ukax anomalías ukanakas jiwasan datos jan ukax modelo ukan pachapar uñt’añataki. Aka proactivo uñakipañaxa yanapt’istuwa pachaparu chiqañchañataki, kunjamatixa modelo ukaru mayampi yatichaña jan ukaxa lurawinakapa machaqar tukuyaña, ukhamata suma lurañataki ukhamaraki alineación ukampi negocio amtanakampi. DLM ukaxa kimsa kasta uñakipaña jan ukaxa churaraki: , ukatxa . Niyakejjay jiwasan tabla de inferencias ukanakas uñakipañ munstanjja, qhepatjja, jukʼampejj uñakiptʼtanjja. Mä tabla uñakipañataki apnaqañataki - jiwasana “ ”, jiwasaxa uñjañasawa tabla ukaxa chiqapa lurawimpi. sapa filaxa mä mayiwimpi chikachasiñawa aka columnanakampi: profile type Serie de Tiempo Instantánea Inferencia tabla primaria ukatakixa , Tabla de inferencia modelo ukan uñacht’awinakapa modelo ukan yatiyawipa modelo ukax id : inferencia mayiwi pacha chimpu pacha chimpu (janiwa munañakiti) . uraqi chiqa ukax wali askiwa kunawsatix walja modelos ukar yanapt’ktan ukat sapa modelo ukan lurawip mä tablón de monitoreo ukan uñakipañ munapxta. Mä id modelo ukat sipans juk’ampi utjki ukhaxa, DLM ukax datos ukar ch’iyjañatakiw apnaqi ukatx sapa t’aqatakix métricas ukat estáticas ukanakax sapa mayniruw jakthapi. Id modelo DLM ukax sapa estadísticas ukat métricas ukanak mä específico tiempo intervalo ukatakiw jakthapi. Inferencia uñakipañatakixa, columna de ukampi apnaqatarakiwa, ukatxa mä usuario definido ventana tama ukampiwa pacha ventanas uñt’ayañataki. juk’ampi aka amparamp qillqt’ata. sello de tiempo DLM ukax pä tablas de inferencia ukatakix yanapt’i: “ ” jan ukax “ ”. Ukax yaqhip wakiskir métricas ukat estadísticas ukanak aka especificación ukarjam jakthapi. problem type clasificación regresión DLM apnaqañatakix mä monitor lurañasawa ukatx mä tabla ukar apkatañasawa. Kunawsatix aka DLM lurasktan ukhax pä lurasi : metric tables : aka tabla ukaxa suma qhananchatawa min, max, porcentaje de nulo ukatxa ceros. Ukhamaraki, yaqha métricas ukanakaw utjaraki, ukax jan walt’äw kasta apnaqirin qhanañchataparjama. Amuyt’añataki , ukat ukax modelos de clasificación ukatakiw, ukatx ukatx ukax modelos de regresión ukatakiw. tabla métrica de perfil precisión recall f1_score mean_squared_error mean_average_error : ukax estadísticas ukaniwa, ukax kunjams datos ukan jaljawipax mayjt’awayi jan ukax mä . Ukax medidas ukanakaw jakthapita, kunjamatix Chi-cuadrado prueba, KS prueba. tabla métrica de deriva pachaparjama valor de base ukar uñtasita (ukax churatarakiwa) sapa tabla ukan phuqhat métricas ukan lista uñakipañatakix documentación ukan uñakipaña. Ukhamaraki, ukax Monitor métrica tabla lurasispa. mä métricas personalizadas ukanakaw Mä wakiskir aspecto sistema de monitoreo lurañatakix jiwasan dashboard de monitoreo ukax qhipa inferencia ukan yatiyawinakap puriñapatakiw uñjañapa. Ukatakix apnaqasispa, ukhamat tabla de inferencias ukan procesadas filas ukanakar uñjañataki. Jiwasax modelo serving ukan inferencia tabla ukarux jiwasan fuente tabla ukham apnaqapxta ( ), ukatx monitoreo tabla ukarux sink tabla ukham apnaqapxta ( ). Ukhamaraki, (CDC) ukaxa panpacha tabla ukanxa ch’amanchatawa (ukaxa Tabla de Inferencia ukanxa nayraqata ch’amanchatawa). Ukhamatx mayjt’awinakak lurapxta - insert/update/delete - ukax phuqhat tabla ukanx taqpach tabla ukar sapa machaqar tukuyañan wasitat lurañat sipansa. Delta tabla streaming ukampiw readStream writeStream Mayjt’awi Datos Captura Amparampi lurata Jiwasan tabla de inferencias ukan uñjañax ch’amanchañatakix aka lurawinakampiw lurasi: Tabla de inferencia ukax mä tabla Streaming ukham uñakipt’atawa Mä machaq tabla delta luraña chiqa esquema ukampiwa despaquete tabla de inferencia ukaxa jiwasana modelo servidor punto final ukampi luratawa. Tabla de base wakicht’aña (ukaxa utjchi ukhaxa) . Mä monitor luraña tabla resultado ukatxa métrica ukar machaqar tukuyaña Mä irnaqawi wakicht’aña, ukhamata tabla de inferencia ukaxa chiqapa lurawiru apsuñataki ukhamaraki métricas ukanaka machaqar tukuyañataki Nayraqatax Lakehouse Monitoring API ukar uñt’ayañaw wakisi. Ukaxa nayraqata uñstayatawa kunatixa Databricks rum time 15.3 LTS ukatxa juk’ampi: %pip install "https://ml-team-public-read.s3.amazonaws.com/wheels/data-monitoring/a4050ef7-b183-47a1-a145-e614628e3146/databricks_lakehouse_monitoring-0.4.14-py3-none-any.whl" dbutils.library.restartPython() Tabla de inferencia ukax mä tabla de streaming ukham uñakipt’añäni requests_raw = spark.readStream\ .format("delta")\ .table(inference_table_name) requests_raw.isStreaming #-> True Ukxarusti, tabla ukarux chiqap formato ukar uchañaw wakisi kunjamatix nayraqat qhanañchawayktan ukhama. Aka tabla ukaxa sapa yatiyawitakixa mä fila ukaniñaparakiwa wakiskiri uñacht’awinakampi ukhamaraki yatiyawi chimpumpi. Tabla de inferencia ukax modelo servidor de punto final ukat apsutawa, ukax punto final mayiwinak ukat jaysawinak mä formato JSON anidado ukham imaña. Akax mä uñacht’äwiwa JSON payload ukax mayiwi ukat jaysawinak columna ukataki. #requests {"dataframe_records": [ {"user_id": 1, "item_id": 346, "rating": 5}, {"user_id": 1, "item_id": 377, "rating": 2}, {"user_id": 1, "item_id": 302, "rating": 4} ] } #reponse {"predictions": [4.248899936676025, 1.1172138452529907, 4.279165744781494] } # --> what we need | user_id | item_id | rating | prediction | |---------|---------|--------|------------| | 1 | 346 | 5 | 4.248900 | | 1 | 377 | 2 | 1.117214 | | 1 | 302 | 4 | 4.279166 | Aka tabla chiqa esquema ukar apsuñatakix aka código apnaqaraksnawa ukax Databricks documentación ukan adaptatawa ( ). Tabla de inferencia Lakehouse Monitoring starter notebook # define the schema of the request and reponse fields in the inference tabel REQUEST_FIELDS = [StructField('user_id', IntegerType(), False),\ StructField('item_id', IntegerType(), False),\ StructField('rating', IntegerType(), False)\ ] RESPONSE_FIELD = [T.StructField("predictions", FloatType())] def process_col_requests(json_str:str) -> str: """ to proccess the JSON payload of request column in inference table """ request = json.loads(json_str) dataframe_records = request.get("dataframe_records", []) return dataframe_records def procces_col_response(json_str: str) -> str: """ to proccess the JSON payload of reponse column in inference table """ reponse = json.loads(json_str) output = [{prediction_col: round(prediction,4)} for prediction in reponse["predictions"]] return output def get_model_id(endpoint_name: str) -> str: """ create the model id by concatinating the model name and the model version. note: the assumption is the endpoint serves only one model """ served_models = get_served_models(endpoint_name) model_id = f"{served_models[0]['model_name']}_{served_models[0]['model_version']}" return model_id def process_requests(requests_raw: DataFrame, request_fields: List[T.StructField], response_field: T.StructField, endpoint_name: str) -> DataFrame: """ Takes a stream of raw requests and processes them by: - Unpacking JSON payloads for requests and responses - Exploding batched requests into individual rows - Converting Unix epoch millisecond timestamps to be Spark TimestampType :param requests_raw: DataFrame containing raw requests. Assumed to contain the following columns: - `request` - `response` - `timestamp_ms` :param request_fields: List of StructFields representing the request schema :param response_field: A StructField representing the response schema :return: A DataFrame containing processed requests """ # Convert the timestamp milliseconds to TimestampType for downstream processing. requests_timestamped = requests_raw \ .withColumn(timestamp_col, (F.col("timestamp_ms") / 1000).cast(T.TimestampType())) \ .drop("timestamp_ms") # create the model identifier column model_id = get_model_id(endpoint_name) # Convert the model name and version columns into a model identifier column. requests_identified = requests_timestamped \ .withColumn(model_id_col, F.lit(model_id)) # Rename the date column to avoid collisions with features. requests_dated = requests_identified.withColumnRenamed("date", date_col) # Consolidate and unpack JSON. request_schema = T.ArrayType(T.StructType(request_fields)) response_schema = T.ArrayType(T.StructType(response_field)) # w udf_request = F.udf(process_col_requests, request_schema) udf_reponse = F.udf(procces_col_response, response_schema) requests_unpacked = requests_dated.withColumn("request", udf_request("request")).\ withColumn("response", udf_reponse("response")) # Explode batched requests into individual rows. DB_PREFIX = "__db" requests_exploded = requests_unpacked \ .withColumn(f"{DB_PREFIX}_request_response", F.arrays_zip(F.col("request"), F.col("response"))) \ .withColumn(f"{DB_PREFIX}_request_response", F.explode(F.col(f"{DB_PREFIX}_request_response"))) \ .select(F.col("*"), F.col(f"{DB_PREFIX}_request_response.request.*"), F.col(f"{DB_PREFIX}_request_response.response.*")) \ .drop(f"{DB_PREFIX}_request_response", "request", "response") requests_cleaned = requests_exploded.drop("status_code", "sampling_fraction", "client_request_id", "databricks_request_id", "request_metadata") return requests_cleaned Uka apsut tablax akham uñtasitäspawa: Ukxarusti, jiwasan lavabo tabla ukar qalltañasawa dt_builder = DeltaTable.createIfNotExists(spark) \ .tableName(unpacked_requests_table_name) \ .addColumns(schema) \ .partitionedBy(requests_cleaned.schema) \ .property("delta.enableChangeDataFeed", "true") \ dt_builder.execute() ukatxa uka yatxatatanaka qillqaña checkpoint_path = f"dbfs:/payload-logging/{endpoint_name}/checkpoint" requests_stream = requests_cleaned.writeStream \ .trigger(once=True) \ .format("delta") \ .partitionBy(date_col) \ .outputMode("append") \ .option("checkpointLocation", checkpoint_path) \ .toTable(unpacked_requests_table_name) \ Tukuyañatakix, jiwasan tabla de base ukar uñstayapxta. DLM ukax aka tabla ukampiw derivas ukanakax jakhthapita, ukax modelos de línea base ukat primaria ukanakan pachpa columnas ukanakan jaljawip uñakipt’asa. Tabla de base ukax pachpa columna de características ukaniñapaw nayrïr columna ukhamarak pachpa columna identificación de modelo ukampi. Tabla de base ukatakix tabla de predicción ukampiw apnaqapxta, ukax jiwasan ukampiw nayraqat imatäski, ukatx modelo ukarux he best hiperparámetro ukampiw yatichapxta. Métrica de deriva uñakipañatakixa, Databricks ukaxa métricas de perfil uka jakhthapitawa, ukaxa tabla primaria ukhamaraki línea base ukanakataki. Aka chiqanx conjunto de datos de validación ukanakat uñakipt'apxasmawa . tabla Primaria ukat tabla de base #read the prediction table df_base_table = spark.table(f"{catalog_name}.{model_schema}.predictions") # create the model id and add it to the table model_id = get_model_id(endpoint_name) df_base_table = df_base_table.withColumn(model_id_col, F.lit(model_id)) #write the new table and enable the CDC on it output_base_table_name = f"{catalog_name}.{model_schema}.{base_table_prefix}_{model_name}" df_base_table.write.format("delta").mode("overwrite").saveAsTable(output_base_table_name) spark.sql(f"ALTER TABLE {output_base_table_name} SET TBLPROPERTIES (delta.enableChangeDataFeed = true)") Jichhax jiwasan tablón de monitoreo lurañatakiw uñakipt’atätanxa. Jiwasax ukampiw luraraksna jan ukax API de Monitoreo de Lagos ukan uñt’ayatawa. Aka chiqanx payïr amtawimpiw apnaqapxta: UI # This is where we store the metric tables. output_schema_name = f"{catalog_name}.{model_schema}" try: info = lm.create_monitor( table_name=unpacked_requests_table_name, profile_type=lm.InferenceLog( timestamp_col=timestamp_col, granularities=granularities,#the aggregation window model_id_col=model_id_col, prediction_col=prediction_col, label_col=label_col, problem_type=problem_type, ), output_schema_name=output_schema_name, schedule=None, # We will refresh the metrics on-demand in this notebook baseline_table_name=output_base_table_name, ) print(info) except Exception as e: # Ensure the exception was expected assert "RESOURCE_ALREADY_EXISTS" in str(e), f"Unexpected error: {e}" # Update the monitor if any parameters of this notebook have changed. lm.update_monitor( table_name=unpacked_requests_table_name, updated_params=dict( profile_type=lm.InferenceLog( timestamp_col=timestamp_col, granularities=granularities, model_id_col=model_id_col, prediction_col=prediction_col, label_col=label_col, problem_type=problem_type, ), output_schema_name=output_schema_name, schedule=None, baseline_table_name=output_base_table_name, ) ) # Refresh metrics calculated on the requests table. refresh_info = lm.run_refresh(table_name=unpacked_requests_table_name) print(refresh_info) ukatx código apnaqañ tukuyatatx mä juk’a pachaw Databricks ukax taqpach métrica ukar jakthapiñkama. Tabla de mando uñakipañatakix ukaruw sarañama, ukax sink tabla ukankiwa (mä arunx ). Mä panka uñjañamawa akhama. Quality unpacked_requests_table_name Uñakipañ ukar ch’iqt’asax t’ijtäwi, pendiente ukat nayra machaqar tukuyatanakam uñjañamawa. ukatsti ukaruw ch’iqt’añama, ukhamat dashboard ukar jist’arañataki. refresh history View Dashboard ukhamatw tabla de inferencia ( ) ukamp qalltawaytanxa, ukax proceso ukat resultado ukax ukar imañawa ukatx aka tabla ukax jiwasan tabla de base ( ) ukampiw API de monitoreo ukar pasañapa. DLM ukax sapa tabla ukan perfil métricas ukar jakthapi ( ) ukatx ukanak apnaqi deriva métricas ukar jakthapiñataki ( ) . my_endpoint_payload my_endpoint_payload_unpacked base_table_als my_endpoint_payload_unpacked_profile_metric my_endpoint_payload_unpacked_drift_metrics ¡Ukanwa jumax sarasma! jumax taqi kunatix wakiski ukax servir ukat monitoreo modelo! Jutïr t’aqapanx kunjams aka lurawix ukat ukanakamp apnaqañax uk uñacht’ayapxäma ! Databricks Assets Bundle Gitlab nayrïr thakhinak lurapxta, mä pipeline MLOps tukuyat tukuykama lurañataki, Databricks ukat Spark uka apnaqasa, Databricks ukan arquitectura de referencia ukamp irpata. Akax mä recapitulación ukawa, kuna jach’a amtawinaktix uñakipt’awayktan ukanakxa: Aka yatichäw serien nayrïr t’aqapanx : Jiwasan yatiyawinakasax bronce, qullqi ukat quri capas ukanakaruw Catálogo de Unidad ukan wakicht’apxta, mä sistema de gestión de datos estructurado ukat eficiente ukham utt’ayasa. Catálogo de Unidad ukax Arquitectura de Medallón ukatakiw utt’ayata : Jiwasax uñacht’ayapxtwa kunjams crudo datos ukax sistema ukar apkatasispa, ukhamat mä consistencia ukat calidad ukax qhipa etapas de procesamiento ukataki. Ingesting Data into Unity Catalogue : Databricks apnaqasa, mä modelo de aprendizaje automático ukar yatichapxta, ukax jiwasan conjunto de datos ukarjam wakicht’atawa, ukax suma lurawinakampiw modelo escalable ukat askinjam lurañataki. Modelo yatichaña : Modelo ukan lurawip jach’anchayañatakixa, HyperOpt ukampiwa irnaqapxta, ukhamata hiperparámetros óptimos ukanaka thaqhañataki, ukhamata chiqapa ukhamaraki suma lurañataki. Sintonización de hiperparámetros con HyperOpt : MLflow ukampiw yant’äwinakas qillqt’añataki ukat uñjañatakis apnaqapxta, mä jach’a qillqt’äwiw modelo ukan versión, métricas ukat parámetros ukanakat jasak chiqanchañataki ukat wasitat uñstayañataki. Databricks MLflow ukampiw yant’äwinak arknaqañax Aka fundamental lurawinakampi tukuyatatxa, modelo ukax jichhax primed ukhamawa despliegue ukataki. Aka payïr t’aqapanx pä componentes criticos ukanakaw sistema ukar mayacht’añatak uñakipt’atäni: : Procesamiento de lote uka phuqhaña jach’a conjuntos de datos ukan yatiyawinak lurañataki, ukax aplicaciones ukanakatakix askiwa kunjamakitix puntuación a granel ukhamarak informe periódico. Inferencia de Lotes : Modelo de tiempo real ukan utt’ayaña servir ukhamat jank’ak yatiyawinak churañataki, ukax wali wakiskiriwa aplicaciones interactivas ukat servicios ukanakataki. Inferencia en Línea (Servicio Modelo) ukhamata modelos desplegados ukanakaxa suma lurawi ukhamaraki confiabilidad uka pachana utjañapataki. Modelo Monitoreo: ¡Ukar mantañäni! Modelo de Despliegue ukax mä juk’a pachanakanwa Qhipa blog ukan saräwipax modelo de evaluación ukawa. Jichhax amuyt’añäni, uñakipt’awix lurawaytanwa ukatx jikxatawayktanwa jiwasan modelo ukax mä jach’a lurawi uñacht’ayi aka modelo de producción ukar uñtasita. Kunjamatix jiwasax (asumi) modelo ukax producción ukan apnaqañ munktanxa, taqi datos ukanak aprovechañ munapxtan kunatix utjkistu ukanakxa. Jutïr amtawix modelo ukar yatichañawa ukat yant’añawa, ukax taqpach conjunto de datos ukampiw lurasi. Ukatx jiwasan modelo ukax qhipa apnaqañatakiw ch’amanchasi, ukax jiwasan modelo campeón ukham uñstayasa. Niyakixay akax qhipa modelo ukax inferencia ukatak apnaqañ munstanxa, ukax cliente de Ingeniería de Características ukampiw modelo ukar yatichañatak apnaqapxta. Ukhamatx janiw linaje modelo ukak jasak arknaqktanti, jan ukasti esquema validación ukat características transformación (ukax utjchi ukhax) cliente ukaruw descargar. with mlflow.start_run(run_name="ALS_best_model") as run: als = ALS() # Now we set the parameters for the method als.setMaxIter(MAX_ITER)\ .setSeed(SEED)\ .setRegParam(best_params["REG_PARAM"])\ .setUserCol(COL_USER)\ .setItemCol(COL_ITEM)\ .setRatingCol(COL_LABEL)\ .setRank(best_params["RANK"]) mlflow.log_param("MAX_ITER", MAX_ITER) mlflow.log_param("RANK", best_params["RANK"]) mlflow.log_param("REG_PARAM", best_params["REG_PARAM"]) # Create the model with these parameters. model = als.fit(df_full_data) #drop predictions where users and products from the test test and didn't make it into the training set. in this case, the prediction is NaN model.setColdStartStrategy('drop') predictions = model.transform(df_full_data) signature = infer_signature(model_input = df_full_data, model_output = predictions.select(COL_LABEL)) #log the model mlflow.spark.log_model(model, model_name, sample_input=df_full_data.limit(3), signature = signature, conda_env=mlflow.spark.get_default_conda_env(), registered_model_name=f"{catalog_name}.{model_schema}.{model_name}") evaluator = RegressionEvaluator(predictionCol=COL_PRED, labelCol=COL_LABEL) rmse = evaluator.setMetricName("rmse").evaluate(predictions) mlflow.log_metric('rmse', rmse) ukhamarakiw apnaqaraksna, modelos ukanakar yatichañataki ukhamarak qillqt’añataki Feature Store jan ukax Feature Engineering APIs ukanak model_info = fe.log_model(model=model, artifact_path = model_name, flavor=mlflow.spark, training_set=fe_full_data, conda_env=mlflow.spark.get_default_conda_env(), registered_model_name= f"{catalog_name}.{model_schema}.{model_name}" ) kunapachatix API ingeniería de características uka apnaqktan ukhax modelo ukan linaje ukar Catálogo Explorer ukan uñakipt’araksnawa Jichhax modelo ukan qhanañchäwip machaqar tukuyañäni ukat mä etiqueta Campeón ukar churañäni. import time from mlflow.tracking.client import MlflowClient from mlflow.entities.model_registry.model_version_status import ModelVersionStatus client = MlflowClient() #find the latest model version model_name_path = f"{catalog_name}.{model_schema}.{model_name}" model_version_infos = client.search_model_versions(f"name ='{model_name_path}'") new_model_version = max([int(model_version_info.version) for model_version_info in model_version_infos]) #add the model and model version descirption client.update_registered_model( name=model_name_path, description="collaborative filtering using Spark mllib ALS. This model use rating table" ) client.update_model_version( name=model_name_path, version=new_model_version, description="this model is optimized Rank and REG_PARAM with Hyperopt and rmse as a loss function. trained on the full dataset" ) # assign alias client.set_registered_model_alias(model_name_path, "Champion", new_model_version) Jichhax nayrar sartañamawa ukat esquema uñakipt’añamawa kunatix modelo ukar qillqantatawa. taqi machaq lurawinakamx akham uñjañamawa : Ukax registro de modelo ukatakix irnaqäw chiqawj apnaqañax modelos ukanakar apnaqañatakix etapas ukanakaw wakisi. Alias ukanakamp apnaqañax janiw irnaqkaniti. uñakipt’añäni kunjams irnaqaski uk uñjañataki Modelo etapas Aka chiqan Inferencia Modelo ukax mä juk’a pachanakanwa Lote Puntuación ukax mä juk’a pachanakanwa Jichhax amuyt’añäni, jiwasan modelo ukar producción ukan inferencia ukar apnaqañ munapxta. Aka thakhinx modelo campeón ukaruw cargapxta ukatx ukampiw 20 películas ukan iwxt’awinakap sapa apnaqirinakatak uñstayañataki. from mlflow.spark import load_model as spark_load_model from mlflow.tracking.client import MlflowClient from create_training_set import split_data #-- set UC as model registray mlflow.set_registry_uri("databricks-uc") #-- initate mlflow client client = MlflowClient() # -- read the config file with open('config.json') as config_file: config = json.load(config_file) catalog_name = config["catalog_name"] gold_layer = config["gold_layer_name"] silver_layer = config["silver_layer_name"] user_item_table_name = config["user_item_table_name"] ft_user_item_name = config["ft_user_item_name"] model_name = config["model_name"] model_schema = config["model_schema"] #-- create the model uri model_path = f"{catalog_name}.{model_schema}.{model_name}" # --create the model_uri: there are two ways to do this # 1: using the alias (we use this*) model_version_uri = f"models:/{model_uri}@champion" # 2: using model version #champion_version = client.get_model_version_by_alias(model_uri, "champion") #model_version_uri = f"models:/{model_uri}/{champion_version.version}" # -- load the model pipline and exctract the model model_pipeline = spark_load_model(model_version_uri) model = model_pipeline.stages[0] # -- batch scoring using the the model fe_full_data, df_full_data, df_train, df_test = split_data() df_batch_input = df_full_data.drop("rating") df_scores = model.transform(df_batch_input) # --- in case you used Feature Engineering to train and register model #from databricks.feature_engineering import FeatureEngineeringClient #fe = FeatureEngineeringClient() # fe.score_batch(model_uri=f"{model_version_uri}",df = df_batch_input) ukat uñjapxasmawa pachpa yatichäw yatiyawinak apnaqapxta lote puntuación ukataki. Sistemas recomendadores ukanx mä amuyunikïkchisa, jilpach aplicación ukanx modelo uka apnaqañ munapxta yaqhip jan uñjat yatiyawinak puntuación ukar puriñapataki. Amuyt’añataki, Imaging your are Netflix ukat apnaqirin iwxt’awinakaparux uru tukuyarux machaq uñjat lista ukarjam machaqar tukuyañ munapxi. Jiwasax irnaqäw wakicht’araksnawa, ukax lote puntuación ukax mä hora específico ukanw uru tukuyanx apnaqasi. Jichhax nayrar sartañasawa ukat sapa apnaqiritak iwxt’awinak uñstayañasawa. Ukatakix sapa apnaqirinakarux 20 nayrïr yänak jikxatapxta from pyspark.sql.window import Window from pyspark.sql.functions import col, split, row_number, collect_list from pyspark.sql.functions import col, collect_list, expr, lit, min, row_number, desc windowSpec = Window.partitionBy("user_id").orderBy(col("prediction").desc()) df_top_20_items = df_scores.withColumn("rank", row_number().over(windowSpec)).filter(col("rank") <= 20) df_user_recs = df_top_20_items.groupBy("user_id") \ .agg(collect_list(col("item_id").cast("double")).alias("top_item_ids")) akax kunjams uka resultadox uñstaski ukawa Tukuyañatakix predicción ukax mä etiqueta delta ukhamaw jiwasan UC ukan imasispa jan ukax sistemas descendentes Mongo DB jan ukax Azure Cosmos DB ukar uñt’ayaraksnawa. Jiwasax firs opción ukampiw sarapxta df_user_recs.write.mode("overwrite").saveAsTable(f"{catalog_name}.{output_schema}.top20_item_recommendations") Streaming/Inferencia en línea ukax mä juk’a pachanakanwa Jichhax mä caso amuyt’añäni, kawkhantix chiqpach pachan apnaqirinakan mayacht’asiwiparjamax jiwasan iwxt’awinakas machaqar tukuyañ munapxta. Aka tuqitx modelo serving ukampiw apnaqasispa. Kunawsatix maynix modelo ukar apnaqañ munki ukhax servidor ukarux datos ukanakaw apayanispa. Ukatx servidor ukax uka datos ukarux modelo desplegado ukarux manq’ayaraki, ukax lurawiruw sararaki, datos ukarux uñakipi, ukatx mä predicción ukaw lurasi. Ukanakax aplicaciones web, apps móviles jan ukax sistemas embebidos ukanakan apnaqasispawa. Aka amtawin mä apnaqawipax A/B yant’awinakatakix ruteo de tráfico ukar yanapt’añawa. Algoritmo ALS ukax janiw chiqak inferencia en línea ukatakix apnaqatäkaspati kunatix modelo ukar wasitat yatichañax wakisiwa taqpach datos (nayra + machaq) apnaqasa, iwxt’awinak machaqar tukuyañataki. Algoritmos de aprendizaje de Gradiente Descenso ukax modelo ukan uñacht’ayatawa, ukax internet tuqin machaq uñstawinakatakiw apnaqasispa. Inas jutïr post ukanx yaqhip uka algoritmos ukanakax uñakipt’araksna. Ukampirus, kunjams ukham modelox irnaqaspa uk uñacht’ayañatakik, mä (inútil) modelo lurapxta, ukax punto final ukar yanapt’i, ukax películas ukan calificación ukarjam uñt’ayi, kunapachatix mä usuariox mä películas ukar uñt’ayi! import requests model_path = f"{catalog_name}.{model_schema}.{model_name}" champion_version = client.get_model_version_by_alias(model_path, "champion") # Set the name of the MLflow endpoint endpoint_name = config["model_serving_endpoint_name"] # Name of the registered MLflow model model_name = model_path # Specify the type of compute (CPU, GPU_SMALL, GPU_MEDIUM, etc.) workload_type = "CPU" # Specify the scale-out size of compute (Small, Medium, Large, etc.) workload_size = "Small" # Get the latest version of the MLflow model model_version = int(champion_version.version) # Specify Scale to Zero(only supported for CPU endpoints) scale_to_zero = False # Get the API endpoint and token for the current notebook context API_ROOT = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiUrl().get() API_TOKEN = dbutils.notebook.entry_point.getDbutils().notebook().getContext().apiToken().get() data = { "name": endpoint_name, "config": { "served_models": [ { "model_name": model_name, "model_version": int(model_version), "workload_size": workload_size, "scale_to_zero_enabled": scale_to_zero, "workload_type": workload_type, } ] }, } headers = {"Context-Type": "text/json", "Authorization": f"Bearer {API_TOKEN}"} response = requests.post( url=f"{API_ROOT}/api/2.0/serving-endpoints", json=data, headers=headers ) Akax lurani ukat chika uru modelo serving cluster jiwasataki ukhamat mä juk’a pachaw munasispa. Jichhax ventana jist’arañax tukuyañ chiqawj uñjañama. Serving mä tukuyañ chiqaw walja modelo ukar yanapt’añatakix apnaqaraksna. Ukatx ruteo de tráfico ukampiw escenarios ukanakatakix apnaqaraksna, kunjamatix A/B yant’awinakataki jan ukax modelos de diferencia ukan lurawip producción ukan uñakipt’araksnawa. Tabla de Inferencia ukax mä juk’a pachanakanwa Tablas de inferencias ukanakax Databricks Model Serving ukanx mä registro automático ukhamaw jiwasan modelos desplegados ukanakatakix irnaqapxi. Kunawsatix ch’amanchatäki ukhax mantanir mayiwinak (datos ukanakax yatiyañatak apayata), modelo correspondiente ukan mistuwinakap (predicciones) ukat yaqhip yaqha metadatos ukanakax mä tabla Delta ukham Catálogo de Unidad ukan katjapxi. Jiwasax tabla de inferencias ukanak apnaqaraksnawa, , , ukatx mä procedimiento de recogida de datos ukampiw modelos ukanakar jan ukax . uñakipañataki ukhamarak depuración linaje ukar arknaqañataki wasitat yatichañataki suma askichañataki Jiwasax ukax jiwasan punto final de servicio ukanx modelo ukar uñjañatakis ch’amancharaksnawa. Jiwasax luraraksnawa, ukan propiedades ukanakax payload ukan uñacht’ayasa, kunapachatix nayrïr kutix tukuyañ chimpu lurasktan ukhaxa. Jan ukax tukuyañ chiqarux qhipat machaqar tukuytan kamachimpi ukat tukuyañ URL ukampiw akham (juk’amp inference table auto_capture_config put config ) aka chiqan data = { "served_models": [ { "model_name": model_name, "model_version": int(model_version), "workload_size": workload_size, "scale_to_zero_enabled": scale_to_zero, "workload_type": workload_type, } ], "auto_capture_config":{ "catalog_name": catalog_name, "schema_name": model_schema, "table_name_prefix": payload_table, } } headers = {"Context-Type": "application/json", "Authorization": f"Bearer {API_TOKEN}"} response = requests.put(url=f"{API_ROOT}/api/2.0/serving-endpoints/{endpoint_name}/config", json=data, headers=headers) print(json.dumps(response.json(), indent=4)) jichhax mä qawqha simulacro usuario interacción datos ukanakamp tukuyañ chiqar manq’ayañäni import random import time all_items = df_full_data.select(col("item_id")).distinct() for user_id in range(50,54): items_not_rated_by_user = df_full_data.where(col("user_id")==user_id).select(col("item_id")).distinct()#collect()[0][0] no_rated_items = [item.item_id for item in all_items.subtract(items_not_rated_by_user).limit(4).collect()] data = { "dataframe_records": [ {"user_id":user_id, "item_id":no_rated_items[0], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[1], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[2], "rating": random.randint(1, 5)}, {"user_id":user_id, "item_id":no_rated_items[2], "rating": random.randint(1, 5)}, ] } response = requests.post( url=f"{API_ROOT}/serving-endpoints/{endpoint_name}/invocations", json=data, headers=headers ) # generate the data within the timespan of 1 to 8 minutes time.sleep(random.randint(60*1, 60*8)) Jiwasax tukuyañ qillqatanakax tabla ukan uñakipt’araksnawa. Ukax niya 10 minutos ukjaw tabla ukan datos ukanakax uñjasi. <catalog>.<schema>.<payload_table> table_path = f"{catalog_name}.{model_schema}.{payload_table}" # Read data from the inference table df_inf_table = spark.read.table(table_path) display(df_inf_table ) jumax uñjañamawa kunatix aka payload tabla Aka inferencia tabla ukan esquema uñakipañatakixa, “Unidad catálogo inferencia tabla esquema==” uñakipt’aña .== aka tuqina Modelo Uñakipañataki Modelo ukat datos uñakipaña mä complejo tema ukax walja pachaw munasispa masterización ukataki. Databricks Lakehouse Monitoring (DLM) ukaxa jisk’acharakiw mä sistema de monitoreo aski lurañatakixa, plantillas estándar ukhamaraki personalizables ukanakampi, casos de uso común ukanakataki. Ukampirus DLM ukat modelo de monitoreo general ukar masterización ukax walja experimentaciones ukanakaw wakisi. Janiw nayax mä jach’a uñakipäw modelo de monitoreo uka tuqit aka chiqan churañ munkti jan ukasti mä qalltañ chiqawj churañ munsma. Nayax jutïrinx mä blog uka tuqitw dedicañ munta. Mä jisk’a qhanañcht’awi DLM lurawinakata ukhamaraki lurawinakata Jichhax jiwasan modelo ukax utt’ayatawa, ukatx tabla de inferencias generadas por nuestro punto final de servicio ukampiw métricas claves ukanakar uñjañatakix kunjamatix modelo ukan lurawipa ukhamarak deriva ukax kuna desviaciones jan ukax anomalías ukanakas jiwasan datos jan ukax modelo ukan pachapar uñt’añataki. Aka proactivo uñakipañaxa yanapt’istuwa pachaparu chiqañchañataki, kunjamatixa modelo ukaru mayampi yatichaña jan ukaxa lurawinakapa machaqar tukuyaña, ukhamata suma lurañataki ukhamaraki alineación ukampi negocio amtanakampi. DLM ukaxa kimsa kasta uñakipaña jan ukaxa churaraki: , ukatxa . Niyakejjay jiwasan tabla de inferencias ukanakas uñakipañ munstanjja, qhepatjja, jukʼampejj uñakiptʼtanjja. Mä tabla uñakipañataki apnaqañataki - jiwasana “ ”, jiwasaxa uñjañasawa tabla ukaxa chiqapa lurawimpi. sapa filaxa mä mayiwimpi chikachasiñawa aka columnanakampi: profile type Serie de Tiempo Instantánea Inferencia tabla primaria ukatakixa , Tabla de inferencia modelo ukan uñacht’awinakapa modelo ukan yatiyawipa modelo ukax id : inferencia mayiwi pacha chimpu pacha chimpu (opcional) . uraqi chiqapa ukax wali askiwa kunawsatix walja modelos ukar yanapt’ktan ukhax sapa modelo ukan lurawip mä tablón de monitoreo ukan uñakipañ munapxta. Mä id modelo ukat sipans juk’ampi utjki ukhaxa, DLM ukax datos ukar ch’iyjañatakiw apnaqi ukatx sapa t’aqatakix métricas ukat estáticas ukanakax sapa mayniruw jakthapi. Id modelo DLM ukax sapa estadísticas ukat métricas ukanak mä específico tiempo intervalo ukatakiw jakthapi. Inferencia uñakipañatakixa, columna de ukampi apnaqatarakiwa, ukatxa mä usuario definido ventana tama ukampiwa pacha ventanas uñt’ayañataki. juk’ampi aka amparamp qillqt’ata. sello de tiempo DLM ukax pä tablas de inferencia ukatakix yanapt’i: “ ” jan ukax “ ”. Ukax yaqhip wakiskir métricas ukat estadísticas ukanak aka especificación ukarjam jakthapi. problem type clasificación regresión DLM apnaqañatakix mä monitor lurañasawa ukatx mä tabla ukar apkatañasawa. Kunawsatix aka DLM lurasktan ukhax pä lurasi : metric tables : aka tabla ukaxa suma qhananchatawa min, max, porcentaje de nulo ukatxa ceros. Ukhamaraki, yaqha métricas ukanakaw utjaraki, ukax jan walt’äw kasta apnaqirin qhanañchataparjama. Amuyt’añataki , ukat ukax modelos de clasificación ukatakiw, ukatx ukatx ukax modelos de regresión ukatakiw. tabla métrica de perfil precisión recall f1_score mean_squared_error mean_average_error : ukax estadísticas ukaniwa, ukax kunjams datos ukan jaljawipax mayjt’awayi jan ukax mä . Ukax medidas ukanakaw jakthapita, kunjamatix Chi-cuadrado prueba, KS prueba. tabla métrica de deriva pachaparjama valor de base ukar uñtasita (ukax churatarakiwa) sapa tabla ukan phuqhat métricas ukan lista uñakipañatakix documentación ukan uñakipaña. Ukhamaraki, ukax Monitor métrica tabla lurasispa. mä métricas personalizadas ukanakaw Mä wakiskir aspecto sistema de monitoreo lurañatakix jiwasan dashboard de monitoreo ukax qhipa inferencia ukan yatiyawinakap puriñapatakiw uñjañapa. Ukatakix apnaqasispa, ukhamat tabla de inferencias ukan procesadas filas ukanakar uñjañataki. Jiwasax modelo serving ukan inferencia tabla ukarux jiwasan fuente tabla ukham apnaqapxta ( ), ukatx monitoreo tabla ukarux sink tabla ukham apnaqapxta ( ). Ukhamaraki, (CDC) ukax panpacha tabla ukan ch’amanchatawa (Tabla de Inferencia ukanx nayraqat ch’amanchatawa). Ukhamatx mayjt’awinakak lurapxta - insert/update/delete - ukax phuqhat tabla ukanx taqpach tabla ukar sapa machaqar tukuyañan wasitat lurañat sipansa. Delta tabla streaming ukampiw readStream writeStream Cambio de Datos Captura Amparampi lurata Jiwasan tabla de inferencias ukan uñjañax ch’amanchañatakix aka lurawinakampiw lurasi: Tabla de inferencia ukax mä tabla Streaming ukham uñakipt’atawa Mä machaq tabla delta luraña chiqa esquema ukampiwa despaquete tabla de inferencia ukaxa jiwasana modelo servidor punto final ukampi luratawa. Tabla de base wakicht’aña (ukaxa utjchi ukhaxa) . Mä monitor luraña tabla resultado ukatxa métrica ukar machaqar tukuyaña Mä irnaqawi wakicht’aña, ukhamata tabla de inferencia ukaxa chiqapa lurawiru apsuñataki ukhamaraki métricas ukanaka machaqar tukuyañataki Nayraqatax Lakehouse Monitoring API ukar uñt’ayañaw wakisi. Ukaxa nayraqata uñstayatawa kunatixa Databricks rum time 15.3 LTS ukatxa juk’ampi: %pip install "https://ml-team-public-read.s3.amazonaws.com/wheels/data-monitoring/a4050ef7-b183-47a1-a145-e614628e3146/databricks_lakehouse_monitoring-0.4.14-py3-none-any.whl" dbutils.library.restartPython() Tabla de inferencia ukax mä tabla de streaming ukham uñakipt’añäni requests_raw = spark.readStream\ .format("delta")\ .table(inference_table_name) requests_raw.isStreaming #-> True Ukxarusti, tabla ukarux chiqap formato ukar uchañaw wakisi kunjamatix nayraqat qhanañchawayktan ukhama. Aka tabla ukaxa sapa yatiyawitakixa mä fila ukaniñaparakiwa wakiskiri uñacht’awinakampi ukhamaraki yatiyawi chimpumpi. Tabla de inferencia ukax modelo servicio de punto final ukat apsutawa, ukax punto final mayiwinak ukat jaysawinak mä formato JSON anidado ukham imañataki. Akax mä uñacht’äwiwa JSON payload ukax mayiwi ukat jaysawinak columna ukataki. #requests {"dataframe_records": [ {"user_id": 1, "item_id": 346, "rating": 5}, {"user_id": 1, "item_id": 377, "rating": 2}, {"user_id": 1, "item_id": 302, "rating": 4} ] } #reponse {"predictions": [4.248899936676025, 1.1172138452529907, 4.279165744781494] } # --> what we need | user_id | item_id | rating | prediction | |---------|---------|--------|------------| | 1 | 346 | 5 | 4.248900 | | 1 | 377 | 2 | 1.117214 | | 1 | 302 | 4 | 4.279166 | Aka tabla chiqa esquema ukar apsuñatakix aka código apnaqaraksnawa ukax Databricks documentación ukan adaptatawa ( ). Tabla de inferencia Lakehouse Monitoring starter notebook # define the schema of the request and reponse fields in the inference tabel REQUEST_FIELDS = [StructField('user_id', IntegerType(), False),\ StructField('item_id', IntegerType(), False),\ StructField('rating', IntegerType(), False)\ ] RESPONSE_FIELD = [T.StructField("predictions", FloatType())] def process_col_requests(json_str:str) -> str: """ to proccess the JSON payload of request column in inference table """ request = json.loads(json_str) dataframe_records = request.get("dataframe_records", []) return dataframe_records def procces_col_response(json_str: str) -> str: """ to proccess the JSON payload of reponse column in inference table """ reponse = json.loads(json_str) output = [{prediction_col: round(prediction,4)} for prediction in reponse["predictions"]] return output def get_model_id(endpoint_name: str) -> str: """ create the model id by concatinating the model name and the model version. note: the assumption is the endpoint serves only one model """ served_models = get_served_models(endpoint_name) model_id = f"{served_models[0]['model_name']}_{served_models[0]['model_version']}" return model_id def process_requests(requests_raw: DataFrame, request_fields: List[T.StructField], response_field: T.StructField, endpoint_name: str) -> DataFrame: """ Takes a stream of raw requests and processes them by: - Unpacking JSON payloads for requests and responses - Exploding batched requests into individual rows - Converting Unix epoch millisecond timestamps to be Spark TimestampType :param requests_raw: DataFrame containing raw requests. Assumed to contain the following columns: - `request` - `response` - `timestamp_ms` :param request_fields: List of StructFields representing the request schema :param response_field: A StructField representing the response schema :return: A DataFrame containing processed requests """ # Convert the timestamp milliseconds to TimestampType for downstream processing. requests_timestamped = requests_raw \ .withColumn(timestamp_col, (F.col("timestamp_ms") / 1000).cast(T.TimestampType())) \ .drop("timestamp_ms") # create the model identifier column model_id = get_model_id(endpoint_name) # Convert the model name and version columns into a model identifier column. requests_identified = requests_timestamped \ .withColumn(model_id_col, F.lit(model_id)) # Rename the date column to avoid collisions with features. requests_dated = requests_identified.withColumnRenamed("date", date_col) # Consolidate and unpack JSON. request_schema = T.ArrayType(T.StructType(request_fields)) response_schema = T.ArrayType(T.StructType(response_field)) # w udf_request = F.udf(process_col_requests, request_schema) udf_reponse = F.udf(procces_col_response, response_schema) requests_unpacked = requests_dated.withColumn("request", udf_request("request")).\ withColumn("response", udf_reponse("response")) # Explode batched requests into individual rows. DB_PREFIX = "__db" requests_exploded = requests_unpacked \ .withColumn(f"{DB_PREFIX}_request_response", F.arrays_zip(F.col("request"), F.col("response"))) \ .withColumn(f"{DB_PREFIX}_request_response", F.explode(F.col(f"{DB_PREFIX}_request_response"))) \ .select(F.col("*"), F.col(f"{DB_PREFIX}_request_response.request.*"), F.col(f"{DB_PREFIX}_request_response.response.*")) \ .drop(f"{DB_PREFIX}_request_response", "request", "response") requests_cleaned = requests_exploded.drop("status_code", "sampling_fraction", "client_request_id", "databricks_request_id", "request_metadata") return requests_cleaned Uka apsut tablax akham uñtasitäspawa: Ukxarusti, jiwasan lavabo tabla ukar qalltañasawa dt_builder = DeltaTable.createIfNotExists(spark) \ .tableName(unpacked_requests_table_name) \ .addColumns(schema) \ .partitionedBy(requests_cleaned.schema) \ .property("delta.enableChangeDataFeed", "true") \ dt_builder.execute() ukatxa uka yatxatatanaka qillqaña checkpoint_path = f"dbfs:/payload-logging/{endpoint_name}/checkpoint" requests_stream = requests_cleaned.writeStream \ .trigger(once=True) \ .format("delta") \ .partitionBy(date_col) \ .outputMode("append") \ .option("checkpointLocation", checkpoint_path) \ .toTable(unpacked_requests_table_name) \ Tukuyañatakix, jiwasan tabla de base ukar uñstayapxta. DLM ukax aka tabla ukampiw derivas ukanakax jakhthapita, ukax modelos de línea base ukat primaria ukanakan pachpa columnas ukanakan jaljawip uñakipt’asa. Tabla de base ukax pachpa columna de características ukaniñapaw nayrïr columna ukhamarak pachpa columna identificación de modelo ukampi. Tabla de base ukatakix tabla de predicción ukampiw apnaqapxta, ukax jiwasan ukampiw nayraqat imatäski, ukatx modelo ukarux he best hiperparámetro ukampiw yatichapxta. Métrica de deriva uñakipañatakixa, Databricks ukaxa métricas de perfil uka jakhthapitawa, ukaxa tabla primaria ukhamaraki línea base ukanakataki. Aka chiqanx conjunto de datos de validación ukanakat uñakipt'apxasmawa . tabla Primaria ukat tabla de base #read the prediction table df_base_table = spark.table(f"{catalog_name}.{model_schema}.predictions") # create the model id and add it to the table model_id = get_model_id(endpoint_name) df_base_table = df_base_table.withColumn(model_id_col, F.lit(model_id)) #write the new table and enable the CDC on it output_base_table_name = f"{catalog_name}.{model_schema}.{base_table_prefix}_{model_name}" df_base_table.write.format("delta").mode("overwrite").saveAsTable(output_base_table_name) spark.sql(f"ALTER TABLE {output_base_table_name} SET TBLPROPERTIES (delta.enableChangeDataFeed = true)") Jichhax jiwasan tablón de monitoreo lurañatakiw uñakipt’atätanxa. Jiwasax ukampiw luraraksna jan ukax API de Monitoreo de Lagos ukan uñt’ayatawa. Aka chiqanx payïr amtawimpiw apnaqapxta: UI # This is where we store the metric tables. output_schema_name = f"{catalog_name}.{model_schema}" try: info = lm.create_monitor( table_name=unpacked_requests_table_name, profile_type=lm.InferenceLog( timestamp_col=timestamp_col, granularities=granularities,#the aggregation window model_id_col=model_id_col, prediction_col=prediction_col, label_col=label_col, problem_type=problem_type, ), output_schema_name=output_schema_name, schedule=None, # We will refresh the metrics on-demand in this notebook baseline_table_name=output_base_table_name, ) print(info) except Exception as e: # Ensure the exception was expected assert "RESOURCE_ALREADY_EXISTS" in str(e), f"Unexpected error: {e}" # Update the monitor if any parameters of this notebook have changed. lm.update_monitor( table_name=unpacked_requests_table_name, updated_params=dict( profile_type=lm.InferenceLog( timestamp_col=timestamp_col, granularities=granularities, model_id_col=model_id_col, prediction_col=prediction_col, label_col=label_col, problem_type=problem_type, ), output_schema_name=output_schema_name, schedule=None, baseline_table_name=output_base_table_name, ) ) # Refresh metrics calculated on the requests table. refresh_info = lm.run_refresh(table_name=unpacked_requests_table_name) print(refresh_info) ukatx código apnaqañ tukuyatatx mä juk’a pachaw Databricks ukax taqpach métrica ukar jakthapiñkama. Tabla de mando uñakipañatakix ukaruw sarañama, ukax sink tabla ukankiwa (mä arunx ). Mä panka uñjañamawa akhama. Quality unpacked_requests_table_name Uñakipañ ukar ch’iqt’asax t’ijtäwi, pendiente ukat nayra machaqar tukuyatanakam uñjañamawa. ukatsti ukaruw ch’iqt’añama, ukhamat dashboard ukar jist’arañataki. refresh history View Dashboard ukhamatw tabla de inferencia ( ) ukamp qalltawaytanxa, ukax proceso ukat resultado ukax ukar imañawa ukatx aka tabla ukax jiwasan tabla de base ( ) ukampiw API de monitoreo ukar pasañapa. DLM ukax sapa tabla ukan perfil métricas ukar jakthapi ( ) ukatx ukanak apnaqi deriva métricas ukar jakthapiñataki ( ) . my_endpoint_payload my_endpoint_payload_unpacked base_table_als my_endpoint_payload_unpacked_profile_metric my_endpoint_payload_unpacked_drift_metrics ¡Ukanwa jumax sarasma! jumax taqi kunatix wakiski ukax servir ukat monitoreo modelo! Jutïr t’aqapanx kunjams aka lurawix ukat ukanakamp apnaqañax uk uñacht’ayapxäma ! Databricks Assets Bundle Gitlab
