Configuration des Openflow Connector for Amazon Kinesis Data Streams¶

Conditions préalables¶

1. Consulter Openflow Connector for Amazon Kinesis Data Streams

2. Assurez-vous d’avoir Configuration d’Openflow - BYOC ou de Configurer Openflow - Déploiements Snowflake.

3. Si vous utilisez OpenFlow - Déploiements Snowflake, assurez-vous d’avoir pris connaissance de la configuration des domaines requis et d’avoir accordé l’accès aux domaines requis pour le connecteur Kinesis.

Configurer les rôles et les stratégies IAM dans AWS¶

En tant qu’administrateur AWS, effectuez les actions suivantes dans votre compte AWS:

1. Créez un utilisateur ou rôle IAM AWS qu’Openflow utilisera pour accéder au flux de données Kinesis. Pour plus d’informations, voir Création d’utilisateurs IAM dans la documentation AWS.

2. Assurez-vous que l’utilisateur AWS a configuré les identifiants de connexion de la clé d’accès.

3. Accordez les autorisations IAM suivantes à l’utilisateur AWS :

   Service	Actions	Ressources (ARNs)	But
   Flux de données Amazon Kinesis	kinesis:DescribeStream, kinesis:DescribeStreamConsumer, kinesis:GetRecords, kinesis:GetShardIterator, kinesis:ListShards, kinesis:RegisterStreamConsumer	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}	Détecte les fragments, lit les enregistrements via une interrogation à débit partagé, résout le flux ARN, enregistre un consommateur Enhanced Fan-Out et interroge l’état du consommateur lors de l’enregistrement.
   Flux de données Amazon Kinesis	kinesis:DeregisterStreamConsumer, kinesis:DescribeStreamConsumer, kinesis:SubscribeToShard	arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*	Décrit, s’abonne à et désenregistre les consommateurs Enhanced Fan-Out par consommateur ARN.
   Amazon DynamoDB	dynamodb:CreateTable, dynamodb:DeleteTable, dynamodb:DescribeTable, dynamodb:GetItem, dynamodb:PutItem, dynamodb:Query, dynamodb:Scan, dynamodb:UpdateItem	arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}, arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration	Crée et gère la table des points de contrôle/baux (baux de fragments, battements de cœur des nœuds, points de contrôle) ainsi qu’une table de migration temporaire utilisée lors d’une migration ponctuelle à partir des anciennes tables de points de contrôle.

   See moreShow less

   Expand

   Exemple de stratégie IAM :

   CopyExpand

   {
       "Version": "2012-10-17",
       "Statement": [
           {
               "Sid": "KinesisStreamAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DescribeStream",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:GetRecords",
                   "kinesis:GetShardIterator",
                   "kinesis:ListShards",
                   "kinesis:RegisterStreamConsumer"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}"
           },
           {
               "Sid": "KinesisConsumerAccess",
               "Effect": "Allow",
               "Action": [
                   "kinesis:DeregisterStreamConsumer",
                   "kinesis:DescribeStreamConsumer",
                   "kinesis:SubscribeToShard"
               ],
               "Resource": "arn:aws:kinesis:${REGION}:${ACCOUNT_ID}:stream/${STREAM_NAME}/consumer/*"
           },
           {
               "Sid": "DynamoDBTableAccess",
               "Effect": "Allow",
               "Action": [
                   "dynamodb:CreateTable",
                   "dynamodb:DeleteTable",
                   "dynamodb:DescribeTable",
                   "dynamodb:GetItem",
                   "dynamodb:PutItem",
                   "dynamodb:Query",
                   "dynamodb:Scan",
                   "dynamodb:UpdateItem"
               ],
               "Resource": [
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}",
                   "arn:aws:dynamodb:${REGION}:${ACCOUNT_ID}:table/${APPLICATION_NAME}_migration"
               ]
           }
       ]
   }
   

   Show lessSee more

   Scroll to top

   Avant d’utiliser l’exemple de stratégie, remplacez les caractères de remplacement suivants :

   Caractère de remplacement	Description
   ${REGION}	Votre région AWS (par exemple, us-east-1)
   ${ACCOUNT_ID}	Votre ID de compte AWS (par exemple, 123456789012)
   ${STREAM_NAME}	La valeur du paramètre du connecteur AWS Kinesis Stream Name
   ${APPLICATION_NAME}	La valeur du paramètre du connecteur AWS Kinesis Application Name Utilisée comme nom de la table de points de contrôle DynamoDB et comme nom du consommateur enregistré Enhanced Fan-Out.

   See moreShow less

   Expand

   Note

   • La table ${APPLICATION_NAME}_migration est une table DynamoDB temporaire créée uniquement lors d’une migration unique des tables de point de contrôle héritées vers le nouveau schéma. Elle est automatiquement supprimée lorsque la migration est terminée. Si votre déploiement n’a jamais utilisé l’ancien connecteur basé sur KCL, vous pouvez omettre la table de migration ARN de la stratégie.

   • L’action dynamodb:DeleteTable est utilisée pendant le processus de migration et peut être supprimée de la stratégie une fois que la migration a été confirmée comme étant terminée.

   • L’action kinesis:DeregisterStreamConsumer est déclenchée lorsque le processeur est supprimé du canevas. Si l’entité IAM ne dispose pas de cette autorisation, le consommateur doit être désenregistré manuellement via la console AWS ou la CLI.

Paramétrage du compte Snowflake¶

En tant qu’administrateur de compte Snowflake, effectuez les tâches suivantes :

1. Créez un nouvel utilisateur du service Snowflake avec le type SERVICE.

2. Créez un nouveau rôle ou utilisez un rôle existant et accordez les privilèges de base de données.

   Le connecteur nécessite que l’utilisateur crée la table de destination. Assurez-vous que l’utilisateur dispose des privilèges nécessaires pour gérer les objets Snowflake :

   Objet	Privilège	Remarques
   Base de données	USAGE
   Schéma	USAGE
   Table	OWNERSHIP	Nécessaire pour que le connecteur ingère des données dans une table.

   See moreShow less

   Expand

   Snowflake recommande de créer un utilisateur et un rôle distincts pour chaque flux Kinesis afin d’assurer un meilleur contrôle des accès.

   Vous pouvez utiliser le script suivant pour créer et configurer un rôle personnalisé (exigence : SECURITYADMIN ou équivalent) :

   CopyExpand

   USE ROLE securityadmin;
   
   CREATE ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON DATABASE kinesis_db TO ROLE openflow_kinesis_connector_role_1;
   GRANT USAGE ON SCHEMA kinesis_schema TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Note

   Notez que les privilèges doivent être accordés directement au rôle de connecteur et ne peuvent pas être hérités.

3. Configurer la table de destination

   Nous vous recommandons fortement d’utiliser l’évolution de schéma côté serveur pour les modifications de schéma et une table d’erreurs pour la journalisation des erreurs DML.

   L’exemple ci-dessous montre comment créer une table et ajouter des autorisations OWNERSHIP.

   CopyExpand

   USE ROLE openflow_kinesis_connector_role_1;
   
   CREATE TABLE kinesis_db.kinesis_schema.<DESTINATION_TABLE_NAME> (
     kinesisMetadata object
   )
   ENABLE_SCHEMA_EVOLUTION = TRUE
   ERROR_LOGGING = TRUE;
   
   USE ROLE securityadmin;
   GRANT OWNERSHIP ON TABLE <DESTINATION_TABLE_NAME> TO ROLE openflow_kinesis_connector_role_1;
   

   Show lessSee more

   Scroll to top

   Ces connecteurs prennent en charge la détection et l’évolution automatiques des schémas. La structure des tables dans Snowflake est définie et évolue automatiquement pour prendre en charge la structure des nouvelles données chargées par le connecteur. Les clés de premier niveau du contenu de l’enregistrement seront automatiquement mappées avec les colonnes de la table correspondant par leur nom (insensibles à la casse).

   Lorsque l’évolution de schéma est activée, Snowflake peut automatiquement développer la table de destination en ajoutant de nouvelles colonnes détectées dans le flux entrant et en supprimant des contraintes NOT NULL pour prendre en charge de nouveaux modèles de données. Pour plus d’informations, consultez Évolution du schéma de la table.

   Si ENABLE_SCHEMA_EVOLUTION n’est pas activée, vous devez créer le schéma manuellement en étendant la définition de la table. Le connecteur tente de faire correspondre les clés de premier niveau du contenu de l’enregistrement aux colonnes de la table par leur nom. Si les clés du fichier JSON ne correspondent pas aux colonnes de la table, le connecteur ignore les clés.

4. (Facultatif) Configurer un gestionnaire de secrets

   Snowflake recommande vivement cette étape. Configurez un gestionnaire de secrets pris en charge par Openflow, par exemple AWS, Azure et Hashicorp, et stockez les clés publiques et privées dans le magasin de secrets.

   1. Une fois le gestionnaire de secrets configuré, déterminez comment vous vous y authentifierez. Sur AWS, il est recommandé d’utiliser le rôle de l’instance EC2 associé à Openflow, car de cette manière, aucun autre secret ne doit être conservé.

   2. Dans le canevas Openflow, configurez un fournisseur de paramètres associé à ce gestionnaire de secrets, à partir du menu hamburger en haut à droite. Naviguez jusqu’à Controller Settings » Parameter Provider, puis récupérez vos valeurs de paramètres.

   3. À ce stade, tous les identifiants peuvent être référencés avec les chemins de paramètres associés et aucune valeur sensible ne doit être conservée dans Openflow.

5. Octroyer l’accès aux utilisateurs

   Tous les autres utilisateurs de Snowflake qui ont besoin d’accéder aux données brutes ingérées par le connecteur (par exemple, pour un traitement personnalisé dans Snowflake), doivent se voir attribuer le rôle créé à l’étape 2.

Définir le connecteur¶

En tant qu’ingénieur des données, effectuez les tâches suivantes pour installer et configurer le connecteur :

Accorder l’utilisation sur un volume externe¶

Par exemple, si votre table Iceberg utilise le volume externe kinesis_external_volume et que le connecteur utilise le rôle openflow_kinesis_connector_role_1, exécutez l’instruction suivante :

USE ROLE ACCOUNTADMIN;
GRANT USAGE ON EXTERNAL VOLUME kinesis_external_volume TO ROLE openflow_kinesis_connector_role_1;

Créer une table Apache Iceberg™ pour l’ingestion¶

Le connecteur ne crée pas automatiquement des tables Iceberg et ne prend pas en charge l’évolution des schémas. Avant d’exécuter le connecteur, vous devez créer manuellement une table Iceberg.

Lorsque vous créez une table Iceberg, vous pouvez utiliser des types de données Iceberg (notamment VARIANT) ou des types Snowflake compatibles </user-guide/tables-iceberg-data-types>.

Prenons l’exemple du message suivant :

{
  "id": 1,
  "name": "Steve",
  "body_temperature": 36.6,
  "approved_coffee_types": ["Espresso", "Doppio", "Ristretto", "Lungo"],
  "animals_possessed": {
    "dogs": true,
    "cats": false
  },
  "options": {
    "can_walk": true,
    "can_talk": false
  },
  "date_added": "2024-10-15"
}

Pour créer une table Iceberg pour le message d’exemple, utilisez les instructions suivantes :

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
  kinesisMetadata OBJECT(
    stream STRING,
    shardId STRING,
    approximateArrival STRING,
    partitionKey STRING,
    sequenceNumber STRING,
    subSequenceNumber INTEGER,
    shardedSequenceNumber STRING
  ),
  id INT,
  name string,
  body_temperature float,
  approved_coffee_types array(string),
  animals_possessed variant,
  date_added date,
  options object(can_walk boolean, can_talk boolean)
)
EXTERNAL_VOLUME = 'my_volume'
CATALOG = 'SNOWFLAKE'
BASE_LOCATION = 'my_location/my_iceberg_table'
ICEBERG_VERSION = 3;