Source de fichier Azure Blob Storage avec Azure Queue Storage (hérité)
Cette documentation a été retirée et pourrait ne pas être mise à jour. Les produits, services ou technologies mentionnés dans ce contenu ne sont plus pris en charge. Voir Qu'est-ce que Auto Loader ?
Le connecteur ABS-AQS fournit une source de fichiers optimisée qui utilise Azure Queue Storage (AQS) pour trouver de nouveaux fichiers écrits dans un conteneur de stockage Azure Blob (ABS) sans lister de manière répétée tous les fichiers. Cela offre deux avantages :
- Latence réduite : pas besoin de lister les structures de répertoires imbriquées sur ABS, ce qui est lent et gourmand en ressources.
- Coûts réduits : plus de requêtes d'API LIST coûteuses effectuées vers ABS.
La source ABS-AQS supprime les messages de la file d'attente AQS au fur et à mesure qu'elle consomme les événements. Si vous souhaitez que d'autres pipelines consomment des messages de cette file d'attente, configurez une file d'attente AQS distincte pour le lecteur optimisé. Vous pouvez configurer plusieurs abonnements Event Grid pour publier sur différentes files d'attente.
Utilisez la source de fichier ABS-AQS
Pour utiliser la source de fichier ABS-AQS, vous devez :
-
Configurez les notifications d'événements ABS en tirant parti des abonnements Azure Event Grid et acheminez-les vers AQS. Consultez Réagir aux événements de stockage Blob.
-
Spécifiez les options
fileFormatetqueueUrlet un schéma. Par exemple :Pythonspark.readStream \
.format("abs-aqs") \
.option("fileFormat", "json") \
.option("queueName", ...) \
.option("connectionString", ...) \
.schema(...) \
.load()
S'authentifier auprès d'Azure Queue Storage et de Blob Storage
Pour s'authentifier auprès de Stockage de files d'attente Azure et Stockage Blob, utilisez des jetons Shared Access Signature (SAS) ou des clés de compte de stockage. Vous devez fournir une chaîne de connexion pour le compte de stockage où votre file d'attente est déployée qui contient votre jeton SAS ou vos clés d'accès à votre compte de stockage. Pour plus d'informations, consultez Configurer les chaînes de connexion du stockage Azure.
Vous devrez également fournir un accès à vos conteneurs de stockage Azure Blob. Consultez Connectez-vous à Azure Data Lake Storage et Blob Storage pour plus d'informations sur la configuration de l'accès à votre conteneur de stockage Azure Blob.
Databricks vous recommande fortement d'utiliser Gérer les secrets pour fournir vos chaînes de connexion.
Configuration
Option | Type | Par défaut | Description |
|---|---|---|---|
allowOverwrites | Booléen |
| Si un blob qui est écrasé doit être retraité. |
connectionString | Chaîne | Aucun (paramètre requis) | La chaîne de connexion pour accéder à votre file d'attente. |
fetchParallelism | Entier | 1 | Nombre de threads à utiliser lors de la récupération de messages du service de file d'attente. |
format de fichier | Chaîne | Aucun (paramètre requis) | Le format des fichiers tels que |
ignoreFileDeletion | Booléen |
| Si vous disposez de configurations de cycle de vie ou si vous supprimez manuellement les fichiers source, vous devez définir cette option sur |
maxFileAge | Entier | 604 800 | Détermine la durée (en secondes) pendant laquelle les notifications de fichier sont stockées en tant qu'état pour éviter les traitements en double. |
pathRewrites | Une chaîne JSON. |
| Si vous utilisez des points de montage, vous pouvez réécrire le préfixe du chemin |
queueFetchInterval | Une chaîne de durée, par exemple, |
| Combien de temps attendre entre les extractions si la file d'attente est vide. Azure facture chaque requête API à AQS. Par conséquent, si les données n'arrivent pas fréquemment, cette valeur peut être définie sur une longue durée. Tant que la file d'attente n'est pas vide, nous récupérerons en continu. Si de nouveaux fichiers sont créés toutes les 5 minutes, vous voudrez peut-être définir une valeur |
queueName | Chaîne | Aucun (paramètre requis) | Le nom de la file d'attente AQS. |
Si vous observez beaucoup de messages dans les logs du Driver qui ressemblent à Fetched 0 new events and 3 old events., où vous avez tendance à observer beaucoup plus d'anciens événements que de nouveaux, vous devriez réduire l'intervalle de Trigger de votre stream.
Si vous consommez des fichiers à partir d'un emplacement sur le stockage Blob où vous vous attendez à ce que certains fichiers puissent être supprimés avant d'être traités, vous pouvez définir la configuration suivante pour ignorer l'erreur et poursuivre le traitement :
spark.sql("SET spark.sql.files.ignoreMissingFiles=true")
Questions fréquemment posées (FAQ)
Si ignoreFileDeletion est False (default) et que l'objet a été supprimé, cela fera-t-il échouer l'ensemble du pipeline ?
Oui, si nous recevons un événement indiquant que le fichier a été supprimé, cela fera échouer tout le pipeline.
Comment dois-je définir maxFileAge?
Azure Queue Storage fournit une sémantique de distribution de messages au moins une fois, nous devons donc conserver l’état pour la déduplication. Le paramètre default pour maxFileAge est de 7 jours, ce qui correspond à la durée de vie maximale (TTL) d’un message dans la file d’attente.