Snowpipe Streaming¶

Snowpipe Streaming API vs 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 via 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.

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, reportez-vous à 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é ou non 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 du 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. 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 10ème enregistrement de la partition Kafka.

3. 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, le flux suivant peut être complété 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.

Autre exemple, supposons que vous ayez une application qui lit les journaux d’un répertoire et exporte ces journaux vers Snowflake à l’aide du client Snowpipe Streaming SDK. 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 correspondant 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 est supprimée et recréée, de sorte que le canal est considéré comme nouveau.

  • S’il n’y a pas d’activité d’ingestion dans un canal pendant 30 jours, le canal est automatiquement nettoyé, ce qui entraîne la perte des informations relatives aux jetons de décalage.

• 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.

Pour plus d’informations sur la façon dont le connecteur Kafka avec Snowpipe Streaming réalise la livraison « exactly-once », reportez-vous à Livraison « exactly-once ».

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 :

Limites¶

• Les tables avec un des paramètres de colonne suivants ne sont pas prises en charge :

  • AUTOINCREMENT ou IDENTITY

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

• Les types de données GEOGRAPHY et GEOMETRY ne sont pas pris en charge.

• Les colonnes avec classements ne sont pas prises en charge.

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

• Les tables TRANSIENT ou TEMPORARY ne sont pas prises en charge.