Snowpipe Streaming¶

Snowpipe Streaming API versus Snowpipe¶

Le API est destiné à compléter Snowpipe, pas à le remplacer. Utilisez Snowpipe Streaming API dans les scénarios de diffusion en streaming où les données sont diffusées via des lignes (par exemple, les sujets Apache Kafka) plutôt qu’écrites dans des fichiers. Le API s’intègre dans un flux d’ingestion qui comprend une application Java personnalisée existante qui produit ou reçoit des enregistrements. Le API supprime la nécessité de créer des fichiers pour charger les données dans les tables Snowflake, et permet le chargement automatique et continu des flux de données dans Snowflake au fur et à mesure que les données sont disponibles.

Le tableau suivant décrit les différences entre Snowpipe Streaming et Snowpipe :

Exigences du logiciel¶

Canaux¶

Le API ingère des lignes à travers un ou plusieurs canaux. Un canal représente une connexion de streaming logique et nommée avec Snowflake pour le chargement de données dans une table. Un seul canal correspond à une seule table dans Snowflake ; toutefois, plusieurs canaux peuvent pointer vers la même table. Le client SDK a la possibilité d’ouvrir plusieurs canaux vers plusieurs tables ; mais le client SDK ne peut pas ouvrir de canaux sur plusieurs comptes. L’ordre des lignes et les jetons de décalage correspondants sont préservés au sein d’un canal, mais pas entre les canaux qui pointent vers la même table.

Les canaux sont conçus pour durer longtemps lorsqu’un client insère activement des données, et ils doivent être réutilisés, car les informations de jeton de décalage sont conservées. Les données à l’intérieur du canal sont automatiquement effacées toutes les secondes, par défaut, et n’ont pas besoin d’être fermées. Pour plus d’informations, voir Latence.

Vous pouvez choisir de supprimer définitivement des canaux à l’aide de l’API DropChannelRequest lorsque le canal et les métadonnées de décalage associées ne sont plus nécessaires. Il existe deux façons de supprimer un canal :

• Suppression d’un canal à la fermeture. Les données à l’intérieur du canal sont automatiquement vidées avant la suppression du canal.

• Suppression d’un canal à l’aveugle. Cela n’est pas recommandé car la suppression d’un canal à l’aveugle supprime toutes les données en attente et peut rendre invalides tous les canaux déjà ouverts.

Vous pouvez lancer la commande SHOW CHANNELS pour obtenir la liste des chaînes pour lesquelles vous disposez de privilèges d’accès. Pour plus d’informations, consultez SHOW CHANNELS.

Jetons de décalage¶

Un jeton de décalage est une chaîne de caractères qu’un client peut inclure dans les demandes de méthode insertRow (une seule ligne) ou insertRows (ensemble de lignes) pour suivre la progression de l’ingestion sur la base par canal. Le jeton est initialisé à la valeur NULL lors de la création du canal et est mis à jour lorsque les lignes avec un jeton de décalage fourni sont validées dans Snowflake par le biais d’un processus asynchrone. Les clients peuvent périodiquement faire des demandes de méthode getLatestCommittedOffsetToken pour obtenir le dernier jeton de décalage engagé pour un canal particulier et l’utiliser pour raisonner sur la progression de l’ingestion. Notez que ce jeton n’est pas utilisé par Snowflake pour effectuer la déduplication ; cependant, les clients peuvent utiliser ce jeton pour effectuer la déduplication en utilisant votre code personnalisé.

Lorsqu’un client rouvre un canal, le dernier jeton de décalage persistant est renvoyé. Le client peut réinitialiser sa position dans la source de données en utilisant le jeton pour éviter d’envoyer deux fois les mêmes données. Notez que lorsqu’un événement de réouverture d’un canal se produit, toutes les données mises en mémoire tampon dans Snowflake sont écartées pour éviter de les engager.

Vous pouvez utiliser le dernier jeton de décalage engagé pour effectuer les opérations suivantes :

• Suivre la progression de l’ingestion.

• Vérifier si un décalage spécifique a été validé en le comparant au dernier jeton de décalage validé.

• Avancer le décalage de la source et purger les données qui ont déjà été engagées.

• Permettre la déduplication et assurer la livraison des données « exactly-once ».

Par exemple, le connecteur Kafka pourrait lire un jeton de décalage à partir d’un sujet tel que <partition>:<décalage>, ou simplement <décalage> si la partition est encodée dans le nom du canal. Prenons la requête suivante :

1. Le connecteur Kafka est en ligne et ouvre un canal correspondant à Partition 1 dans le sujet Kafka T avec le nom de canal T:P1.

2. Le connecteur commence à lire les enregistrements de la partition Kafka.

3. Le connecteur appelle le API, en faisant une demande de méthode insertRows avec le décalage associé à l’enregistrement comme jeton de décalage.

   Par exemple, le jeton de décalage pourrait être 10, faisant référence au dixième enregistrement de la partition Kafka.

4. Le connecteur effectue périodiquement des getLatestCommittedOffsetToken demandes de méthode pour déterminer la progression de l’ingestion.

Si le connecteur Kafka tombe en panne, la procédure suivante peut être complétée pour reprendre la lecture des enregistrements à partir du décalage correct pour la partition Kafka :

1. Le connecteur Kafka revient en ligne et rouvre le canal en utilisant le même nom que précédemment.

2. Le connecteur appelle le API, en faisant une demande de méthode getLatestCommittedOffsetToken pour obtenir le dernier décalage engagé pour la partition.

   Par exemple, supposons que le dernier jeton de décalage persistant est 20.

3. Le connecteur utilise la lecture Kafka APIs pour réinitialiser un curseur correspondant au décalage plus 1 (21 dans cet exemple).

4. Le connecteur reprend la lecture des enregistrements. Aucune donnée en double n’est récupérée après le repositionnement réussi du curseur de lecture.

Dans un autre exemple, une application lit les journaux d’un répertoire et utilise le client Snowpipe Streaming SDK pour exporter ces journaux vers Snowflake. Vous pouvez construire une application d’exportation de journaux qui fait ce qui suit :

1. Lister les fichiers dans le répertoire du journal.

   Supposons que le framework de journalisation génère des fichiers journaux qui peuvent être ordonnés lexicographiquement et que les nouveaux fichiers journaux sont placés à la fin de cet ordre.

2. Lit un fichier journal ligne par ligne et appelle les requêtes de la méthode API, effectuant insertRows avec un jeton de décalage correspondant au nom du fichier journal et au nombre de lignes ou à la position des octets.

   Par exemple, un jeton de décalage pourrait être messages_1.log:20, où messages_1.log est le nom du fichier journal et 20 est le numéro de ligne.

Si l’application tombe en panne ou doit être redémarrée, elle appelle alors le API, en faisant une demande de méthode getLatestCommittedOffsetToken pour récupérer un jeton de décalage qui correspond à au dernier fichier journal et à la dernière ligne exportés. En continuant avec l’exemple, cela pourrait être messages_1.log:20. L’application ouvrirait alors messages_1.log et chercherait la ligne 21 pour éviter que la même ligne de journal soit ingérée deux fois.

Meilleures pratiques en matière de livraison « exactly-once »¶

Il peut être difficile d’assurer une livraison en une seule fois, et il est essentiel de respecter les principes suivants dans votre code personnalisé :

• Pour garantir une récupération appropriée en cas d’exceptions, de défaillances ou de pannes, vous devez toujours rouvrir le canal et redémarrer l’ingestion en utilisant le dernier jeton de décalage validé.

• Bien que votre application puisse maintenir son propre décalage, il est crucial d’utiliser le dernier jeton de décalage engagé fourni par Snowflake comme source de vérité et de réinitialiser votre propre décalage en conséquence.

• Le seul cas où votre propre décalage doit être considéré comme la source de vérité est lorsque le jeton de décalage de Snowflake est défini ou réinitialisé sur une valeur NULL. Un jeton de décalage NULL signifie généralement l’une des choses suivantes :

  • Il s’agit d’un nouveau canal, aucun jeton de décalage n’a donc été défini.

  • La table cible a été supprimée et recréée, de sorte que le canal est considéré comme nouveau.

  • Aucune activité d’ingestion n’ayant eu lieu pendant 30 jours, le canal a été automatiquement nettoyé et les informations relatives au jeton de décalage ont été perdues.

• Si nécessaire, vous pouvez périodiquement purger les données source qui ont déjà été validées sur la base du dernier jeton de décalage validé et avancer votre propre décalage.

• Si le schéma de la table est modifié lorsque les canaux Snowpipe Streaming sont actifs, le canal doit être rouvert. Le connecteur Snowflake Kafka gère ce scénario automatiquement, mais si vous utilisez le SDK de Snowflake Ingest directement, vous devez rouvrir le canal vous-même.

Pour plus d’informations sur la façon dont le connecteur Kafka avec Snowpipe Streaming réalise la livraison « exactly-once », consultez Sémantique unique et exacte.

Charger des données dans des tables Apache Iceberg™¶

Avec la version 3.0.0 et ultérieures du SDK Snowflake Ingest, Snowpipe Streaming peut ingérer des données dans les tables Apache Iceberg gérées par Snowflake. Le SDK Snowpipe Ingest Java prend en charge le chargement dans les tables Snowflake standard (non-Iceberg) et les tables Iceberg.

Pour plus d’informations, voir Utiliser Snowpipe Streaming avec des tables Apache Iceberg™.

Latence¶

Snowpipe Streaming efface automatiquement les données des canaux toutes les secondes. Vous n’avez pas besoin de fermer le canal pour que les données soient vidées.

Avec les versions de SDK 2.0.4 et ultérieures de Snowflake Ingest, vous pouvez configurer la latence via l’option MAX_CLIENT_LAG.

• Pour les tables Snowflake standard (non-Iceberg), la valeur par défaut de MAX_CLIENT_LAG est de 1 seconde.

• Pour les tables Iceberg (prises en charge par la version 3.0.0 et ultérieures du SDK Snowflake Ingest), la valeur par défaut de MAX_CLIENT_LAG est de 30 secondes.

La latence maximale peut être fixée à 10 minutes. Pour plus d’informations, consultez MAX_CLIENT_LAG et les configurations de latence recommandées pour Snowpipe Streaming.

Notez que le connecteur Kafka pour Snowpipe Streaming a son propre tampon. Une fois que le temps de vidage de tampon Kafka est atteint, les données seront envoyées avec une seconde de latence à Snowflake par le biais de Snowpipe Streaming. Pour plus d’informations, voir temps.vidage.tampon.

Migration vers des fichiers optimisés¶

Le API écrit les lignes des canaux dans les blobs du stockage cloud, qui sont ensuite transférés dans la table cible. Initialement, les données streamées écrites dans une table cible sont stockées dans un format de fichier intermédiaire temporaire. À ce stade, la table est considérée comme une « table mixte », car les données partitionnées sont stockées dans un mélange de fichiers natifs et intermédiaires. Un processus automatisé en arrière-plan fait migrer les données des fichiers intermédiaires actifs vers des fichiers natifs optimisés pour les requêtes et les opérations DML, selon les besoins.

Réplication¶

Le Streaming Snowpipe prend en charge la réplication et le basculement des tables Snowflake remplies par Snowpipe Streaming et ses décalages de canaux associés d’un compte source vers un compte cible dans différentes régions et sur les plateformes Cloud.

Pour plus d’informations, voir Réplication et Streaming Snowpipe.

Opérations à insertion uniquement¶

L” API est actuellement limité à l’insertion de lignes. Pour modifier, supprimer ou combiner des données, écrivez les enregistrements «  bruts «  dans une ou plusieurs tables mises en zone de préparation. Fusionnez, joignez ou transformez les données en utilisant des pipelines de données continus pour insérer les données modifiées dans les tableaux de reporting de destination.

Classes et interfaces¶

Pour la documentation sur les classes et les interfaces, voir API SDK Snowflake Ingest.

Types de données Java pris en charge¶

Le tableau suivant résume les types de données Java pris en charge pour l’ingestion dans les colonnes Snowflake :

Privilèges d’accès requis¶

L’appel de l’API de Snowpipe Streaming nécessite un rôle avec les privilèges suivants :

Limitations¶

Snowpipe Streaming prend uniquement en charge l’utilisation de clés AES 256 bits pour le chiffrement des données.

Si le clustering automatique est également activé sur la même table que celle dans laquelle Snowpipe Streaming insère, les coûts de calcul pour la migration des fichiers peuvent être réduits. Pour plus d’informations, voir Snowpipe Streaming : meilleures pratiques.

Les objets ou types suivants ne sont pas pris en charge :

• Les types de données GEOGRAPHY et GEOMETRY

• Colonnes avec classements

• Tables TRANSIENT ou TEMPORARY

• Tables avec l’une des options de colonne suivantes :

  • AUTOINCREMENT ou IDENTITY

  • Valeur par défaut de la colonne qui n’est pas NULL.

• Le nombre total de canaux par table ne peut pas dépasser 10 000. Nous vous recommandons de réutiliser les canaux si nécessaire. Contactez le support Snowflake si vous devez ouvrir plus de 10 000 canaux par table.