Partager via

Utiliser tables en ligne pour la diffusion de fonctionnalités en temps réel

  • Article

Important

Les tables en ligne sont en version publique préliminaire dans les régions suivantes : westus, eastus, eastus2, northeurope, westeurope. Pour plus d’informations sur la tarification, consultez Tables tarification en ligne.

Un table en ligne est une copie en lecture seule d’un Table Delta stocké dans un format orienté ligne optimisé pour l’accès en ligne. Les tables en ligne sont entièrement serverless tables qui mettent à l’échelle automatiquement la capacité de débit en fonction de la charge de requêtes et fournissent un accès à faible latence et à un débit élevé aux données de toute échelle. Les tables en ligne sont conçus pour fonctionner avec Mosaic AI Model Serving, Feature Serving et les applications de génération augmentée par récupération (RAG). where Elles sont utilisées pour les recherches rapides de données.

Vous pouvez également utiliser des tables en ligne dans des requêtes à l’aide de lakehouse Federation. Lorsque vous utilisez Lakehouse Federation, vous devez utiliser un entrepôt SQL sans serveur pour accéder aux tablesen ligne. Seules les opérations de lecture (SELECT) sont autorisées. Cette fonctionnalité est destinée à des fins interactives ou de débogage uniquement et ne doit pas être utilisée pour les charges de travail critiques pour la mission.

La création d’un table en ligne à l’aide de l’interface utilisateur Databricks est un processus en une seule étape. Faites juste select le Delta table dans Catalog Explorer et selectcréez tableen ligne. Vous pouvez également utiliser l’API REST ou le Kit de développement logiciel (SDK) Databricks pour créer et gérer des tablesen ligne. Consultez Travailler avec en ligne tables à l’aide d’API.

Spécifications

  • L’espace de travail doit être activé pour Unity Catalog. Suivez la documentation pour créer un metastore Unity Catalog, l’activer dans un espace de travail et créer un Catalog.
  • Un modèle doit être inscrit dans Unity Catalog pour accéder en ligne à tables.

Travailler avec des tables en ligne à l’aide de l’interface utilisateur

Cette section explique comment créer et supprimer des tablesen ligne, et comment vérifier l’état et déclencher les mises à jour des tablesen ligne.

Créer un table en ligne à l’aide de l’interface utilisateur

Vous créez un table en ligne à l’aide de l’Explorateur Catalog. Pour plus d’informations sur les autorisations requises, consultez Autorisations utilisateur.

  1. Pour créer un tableen ligne, le table Delta source doit avoir une clé primaire. Si le table Delta que vous souhaitez utiliser n’a pas de clé primaire, créez-en une en suivant les instructions suivantes : Utiliser un table Delta existant dans Unity Catalog en tant que fonctionnalité table.

  2. Dans l'Explorateur Catalog, accédez à la source table que vous souhaitez sync vers un tableen ligne. Dans le menu Créer, selectEn ligne table.

    select créer des table en ligne

  3. Utilisez les sélecteurs dans la boîte de dialogue pour configurer le tableen ligne.

    configurer en ligne la boîte de dialogue table

    Nom: nom à utiliser pour l'table en ligne dans Unity Catalog.

    clé(s) primaires: Columndans la source table à utiliser comme clé(s) primaires dans le tableen ligne.

    Clé de série chronologique : (facultative). Column dans le table source à utiliser comme clé de série chronologique. Lorsque c'est spécifié, le table en ligne inclut uniquement la ligne avec la dernière valeur de la clé de séries temporelles pour chaque clé primaire.

    Sync mode: spécifie comment le pipeline de synchronisation met à jour tableen ligne. Select l’un de instantané, déclenché, ou continu.

    Stratégie Description
    Snapshot Le pipeline s’exécute une fois pour prendre un instantané du table source et le copier dans le tableen ligne. Les modifications ultérieures apportées à la table source sont automatiquement reflétées dans le table en ligne en prenant un nouvel instantané de la source et en créant une nouvelle copie. Le contenu du table en ligne est mis à jour de manière atomique.
    Déclenchée Le pipeline s’exécute une fois pour créer une copie d’instantané initiale du table source dans le tableen ligne. Contrairement au mode instantané sync, lorsque le table en ligne est actualisé, seules les modifications depuis la dernière exécution du pipeline sont récupérées et appliquées en ligne au table. L'incrémentiel refresh peut être déclenché manuellement ou automatiquement selon un horaire.
    Continue Le pipeline s’exécute en continu. Les modifications ultérieures apportées à la source table sont appliquées de manière incrémentielle au table en ligne en mode de diffusion en temps réel. Aucun manuel refresh n’est nécessaire.

Remarque

Pour prendre en charge le mode déclenché ou le mode continu sync, la source table doit avoir le flux de données de modification activé .

  1. Lorsque vous avez terminé, cliquez sur Confirmer. La page en ligne table s’affiche.
  2. La nouvelle table en ligne est créée sous les catalog, schemaet le nom tels que spécifiés dans la boîte de dialogue de création. Dans Catalog Explorer, le table en ligne est indiqué par l'icône en ligne table.

Get mettre à jour le statut et déclencher des mises à jour à l'aide de l'interface utilisateur

Pour vérifier l’état du tableen ligne, cliquez sur le nom du table dans le Catalog pour l’ouvrir. La page en ligne s’affiche avec l’onglet Vue d’ensemble ouvert. La section d'ingestion de données affiche l’état du dernier update. Pour déclencher un update, cliquez sur Sync maintenant. La section Data Ingest inclut également un lien vers le pipeline Delta Live Tables qui met à jour le table.

vue en ligne de la page table dans catalog

Planifier des mises à jour périodiques

Pour le mode tables en ligne avec instantané ou le mode déclenché sync, vous pouvez planifier des mises à jour périodiques automatiques. L'horaire update est géré par le pipeline Delta Live Tables qui met à jour le table.

  1. Dans Catalog Explorer, accédez à tableen ligne.
  2. Dans la section Ingestion des données, cliquez sur le lien vers le pipeline.
  3. Dans le coin supérieur droit, cliquez sur Horaire, et ajoutez un nouvel horaire ou gérez les horaires existants update.

Supprimer un table en ligne à l’aide de l’interface utilisateur

Dans la page de table en ligne, selectSupprimer du menu Menu kebab kebab.

Travailler avec des tables en ligne à l’aide d’API

Vous pouvez également utiliser le Kit de développement logiciel (SDK) Databricks ou l’API REST pour créer et gérer des tablesen ligne.

Pour obtenir des informations de référence, consultez la documentation de référence pour le KIT de développement logiciel (SDK) Databricks pour Python ou l’API REST.

Spécifications

SDK Databricks version 0.20 ou ultérieure.

Créer un table en ligne à l’aide d’API

Kit de développement logiciel (SDK) Databricks – Python

from pprint import pprint
from databricks.sdk import WorkspaceClient
from databricks.sdk.service.catalog import *

w = WorkspaceClient(host='https://xxx.databricks.com', token='xxx')

# Create an online table
spec = OnlineTableSpec(
  primary_key_columns=["pk_col"],
  source_table_full_name="main.default.source_table",
  run_triggered=OnlineTableSpecTriggeredSchedulingPolicy.from_dict({'triggered': 'true'})
)

online_table = OnlineTable(
  name="main.default.my_online_table",  # Fully qualified table name
  spec=spec  # Online table specification
)

w.online_tables.create_and_wait(table=online_table)

API REST

curl --request POST "https://xxx.databricks.com/api/2.0/online-tables" \
--header "Authorization: Bearer xxx" \
--data '{
    "name": "main.default.my_online_table",
    "spec": {
        "run_triggered": {},
        "source_table_full_name": "main.default.source_table",
        "primary_key_columns": ["a"]
    }
  }'

Le table en ligne démarre automatiquement la synchronisation après sa création.

Get état et déclenchement refresh à l’aide des API

Vous pouvez consulter l’état et les caractéristiques du table en ligne, comme dans l’exemple ci-dessous. Si votre table en ligne n’est pas continue et que vous souhaitez déclencher une refresh manuelle de ses données, vous pouvez utiliser l’API de pipeline pour le faire.

Utilisez l'ID de pipeline associé au table en ligne dans la spécification en ligne table et lancez une nouvelle update sur le pipeline pour déclencher le refresh. Cela équivaut à cliquer sur Sync maintenant dans l’interface utilisateur de table en ligne dans Catalog Explorer.

Kit de développement logiciel (SDK) Databricks – Python

pprint(w.online_tables.get('main.default.my_online_table'))

# Sample response
OnlineTable(name='main.default.my_online_table',
    spec=OnlineTableSpec(perform_full_copy=None,
        pipeline_id='some-pipeline-id',
        primary_key_columns=['pk_col'],
        run_continuously=None,
        run_triggered={},
        source_table_full_name='main.default.source_table',
        timeseries_key=None),
    status=OnlineTableStatus(continuous_update_status=None,
        detailed_state=OnlineTableState.PROVISIONING,
        failed_status=None,
        message='Online Table creation is '
            'pending. Check latest status in '
            'Delta Live Tables: '
            'https://xxx.databricks.com/pipelines/some-pipeline-id',
        provisioning_status=None,
        triggered_update_status=None))

# Trigger an online table refresh by calling the pipeline API. To discard all existing data
# in the online table before refreshing, set "full_refresh" to "True". This is useful if your
# online table sync is stuck due to, for example, the source table being deleted and recreated
# with the same name while the sync was running.
w.pipelines.start_update(pipeline_id='some-pipeline-id', full_refresh=True)

API REST

curl --request GET \
  "https://xxx.databricks.com/api/2.0/online-tables/main.default.my_online_table" \
  --header "Authorization: Bearer xxx"

# Sample response
{
  "name": "main.default.my_online_table",
  "spec": {
    "run_triggered": {},
    "source_table_full_name": "main.default.source_table",
    "primary_key_columns": ["pk_col"],
    "pipeline_id": "some-pipeline-id"
  },
  "status": {
    "detailed_state": "PROVISIONING",
    "message": "Online Table creation is pending. Check latest status in Delta Live Tables: https://xxx.databricks.com#joblist/pipelines/some-pipeline-id"
  }
}

# Trigger an online table refresh by calling the pipeline API. To discard all existing data
# in the online table before refreshing, set "full_refresh" to "True". This is useful if your
# online table sync is stuck due to, for example, the source table being deleted and recreated
# with the same name while the sync was running.
curl --request POST "https://xxx.databricks.com/api/2.0/pipelines/some-pipeline-id/updates" \
  --header "Authorization: Bearer xxx" \
  --data '{
    "full_refresh": true
  }'

Supprimer un table en ligne à l’aide d’API

Kit de développement logiciel (SDK) Databricks – Python

w.online_tables.delete('main.default.my_online_table')

API REST

curl --request DELETE \
  "https://xxx.databricks.com/api/2.0/online-tables/main.default.my_online_table" \
  --header "Authorization: Bearer xxx"

La suppression du table en ligne arrête toute synchronisation continue des données et libère toutes ses ressources.

Servir des données table en ligne à l’aide d’un point de terminaison de service de fonctionnalité

Pour les modèles et les applications hébergés en dehors de Databricks, vous pouvez créer un point de terminaison pour la diffusion de fonctionnalités à partir de tablesen ligne. Le point de terminaison rend les fonctionnalités disponibles à faible latence à l’aide d’une API REST.

  1. Créer une spécification de fonctionnalité.

    Lorsque vous créez une spécification de fonctionnalité, vous spécifiez la source Delta table. Cela permet à la spécification de fonctionnalité d’être utilisée dans les scénarios hors connexion et en ligne. Pour les recherches en ligne, le point de terminaison de service utilise automatiquement le table en ligne pour effectuer des recherches de fonctionnalités à faible latence.

    Le Delta source table et l'entité en ligne table doivent utiliser la même clé primaire.

    La spécification de fonctionnalité peut être consultée sous l’onglet de la fonction dans l'Explorateur Catalog.

    from databricks.feature_engineering import FeatureEngineeringClient, FeatureLookup

fe = FeatureEngineeringClient()
fe.create_feature_spec(
  name="catalog.default.user_preferences_spec",
  features=[
    FeatureLookup(
      table_name="user_preferences",
      lookup_key="user_id"
    )
  ]
)

  2. Créer un point de terminaison de mise en service de fonctionnalités.

    Cette étape suppose que vous avez créé un table en ligne nommé user_preferences_online_table qui synchronise les données du Delta tableuser_preferences. Utilisez la spécification de fonctionnalité pour créer un point de terminaison de mise en service de fonctionnalités. Le point de terminaison rend les données disponibles via une API REST à l’aide du tableen ligne associé.

    Remarque

    L’utilisateur qui effectue cette opération doit être le propriétaire des table hors connexion et des tableen ligne.

    Kit de développement logiciel (SDK) Databricks – Python

    from databricks.sdk import WorkspaceClient
from databricks.sdk.service.serving import EndpointCoreConfigInput, ServedEntityInput

workspace = WorkspaceClient()

# Create endpoint
endpoint_name = "fse-location"

workspace.serving_endpoints.create_and_wait(
  name=endpoint_name,
  config=EndpointCoreConfigInput(
    served_entities=[
      ServedEntityInput(
        entity_name=feature_spec_name,
        scale_to_zero_enabled=True,
        workload_size="Small"
      )
    ]
  )
)

    API Python

    from databricks.feature_engineering.entities.feature_serving_endpoint import (
  ServedEntity,
  EndpointCoreConfig,
)

fe.create_feature_serving_endpoint(
  name="user-preferences",
  config=EndpointCoreConfig(
    served_entities=ServedEntity(
      feature_spec_name="catalog.default.user_preferences_spec",
      workload_size="Small",
      scale_to_zero_enabled=True
    )
  )
)

  3. Get données du point d'accès de service de fonctionnalités.

    Pour accéder au point de terminaison de l’API, envoyez une requête HTTP GET à l’URL du point de terminaison. L’exemple montre comment effectuer cette opération avec des API Python. Pour d’autres langages et outils, consultez Service de fonctionnalités.

    # Set up credentials
export DATABRICKS_TOKEN=...

    url = "https://{workspace_url}/serving-endpoints/user-preferences/invocations"

headers = {'Authorization': f'Bearer {DATABRICKS_TOKEN}', 'Content-Type': 'application/json'}

data = {
  "dataframe_records": [{"user_id": user_id}]
}
data_json = json.dumps(data, allow_nan=True)

response = requests.request(method='POST', headers=headers, url=url, data=data_json)
if response.status_code != 200:
  raise Exception(f'Request failed with status {response.status_code}, {response.text}')

print(response.json()['outputs'][0]['hotel_preference'])

Utiliser des tables en ligne avec des applications RAG

Les applications RAG représentent un cas d’usage courant des tablesen ligne. Vous créez un table en ligne pour les données structurées dont a besoin l’application RAG, et vous l’hébergez sur un serveur de fonctionnalités. L’application RAG utilise le point de terminaison de service de fonctionnalité pour rechercher les données pertinentes de la tableen ligne.

Les étapes typiques sont les suivantes :

  1. Créer un point de terminaison de mise en service de fonctionnalités.
  2. Créer un outil à l’aide de LangChain ou d’un package similaire qui utilise le point de terminaison pour rechercher des données pertinentes.
  3. Utiliser l’outil dans un agent LangChain ou un agent similaire pour récupérer les données pertinentes.
  4. Créer un point de terminaison de service de modèles pour héberger l’application.

Pour obtenir des instructions pas à pas et un exemple de notebook, consultez Exemple de caractérisation : application RAG structurée.

Exemples de Notebook

Le bloc-notes suivant montre comment publier des caractéristiques vers tables en ligne pour une diffusion en temps réel et la recherche automatique de caractéristiques.

Bloc-notes de démonstration en ligne tables

Get notebook

Utiliser tables en ligne avec Mosaic AI Service de Modèle

Vous pouvez utiliser tables en ligne pour rechercher les fonctionnalités de Mosaic AI Model Serving. Lorsque vous sync une fonctionnalité table à un tableen ligne, les modèles formés à l’aide des fonctionnalités de cette fonctionnalité table recherchent automatiquement les fonctionnalités values à partir du table en ligne pendant l’inférence. Aucune configuration supplémentaire n’est nécessaire.

  1. Utiliser un FeatureLookup pour effectuer l'apprentissage du modèle.

    Pour l’entraînement du modèle, utilisez des caractéristiques de la fonction hors ligne table durant l’entraînement du modèle set, comme illustré dans l’exemple suivant :

    training_set = fe.create_training_set(
  df=id_rt_feature_labels,
  label='quality',
  feature_lookups=[
      FeatureLookup(
          table_name="user_preferences",
          lookup_key="user_id"
      )
  ],
  exclude_columns=['user_id'],
)

  2. Servez le modèle avec le service de modèles Mosaic AI. Le modèle recherche automatiquement les caractéristiques du tableen ligne. Consultez Recherche automatique de caractéristiques avec le service de modèles Databricks pour plus d’informations.

Autorisations utilisateur

Vous devez disposer des autorisations suivantes pour créer un tableen ligne :

  • SELECT privilège sur la tablesource.
  • Privilège USE_CATALOG sur la destination catalog.
  • USE_SCHEMA et privilège CREATE_TABLE sur le schemade destination.

Pour gérer le pipeline de synchronisation des données d’un tableen ligne, vous devez être le propriétaire du table en ligne ou qu'on vous accorde le privilège REFRESH sur le tableen ligne. Les utilisateurs qui n’ont pas de privilèges USE_CATALOG et USE_SCHEMA sur le catalog ne verront pas le table en ligne dans Catalog'Explorateur.

Le metastore Unity Catalog doit avoir Privilege Model version 1.0.

Modèle d’autorisation de point de terminaison

Un principal de service unique est automatiquement créé pour un service de fonctionnalité ou un point de terminaison de service de modèles avec des autorisations limitées nécessaires pour interroger des données en ligne à partir de tables. Ce principal de service permet aux points de terminaison d’accéder aux données indépendamment de l’utilisateur qui a créé la ressource et garantit que le point de terminaison peut continuer à fonctionner si le créateur quitte l’espace de travail.

La durée de vie de ce principal de service est la durée de vie du point de terminaison. Les journaux d’audit peuvent indiquer les enregistrements générés par le système pour le propriétaire de l'Unity Catalogcatalog afin d’accorder les privilèges nécessaires à cette entité de service.

Limites

  • Une seule table en ligne est prise en charge par la source table.
  • Un table en ligne et son table source peuvent avoir au maximum 1 000 columns.
  • Columns des types de données ARRAY, MAP ou STRUCT ne peuvent pas être utilisés comme clés primaires dans le tableen ligne.
  • Si un column est utilisé comme clé primaire dans l'tableen ligne, toutes les lignes de la source tablewhere que l'column contient des valeurs nulles pour values sont ignorées.
  • Les systèmes étrangers, internes et tables ne sont pas pris en charge en tant que sources tables.
  • Les sources tables sans flux de données de modifications Delta activé ne prennent en charge que le mode Snapshot sync.
  • Les tables de partage Delta ne sont pris en charge que dans le mode instantanésync.
  • Les noms de Catalog, schemaet table des table en ligne peuvent seulement contenir des caractères alphanumériques et des traits de soulignement, et ne doivent pas commencer par des chiffres. Les tirets (-) ne sont pas autorisés.
  • Columns de type String sont limitées à 64 Ko de longueur.
  • Les noms Column sont limités à 64 caractères.
  • La taille maximale de la ligne est de 2 Mo.
  • La taille combinée de toutes les tables en ligne dans un metastore Unity Catalog pendant la préversion publique est de 2 To de données utilisateur non compressées.
  • Le nombre maximal de requêtes par seconde (QPS) est de 12 000. Contactez votre équipe de gestion de compte Databricks pour augmenter la limit.

Dépannage

Je ne vois pas l’option Créer en ligne table

Généralement, la cause est que le table à partir duquel vous essayez de sync (la source table) n'est pas un type pris en charge. Vérifiez que le type sécurisable de la source table(illustré dans l'onglet CatalogDétails de l'Explorateur ) correspond à l'une des options prises en charge ci-dessous :

  • TABLE_EXTERNAL
  • TABLE_DELTA
  • TABLE_DELTA_EXTERNAL
  • TABLE_DELTASHARING
  • TABLE_DELTASHARING_MUTABLE
  • TABLE_STREAMING_LIVE_TABLE
  • TABLE_STANDARD
  • TABLE_FEATURE_STORE
  • TABLE_FEATURE_STORE_EXTERNAL
  • TABLE_VIEW
  • TABLE_VIEW_DELTASHARING
  • TABLE_MATERIALIZED_VIEW

Je ne peux pas selectnon plus activer les modes de déclenchement ousync continu lorsque je crée un table en ligne

Ce problème survient si la source table n'a pas activé le flux de données de modification Delta ou s'il s'agit d'une vue ou d'une vue matérialisée. Pour utiliser le mode incrémentiel sync, activez le flux de données modifiées sur la source tableou utilisez une non-vue table.

Le tableupdate en ligne échoue ou affiche l’état hors connexion

Pour commencer à résoudre cette erreur, cliquez sur l'identifiant du pipeline qui apparaît dans l'onglet Vue d’ensemble de l'table en ligne dans l'Catalog Explorateur.

tables échec de pipeline en ligne

Dans la page d’interface utilisateur du pipeline qui s’affiche, cliquez sur l’entrée indiquant « Failed to resolve flow __online_table ».

message d’erreur en ligne de pipeline tables

Une fenêtre contextuelle s’affiche avec des détails dans la section Error details.

tables détails de l’erreur

Les causes courantes des erreurs sont les suivantes :

  • La source table a été supprimée, ou supprimée puis recréée avec le même nom, tandis que la synchronisation du table en ligne était en cours. Cela est particulièrement courant avec les tablesen ligne continue, car ils sont constamment synchronisés.

  • La source table ne peut pas être accessible via le calcul serverless en raison des paramètres de pare-feu. Dans ce cas, la section Error details peut afficher le message d’erreur « Failed to start the DLT service on cluster xxx… ».

  • La taille agrégée des tables en ligne dépasse 2 To (taille non compressée) à l'échelle du metastore limit. La taille de 2 To de la limit fait référence à la taille non compressée après l'expansion du Delta table au format orienté ligne. La taille du table au format ligne peut être nettement supérieure à la taille du table Delta affiché dans Catalog Explorer, qui fait référence à la taille compressée du table dans un format orienté column. La différence peut être aussi grande que 100x, en fonction du contenu du table.

    Pour estimer la taille non compressée et développée par ligne d’un Delta table, utilisez la requête suivante à partir d'un entrepôt SQL sans serveur. La requête retourne la taille de table développée estimée en octets. L’exécution réussie de cette requête confirme également que le calcul sans serveur peut accéder à la source table.

    SELECT sum(length(to_csv(struct(*)))) FROM `source_table`;