Présentation du connecteur Databricks, une solution éclairée pour rationaliser la migration et la transformation des données non structurées
Avec l’évolution rapide des technologies d’IA et de machine learning (AI/ML), les vector embeddings sont devenus la méthode de choix pour indexer des données non structurées et effectuer des recherches sémantiques.
La recherche pilotée par l’IA comprend généralement deux phases distinctes : l’indexation hors ligne des données et le service des requêtes en ligne, ce qui nécessite l’utilisation de piles technologiques différentes. Cependant, tenter de transférer efficacement les données de manière transparente de la pile de traitement hors ligne vers la pile de service en ligne peut s’avérer difficile. La dernière version de Zilliz Cloud introduit un Databricks Connector, une solution bien pensée pour rationaliser ce processus en intégrant Apache Spark/Databricks et Milvus/Zilliz Cloud.
Dans cet article, nous présenterons cette intégration, explorerons son utilisation pratique dans des scénarios réels et vous guiderons dans son utilisation.
Fonctionnement du Databricks Connector et ses cas d’utilisation
Spark est réputé pour sa capacité à traiter des données à grande échelle et pour son efficacité en machine learning. De son côté, Milvus excelle dans la gestion et la recherche efficaces des vector embeddings générés par des modèles de machine learning. La combinaison de ces deux technologies puissantes facilite le développement d’applications de pointe dans des domaines tels que l’IA générative, les systèmes de recommandation et la recherche d’images et de vidéos.
Le transfert de données de Spark vers Milvus est une tâche courante dans la création d’une recherche alimentée par l’IA, mais il implique souvent un code de liaison complexe dans le backend du système de recherche. Le connecteur Spark-Milvus simplifie ce processus, le réduisant à un seul appel de fonction dans le programme Spark.
Ce connecteur s’avère utile dans divers scénarios.
Importation de données par lots
Les équipes dotées d’une expertise en machine learning mettent souvent à jour leurs modèles d’embedding afin d’intégrer les dernières avancées de la recherche. Après chaque mise à niveau du modèle d’embedding, qui nécessite que l’ensemble du corpus de données soit retraité par la tâche Spark et qu’un nouvel ensemble de vecteurs soit généré, les équipes ont souvent besoin d’un « code de liaison » personnalisé, d’un service dédié ou d’une autre tâche Spark pour intégrer ces nouveaux vecteurs dans la pile de service. Cependant, avec le Databricks Connector, cette tâche devient aussi simple que d’accorder à la tâche l’accès en écriture au bucket S3 de Milvus (ou à un bucket éphémère lors de l’utilisation de Zilliz Cloud). Ce processus rationalisé permet à la tâche Spark générant des vecteurs de charger les données directement dans l’instance Milvus au moyen d’un simple appel à une fonction utilitaire.
Insertion itérative
Les utilisateurs qui ne travaillent pas à grande échelle peuvent également bénéficier de l’insertion directe des enregistrements Spark DataFrame dans Milvus à l’aide du connecteur Spark-Milvus. Cette approche évite d’avoir à écrire du code d’établissement de connexion et des appels d’API, ce qui rend le processus d’intégration plus fluide.
Comment utiliser le Databricks Connector
Cette section montre comment utiliser le Databricks Connector pour rationaliser la migration et la transformation des données.
Insertion itérative d’un Spark Dataframe
Avec la connexion Spark-Milvus, le streaming de données de Spark vers Milvus n’a jamais été aussi simple. Vous pouvez désormais envoyer directement des données de Spark vers Milvus avec l’API Dataframe native de Spark. Le même code fonctionne également pour Databricks et Zilliz (Milvus entièrement géré). Voici un extrait de code qui illustre cette approche :
// Specify the target Milvus instance and vector data collection
df.write.format("milvus")
.option(MILVUS_URI, "https://in01-xxxxxxxxx.aws-us-west-2.vectordb.zillizcloud.com:19535")
.option(MILVUS_TOKEN, dbutils.secrets.get(scope = "zillizcloud", key = "token"))
.option(MILVUS_COLLECTION_NAME, "text_embedding")
.option(MILVUS_COLLECTION_VECTOR_FIELD, "embedding")
.option(MILVUS_COLLECTION_VECTOR_DIM, "128")
.option(MILVUS_COLLECTION_PRIMARY_KEY, "id")
.mode(SaveMode.Append)
.save()
Charger une collection par lots
Nous recommandons d’utiliser la fonction `MilvusUtils. bulkInsertFromSpark ()` dans les situations où vous devez transférer efficacement de grands volumes de données. Cette approche est extrêmement efficace pour gérer d’énormes jeux de données.
Approche pour Milvus
L’intégration implique des buckets S3 ou MinIO pour les instances Milvus auto-hébergées en tant que stockage interne. En accordant l’accès à Spark ou Databricks, la tâche Spark peut utiliser les connecteurs Milvus pour écrire les données dans le bucket par lots, puis insérer en bloc toute la collection pour la mise à disposition.
// Write the data in batch into the Milvus bucket storage.
val outputPath = "s3a://milvus-bucket/result"
df.write
.mode("overwrite")
.format("parquet")
.save(outputPath)
// Specify Milvus options.
val targetProperties = Map(
MilvusOptions.MILVUS_HOST -> host,
MilvusOptions.MILVUS_PORT -> port.toString,
MilvusOptions.MILVUS_COLLECTION_NAME -> targetCollectionName,
MilvusOptions.MILVUS_BUCKET -> bucketName,
MilvusOptions.MILVUS_ROOTPATH -> rootPath,
MilvusOptions.MILVUS_FS -> fs,
MilvusOptions.MILVUS_STORAGE_ENDPOINT -> minioEndpoint,
MilvusOptions.MILVUS_STORAGE_USER -> minioAK,
MilvusOptions.MILVUS_STORAGE_PASSWORD -> minioSK,
)
val targetMilvusOptions = new MilvusOptions(new CaseInsensitiveStringMap(targetProperties.asJava))
// Bulk insert Spark output files into Milvus
MilvusUtils.bulkInsertFromSpark(spark, targetMilvusOptions, outputPath, "parquet")
Une approche pour Zilliz Cloud
Si vous utilisez Zilliz Cloud (le Milvus managé), vous pouvez tirer parti de sa pratique Data Import API. Zilliz Cloud fournit des outils et une documentation complets pour vous aider à déplacer efficacement vos données depuis diverses sources de données, y compris Spark. En configurant un bucket S3 comme intermédiaire et en accordant l’accès à Zilliz Cloud, la Data Import API charge de manière transparente les données du bucket S3 vers la base de données vectorielle.
Avant d’exécuter cette intégration, vous devez charger l’environnement d’exécution Spark en ajoutant un fichier jar au cluster Databricks. Il existe différentes façons d’installer une bibliothèque. La capture d’écran ci-dessous montre le téléversement d’un jar depuis le poste local vers le cluster.
Pour plus de détails sur l’installation d’une bibliothèque dans l’espace de travail Databricks, consultez la documentation officielle de Databricks pour en savoir plus.
L’insertion en bloc nécessite de stocker les données dans un bucket temporaire afin que Zilliz Cloud puisse les importer par lots. Vous pouvez créer un bucket S3 et le configurer comme emplacement externe de Databricks. Consultez cette documentation pour plus de détails. Afin de réduire le risque de sécurité lié à vos identifiants Zilliz Cloud, vous pouvez les gérer en toute sécurité sur Databricks en suivant les instructions Databricks.
Voici un extrait de code présentant le processus de migration des données par lots. Comme dans l’exemple Milvus ci-dessus, il vous suffit de remplacer les identifiants et l’adresse du bucket S3.
// Écrire les données par lots dans le stockage de bucket Milvus.
val outputPath = "s3://my-temp-bucket/result"
df.write
.mode("overwrite")
.format("mjson")
.save(outputPath)
// Spécifier les options Milvus.
val targetProperties = Map(
MilvusOptions.MILVUS_URI -> zilliz_uri,
MilvusOptions.MILVUS_TOKEN -> zilliz_token,
MilvusOptions.MILVUS_COLLECTION_NAME -> targetCollectionName,
MilvusOptions.MILVUS_BUCKET -> bucketName,
MilvusOptions.MILVUS_ROOTPATH -> rootPath,
MilvusOptions.MILVUS_FS -> fs,
MilvusOptions.MILVUS_STORAGE_ENDPOINT -> minioEndpoint,
MilvusOptions.MILVUS_STORAGE_USER -> minioAK,
MilvusOptions.MILVUS_STORAGE_PASSWORD -> minioSK,
)
val targetMilvusOptions = new MilvusOptions(new CaseInsensitiveStringMap(targetProperties.asJava))
// Insérer en bloc les fichiers de sortie Spark dans Milvus
MilvusUtils.bulkInsertFromSpark(spark, targetMilvusOptions, outputPath, "mjson")
Tout rassembler : un exemple de notebook qui vous guide tout au long du processus
Pour vous aider à démarrer rapidement, nous avons préparé un exemple de notebook qui vous guide à travers les processus de transfert de données en streaming et par lots avec Milvus et Zilliz Cloud.
Conclusion
L’intégration de Spark et Milvus ouvre des possibilités passionnantes pour les applications alimentées par l’IA. Grâce à notre approche simplifiée de la portabilité des données, les développeurs peuvent transférer sans effort des données de Spark/Databricks vers Milvus/Zilliz Cloud, que ce soit en temps réel ou en mode par lots. Cette intégration vous permet de créer des solutions d’IA efficaces et évolutives, libérant tout le potentiel de ces technologies puissantes.
Prêt à vous lancer dans votre parcours IA ? Commencez gratuitement avec Zilliz Cloud dès aujourd’hui, sans contraintes d’installation et sans carte de crédit requise.
Continuer à lire
Zilliz Skills Breakdown: How AI Agents Master Vector Databases
Zilliz's Milvus Skill (pymilvus, 7 files) and Zilliz Cloud Skill (zilliz-cli, 14 modules) bring vector-DB dev and ops into one Claude Code session.
My Wife Wanted Dior. I Spent $600 on Claude Code to Vibe-Code a 2M-Line Database Instead.
Write tests, not code reviews. How a test-first workflow with 6 parallel Claude Code sessions turns a 2M-line C++ codebase into a daily shipping pipeline.
Zilliz Cloud Update: Tiered Storage, Business Critical Plan, Cross-Region Backup, and Pricing Changes
This release offers a rebuilt tiered storage with lower costs, a new Business Critical plan for enhanced security, and pricing updates, among other features.