Set up the Openflow Connector for Kafka¶

Flow parameters¶

Kafka-related parameters

Snowflake-related parameters

Other parameters

Topic To Table Map example values:

• topic1:low_range,topic2:low_range,topic5:high_range,topic6:high_range

• topic[0-4]:low_range,topic[5-9]:high_range

SASL¶

To configure SASL authentication, you will need to provide the following configuration properties:

• Kafka Connection Service: ID of the Kafka Connection controller service. For more information, refer to Getting service ID

• Kafka Security Protocol: SASL_PLAINTEXT or SASL_SSL

• Kafka SASL Mechanism: PLAIN, SCRAM-SHA-256 or SCRAM-SHA-512

• SASL Username: Kafka username

• SASL Password: Kafka password

SASL with AWS MSK IAM¶

To configure SASL authentication with IAM, use the following configuration properties:

• Kafka Connection Service: ID of the Kafka Connection IAM controller service. For more information, refer to Getting service ID

• Kafka Security Protocol: SASL_PLAINTEXT or SASL_SSL

• Kafka SASL Mechanism: AWS_MSK_IAM

You also need to provide IAM credentials in Openflow with BYOC (bring your own cloud) configurations, deployed in your cloud.

mTLS¶

To configure mTLS authentication, you will need to generate and configure the necessary certificates for both the connector and the Kafka broker.

After obtaining the necessary certificates, generate a keystore for secure storage of both the connector’s private key and its certificate, and a truststore for confirming the Kafka broker certificate. The truststore must contain either the Kafka broker certificate or an alternative certificate in the certification chain that can be utilized to verify the broker’s certificate.

The supported keystore/truststore formats are PKCS12, JKS, and BCFKS.

Once the certificates are prepared, provide the following configuration properties:

• Kafka Connection Service: ID of the Kafka Connection SSL controller service. For more information, refer to Getting service ID

• Kafka Security Protocol - SSL

• mTLS Keystore Filename: A full path to the keystore file, or Asset reference

• mTLS Keystore Password: The keystore password

• mTLS Keystore Type: The keystore type, such as PKCS12, JKS, or BCFKS

• mTLS Key Password: The key password for the private key, if used

• mTLS Truststore Filename: A full path to the truststore file, or Asset reference

• mTLS Truststore Password: the truststore password

• mTLS Truststore Type: the truststore type, such as PKCS12, JKS, or BCFKS

Enable ingestion into Iceberg table¶

To enable ingestion into an Iceberg table, you must set the Iceberg Enabled parameter to true.

Create an Iceberg table for ingestion¶

Before you run the connector, you must create an Iceberg table. The initial table schema depends on your connector Schematization Enabled property settings.

If you enable schematization, you must create a table with a column named record_metadata:

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

The connector automatically creates the columns for message fields and alters the record_metadata column schema.

If you don’t enable schematization, you must create a table with a column named record_content of a type that matches the actual Kafka message content. The connector automatically creates the record_metadata column.

When you create an Iceberg table, you can use Iceberg data types or compatible Snowflake types. The semi-structured VARIANT type isn’t supported. Instead, use a structured OBJECT or MAP.

For example, consider the following message:

{
    "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"
}

To create an Iceberg table for the example message and schematization enabled, use the following statement:

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    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';

Copy

To create an Iceberg table for the example message and schematization disabled, run the following:

CREATE OR REPLACE ICEBERG TABLE my_iceberg_table (
    RECORD_METADATA OBJECT(
        offset INTEGER,
        topic STRING,
        partition INTEGER,
        key STRING,
        timestamp TIMESTAMP,
        SnowflakeConnectorPushTime BIGINT,
        headers MAP(VARCHAR, VARCHAR)
    ),
    RECORD_CONTENT OBJECT(
        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';

Copy