Einführung des Databricks Connector, einer gut beleuchteten Lösung zur Optimierung der Migration und Transformation unstrukturierter Daten
Mit der rasanten Entwicklung von KI- und Machine-Learning-Technologien (AI/ML) sind Vektoreinbettungen zur bevorzugten Methode geworden, um unstrukturierte Daten zu indexieren und semantische Suchen durchzuführen.
Die KI-gestützte Suche umfasst typischerweise zwei unterschiedliche Phasen: Offline-Datenindexierung und Online-Abfragebereitstellung, was die Nutzung verschiedener Technologie-Stacks erforderlich macht. Der Versuch, Daten effizient und nahtlos vom Offline-Verarbeitungs-Stack zum Online-Bereitstellungs-Stack zu übertragen, kann jedoch herausfordernd sein. Das neueste Release von Zilliz Cloud führt einen Databricks Connector ein, eine gut durchdachte Lösung zur Optimierung dieses Prozesses durch die Integration von Apache Spark/Databricks und Milvus/Zilliz Cloud.
In diesem Beitrag stellen wir diese Integration vor, untersuchen ihre praktische Nutzung in realen Szenarien und zeigen Ihnen Schritt für Schritt, wie Sie sie verwenden.
Wie der Databricks Connector funktioniert und seine Anwendungsfälle
Spark ist bekannt für seine Fähigkeit, Daten in großem Maßstab zu verarbeiten, sowie für seine Kompetenz im Bereich Machine Learning. Andererseits zeichnet sich Milvus durch die effiziente Verarbeitung und Suche von Vektoreinbettungen aus, die von Machine-Learning-Modellen erzeugt werden. Die Kombination dieser beiden leistungsstarken Technologien erleichtert die Entwicklung modernster Anwendungen in Bereichen wie Generative AI, Empfehlungssystemen sowie Bild- und Videosuche.
Die Übertragung von Daten von Spark nach Milvus ist eine häufige Aufgabe beim Aufbau KI-gestützter Suche, erfordert jedoch oft komplexen Glue Code im Backend des Suchsystems. Der Spark-Milvus Connector vereinfacht diesen Prozess und reduziert ihn auf einen einzigen Funktionsaufruf innerhalb des Spark-Programms.
Dieser Connector erweist sich in verschiedenen Szenarien als vorteilhaft.
Batch-Datenimport
Teams mit Machine-Learning-Expertise aktualisieren ihre Einbettungsmodelle häufig, um die neuesten Forschungsergebnisse einzubeziehen. Nach jedem Upgrade des Einbettungsmodells, bei dem der gesamte Datenkorpus durch den Spark-Job erneut verarbeitet und ein neuer Satz von Vektoren generiert werden muss, benötigen Teams oft benutzerdefinierten „Glue Code“, einen dedizierten Service oder einen weiteren Spark-Job, um diese neuen Vektoren in den Bereitstellungs-Stack zu integrieren. Mit dem Databricks Connector wird diese Aufgabe jedoch so einfach wie dem Job Zugriff zum Schreiben in den S3-Bucket von Milvus zu gewähren (oder in einen temporären Bucket bei der Nutzung von Zilliz Cloud). Dieser optimierte Prozess ermöglicht es dem vektorgenerierenden Spark-Job, Daten direkt über einen einfachen Aufruf einer Hilfsfunktion in die Milvus-Instanz zu laden.
Iteratives Einfügen
Auch Nutzer, die nicht in großem Maßstab arbeiten, können davon profitieren, Spark-DataFrame-Datensätze mithilfe des Spark-Milvus Connectors direkt in Milvus einzufügen. Dieser Ansatz spart den Aufwand, Code zum Herstellen von Verbindungen und API-Aufrufe zu schreiben, und macht den Integrationsprozess reibungsloser.
So verwenden Sie den Databricks Connector
Dieser Abschnitt demonstriert die Verwendung des Databricks Connectors zur Optimierung von Datenmigration und -transformation.
Iteratives Einfügen eines Spark Dataframes
Mit der Spark-Milvus-Verbindung war das Streamen von Daten von Spark nach Milvus noch nie einfacher. Sie können Daten jetzt direkt von Spark nach Milvus mit Sparks nativer Dataframe-API übertragen. Derselbe Code funktioniert auch für Databricks und Zilliz (vollständig verwaltetes Milvus). Hier ist ein Codeausschnitt, der diesen Ansatz demonstriert:
// 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()
Eine Collection per Batch laden
Wir empfehlen die Verwendung der Funktion `MilvusUtils. bulkInsertFromSpark ()` für Situationen, in denen Sie große Datenmengen effizient übertragen müssen. Dieser Ansatz ist äußerst effizient für die Verarbeitung riesiger Datensätze.
Ansatz für Milvus
Die Integration umfasst S3- oder MinIO-Buckets für selbst gehostete Milvus-Instanzen als internen Speicher. Durch die Gewährung des Zugriffs auf Spark oder Databricks kann der Spark-Job Milvus-Connectors verwenden, um Daten im Batch in den Bucket zu schreiben und anschließend die gesamte Collection per Bulk-Insert für die Bereitstellung einzufügen.
// 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")
Ein Ansatz für Zilliz Cloud
Wenn Sie Zilliz Cloud (das verwaltete Milvus) verwenden, können Sie die praktische Data Import API nutzen. Zilliz Cloud bietet umfassende Tools und Dokumentation, die Ihnen helfen, Ihre Daten effizient aus verschiedenen Datenquellen, einschließlich Spark, zu verschieben. Durch das Einrichten eines S3-Buckets als Zwischenstation und das Gewähren des Zugriffs für Zilliz Cloud lädt die Data Import API Daten nahtlos aus dem S3-Bucket in die Vektordatenbank.
Bevor Sie diese Integration ausführen, müssen Sie die Spark-Laufzeitumgebung laden, indem Sie dem Databricks-Cluster eine jar-Datei hinzufügen. Es gibt verschiedene Möglichkeiten, eine Bibliothek zu installieren. Der folgende Screenshot zeigt das Hochladen einer jar-Datei von lokal in den Cluster.
Weitere Details zur Installation einer Bibliothek im Databricks-Arbeitsbereich finden Sie in der offiziellen Dokumentation von Databricks.
Bulk-Insert erfordert das Speichern von Daten in einem temporären Bucket, damit Zilliz Cloud sie stapelweise importieren kann. Sie können einen S3-Bucket erstellen und ihn als externen Speicherort von Databricks konfigurieren. Details finden Sie in dieser Dokumentation. Um das Sicherheitsrisiko Ihrer Zilliz Cloud-Anmeldeinformationen zu verringern, können Sie diese sicher in Databricks verwalten, indem Sie den Databricks-Anweisungen folgen.
Hier ist ein Codeausschnitt, der den Batch-Datenmigrationsprozess zeigt. Ähnlich wie im obigen Milvus-Beispiel müssen Sie lediglich die Anmeldeinformationen und die S3-Bucket-Adresse ersetzen.
// Schreibe die Daten im Batch in den Milvus-Bucket-Speicher.
val outputPath = "s3://my-temp-bucket/result"
df.write
.mode("overwrite")
.format("mjson")
.save(outputPath)
// Gib die Milvus-Optionen an.
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))
// Spark-Ausgabedateien per Bulk Insert in Milvus einfügen
MilvusUtils.bulkInsertFromSpark(spark, targetMilvusOptions, outputPath, "mjson")
Alles zusammenführen: ein Notebook-Beispiel, das Sie durch den gesamten Prozess führt
Damit Sie schnell loslegen können, haben wir ein Notebook-Beispiel vorbereitet, das Sie durch die Streaming- und Batch-Datenübertragungsprozesse mit Milvus und Zilliz Cloud führt.
Fazit
Die Integration von Spark und Milvus eröffnet spannende Möglichkeiten für KI-gestützte Anwendungen. Mit unserem optimierten Ansatz für Datenportabilität können Entwickler Daten mühelos von Spark/Databricks nach Milvus/Zilliz Cloud übertragen, entweder in Echtzeit oder im Batch-Modus. Diese Integration ermöglicht es Ihnen, effiziente und skalierbare KI-Lösungen zu entwickeln und das volle Potenzial dieser leistungsstarken Technologien auszuschöpfen.
Bereit, Ihre KI-Reise zu beginnen? Starten Sie kostenlos mit Zilliz Cloud – ganz ohne Installationsaufwand und ohne Kreditkarte.
Weiterlesen
3 Easiest Ways to Use Claude Code on Your Mobile Phone
Run Claude Code from your phone with Remote Control, Happy Coder, or SSH + Tailscale. Comparison table, setup steps, and tools for typing, memory, and parallel tasks.
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.
Why Teams Are Migrating from Weaviate to Zilliz Cloud — and How to Do It Seamlessly
Explore how Milvus scales for large datasets and complex queries with advanced features, and discover how to migrate from Weaviate to Zilliz Cloud.