Configurer Openflow Connector for Kinesis pour le format de données JSON¶

Conditions préalables¶

1. Consulter À propos de Openflow Connector for Kinesis

2. Assurez-vous d’avoir Configurer Openflow avec BYOC ou Configurer Openflow avec les déploiements Snowflake.

3. Si vous utilisez Openflow - Snowflake Deployments, assurez-vous d’avoir consulté la configuration des domaines requis et d’avoir accordé l’accès aux domaines requis pour le connecteur Kinesis.

Configurez un flux Kinesis¶

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

1. Assurez-vous que vous disposez d’un utilisateur AWS avec des autorisations IAM d’accès à Kinesis Streams et DynamoDB.

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

Paramétrage du compte Snowflake¶

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

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

2. Créez une base de données de destination et un schéma de destination qui serviront à créer les tables de destination pour le stockage des données.

   1. Si vous planifiez d’utiliser la capacité du connecteur à créer automatiquement la table de destination si elle n’existe pas encore, assurez-vous que l’utilisateur dispose des privilèges requis pour la création et la gestion d’objets Snowflake :

      Objet	Privilège	Remarques
      Base de données	USAGE
      Schéma	USAGE . CREATE TABLE .	Une fois les objets au niveau du schéma ont été créés, les privilèges CREATE object peuvent être révoqués.
      Table	OWNERSHIP	Uniquement requis lorsque vous utilisez le connecteur Kinesis pour ingérer des données dans une table existante. . Si le connecteur crée une nouvelle table cible pour les enregistrements du flux Kinesis, le rôle par défaut de l’utilisateur spécifié dans la configuration devient le propriétaire de la table.

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

      USE ROLE SECURITYADMIN;
      
      CREATE ROLE kinesis_connector_role;
      GRANT USAGE ON DATABASE kinesis_db TO ROLE kinesis_connector_role;
      GRANT USAGE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      GRANT CREATE TABLE ON SCHEMA kinesis_schema TO ROLE kinesis_connector_role;
      
      -- Only for existing tables.
      GRANT OWNERSHIP ON TABLE existing_table TO ROLE kinesis_connector_role;
      

      Copy

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

4. Autorisez l’utilisateur du service Snowflake le rôle que vous avez créé dans les étapes précédentes.

   GRANT ROLE kinesis_connector_role TO USER kinesis_connector_user;
   ALTER USER kinesis_connector_user SET DEFAULT_ROLE = kinesis_connector_role;
   

   Copy

5. Configurez avec l’authentification par paire de clés pour l’utilisateur de Snowflake SERVICE de l’étape 3.

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

   Note

   Si, pour une raison quelconque, vous ne souhaitez pas utiliser un gestionnaire de secrets, il vous incombe de protéger les fichiers de clés publiques et privées utilisés pour l’authentification par paires de clés conformément aux politiques de sécurité de votre organisation.

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

7. Si d’autres utilisateurs de Snowflake ont besoin d’accéder aux données ingérées et aux tables créées (par exemple, pour un traitement personnalisé dans Snowflake), accordez à ces utilisateurs 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 :

Paramètres¶

Cette section décrit tous les paramètres de Openflow Connector for Kinesis pour le format de données JSON.

Le connecteur se compose de plusieurs modules. Pour voir l’ensemble, double-cliquez sur le groupe de processus du connecteur. Vous pouvez définir les paramètres de chaque module dans le contexte de paramètres du module.

Exécutez le flux¶

1. Cliquez avec le bouton droit de la souris sur le Plan et sélectionnez Enable all Controller Services.

2. Cliquez à droite sur le groupe de processus du connecteur et sélectionnez Start.

Le connecteur démarre l’ingestion des données.

Prise en charge des tables Iceberg¶

Openflow Connector for Kinesis peut ingérer des données dans une Table Apache Iceberg™ gérée par Snowflake lorsque Iceberg activé est défini sur true.

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    kinesisMetadata OBJECT()
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table'
  ENABLE_SCHEMA_EVOLUTION = true;

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

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,
    body_temperature FLOAT,
    name STRING,
    approved_coffee_types ARRAY(STRING),
    animals_possessed OBJECT(dogs BOOLEAN, cats BOOLEAN),
    date_added DATE
  )
  EXTERNAL_VOLUME = 'my_volume'
  CATALOG = 'SNOWFLAKE'
  BASE_LOCATION = 'my_location/my_iceberg_table';