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.

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

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.

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.

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.