Einrichten von Openflow Connector for Kinesis für JSON-Datenformat¶

Voraussetzungen¶

1. Lesen Sie Allgemeine Informationen zu Openflow Connector for Kinesis.

2. Stellen Sie sicher, dass Sie folgende Aufgaben ausgeführt haben: Openflow einrichten mit BYOC oder Openflow einrichten – Snowflake-Bereitstellungen.

3. Stellen Sie bei der Verwendung von Openflow - Snowflake Deployments sicher, dass Sie Konfigurieren der erforderlichen Domänen gelesen haben und Zugriff auf die erforderlichen Domänen für den Kinesis-Konnektor gewährt haben.

Einen Kinesis-Stream einrichten¶

Als AWS-Administrator führen Sie die folgenden Aktionen in Ihrem AWS-Konto durch:

1. Stellen Sie sicher, dass Sie einen AWS-Benutzer mit IAM-Berechtigungen für den Zugriff auf Kinesis-Streams und DynamoDB haben.

2. Stellen Sie sicher, dass der AWS-Benutzer Zugriffsschlüssel-Anmeldeinformationen konfiguriert hat.

Snowflake-Konto einrichten¶

Als Snowflake-Kontoadministrator führen Sie die folgenden Aufgaben aus:

1. Erstellen Sie eine neue Rolle oder verwenden Sie eine vorhandene Rolle und erteilen Sie die Berechtigungen von Datenbanken.

2. Erstellen Sie eine Zieldatenbank und ein Zielschema, die verwendet werden, um Zieltabellen zum Speichern der Daten zu erstellen.

   1. Wenn Sie die Fähigkeit des Konnektors nutzen möchten, automatisch eine Zieltabelle zu erstellen, wenn diese noch nicht existiert, stellen Sie sicher, dass der Benutzer über die erforderlichen Berechtigungen zum Erstellen und Verwalten von Snowflake-Objekten verfügt:

      Objekt	Berechtigung	Anmerkungen
      Datenbank	USAGE
      Schema	USAGE . CREATE TABLE .	Nachdem die Objekte auf Schemaebene erstellt wurden, können die Berechtigungen für CREATE object widerrufen werden.
      Tabelle	OWNERSHIP	Nur erforderlich, wenn Sie den Kinesis-Konnektor verwenden, um Daten in eine bestehende Tabelle zu importieren. . Wenn der Konnektor eine neue Zieltabelle für Datensätze aus dem Kinesis-Stream erstellt, wird die Standardrolle für den in der Konfiguration angegebenen Benutzer zum Tabelleneigentümer.

      Sie können das folgende Skript verwenden, um eine benutzerdefinierte Rolle zu erstellen und zu konfigurieren (erfordert SECURITYADMIN oder gleichwertig):

      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. Erstellen Sie einen neuen Benutzer für den Snowflake-Dienst mit dem Typ SERVICE.

4. Weisen Sie dem Benutzer des Snowflake-Dienstes die Rolle zu, die Sie in den vorherigen Schritten erstellt haben.

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

   Copy

5. Konfigurieren Sie mit Schlüsselpaar-Authentifizierung für den Snowflake SERVICE-Benutzer aus Schritt 3.

6. Snowflake empfiehlt diesen Schritt dringend. Konfigurieren Sie einen von Openflow unterstützten Geheimnismanager, z. B. AWS, Azure und Hashicorp, und speichern Sie die öffentlichen und privaten Schlüssel im Geheimnisspeicher.

   Bemerkung

   Wenn Sie aus irgendeinem Grund keinen Geheimnismanager verwenden möchten, sind Sie dafür verantwortlich, die für die Schlüsselpaar-Authentifizierung verwendeten öffentlichen und privaten Schlüsseldateien gemäß den Sicherheitsrichtlinien Ihrer Organisation zu schützen.

   1. Nachdem der Geheimnismanager konfiguriert wurde, legen Sie fest, wie Sie sich bei diesem authentifizieren möchten. Für AWS wird empfohlen, dass Sie die EC2-Instanzrolle verwenden, die mit Openflow verbunden ist, da auf diese Weise keine weiteren Geheimnisse bestehen bleiben müssen.

   2. Konfigurieren Sie in Openflow über das Hamburger-Menü oben rechts einen Parameteranbieter, der mit diesem Secrets Manager verbunden ist. Navigieren Sie zu Controller Settings » Parameter Provider und rufen Sie dann Ihre Parameterwerte ab.

   3. Zu diesem Zeitpunkt können alle Anmeldeinformationen mit den zugehörigen Parameterpfaden referenziert werden, und es müssen keine sensiblen Werte innerhalb von Openflow aufbewahrt werden.

7. Wenn andere Snowflake-Benutzer Zugriff auf die aufgenommenen Daten und die erstellten Tabellen benötigen (z. B. für die kundenspezifische Verarbeitung in Snowflake), weisen Sie diesen Benutzern die in Schritt 2 erstellte Rolle zu.

Einrichten des Konnektors¶

Als Data Engineer führen Sie die folgenden Aufgaben aus, um den Konnektor zu installieren und zu konfigurieren:

Parameter¶

Dieser Abschnitt beschreibt alle Parameter für das Openflow Connector for Kinesis für JSON-Datenformat.

Der Konnektor besteht aus mehreren Modulen. Um das Set anzuzeigen, doppelklicken Sie auf die Konnektor-Prozessgruppe. Sie können die Parameter für jedes Modul im Parameterkontext des Moduls festlegen.

Führen Sie den Ablauf aus¶

1. Klicken Sie mit der rechten Maustaste auf die Ebene, und wählen Sie Enable all Controller Services aus.

2. Klicken Sie mit der rechten Maustaste auf die Prozessgruppe des Konnektors, und wählen Sie Start aus.

Der Konnektor startet die Datenaufnahme.

Unterstützung von Iceberg-Tabellen¶

Openflow Connector for Kinesis kann Daten in eine von Snowflake verwaltete Apache Iceberg™-Tabelle aufnehmen, wenn Iceberg Enabled auf true gesetzt ist.

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';