Transferring data from an Apache Kafka® source endpoint

1. Make sure the settings for the network hosting the cluster allow public connections from IP addresses used by Data Transfer.

2. Configure access to the source cluster from Yandex Cloud.

3. Configure user access rights to the topic you need.

4. Grant the READ permissions to the consumer group whose ID matches the transfer ID.

   bin/kafka-acls --bootstrap-server localhost:9092 \
     --command-config adminclient-configs.conf \
     --add \
     --allow-principal User:username \
     --operation Read \
     --group <transfer_ID>
   

5. (Optional) To use username and password authorization, configure SASL authentication.

• Endpoint type: kafka_source.

• security_groups: Security groups for network traffic.

  Security group rules apply to a transfer. They allow opening up network access from the transfer VM to the cluster. For more information, see Networking in Yandex Data Transfer.

  Security groups must belong to the same network as the cluster.

  Note

  In Terraform, it is not required to specify a network for security groups.

• connection.cluster_id: ID of the cluster to connect to.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication.
  • no_auth: Without authentication.

• topic_names: Names of the topics to connect to.

Here is an example of the configuration file structure:

resource "yandex_datatransfer_endpoint" "<endpoint_name_in_Terraform>" {
  name = "<endpoint_name>"
  settings {
    kafka_source {
      security_groups = ["<list_of_security_group_IDs>"]
      connection {
        cluster_id = "<cluster_ID>"
      }
      auth {
        <authentication_method>
      }
      topic_names = ["<list_of_topic_names>"]
      <extended_endpoint_settings>
    }
  }
}

For more information, see the Terraform provider documentation.

• securityGroups: Security groups for network traffic, whose rules will apply to VMs and clusters without changing their settings. For more information, see Networking in Yandex Data Transfer.

• connection.clusterId: ID of the cluster to connect to.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication. The following parameters are required:

    • user: Name of the account Data Transfer will use to connect to the topic.
    • password.raw: Password for the account in text form.
    • mechanism: Hashing mechanism.

  • noAuth: Without authentication.

• topicNames: Names of the topics to connect to.

• Endpoint type: kafka_source.

• security_groups: Security groups for network traffic.

  Security group rules apply to a transfer. They allow opening up network access from the transfer VM to the broker hosts. For more information, see Networking in Yandex Data Transfer.

  Security groups must belong to the same network as the subnet_id subnet, if the latter is specified.

  Note

  In Terraform, it is not required to specify a network for security groups.

• connection.on_premise: Parameters for connecting to the broker hosts:

  • broker_urls: IP addresses or FQDNs of the broker hosts.

    If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name:

    <broker_host_IP_address_or_FQDN>:<port_number>
    

  • tls_mode: Parameters for encrypting transmitted data if required, e.g., to comply with the PCI DSS.

    • disabled: Disabled.
    • enabled: Enabled.
      • ca_certificate: CA certificate.

  • subnet_id: ID of the subnet hosting the broker hosts.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication.
  • no_auth: Without authentication.

• topic_names: Names of the topics to connect to.

Here is an example of the configuration file structure:

resource "yandex_datatransfer_endpoint" "<endpoint_name_in_Terraform>" {
  name = "<endpoint_name>"
  settings {
    kafka_source {
      security_groups = ["<list_of_security_group_IDs>"]
      connection {
        on_premise {
          broker_urls = ["<list_of_IP_addresses_or_FQDNs_of_broker_hosts>"]
          subnet_id  = "<ID_of_subnet_with_broker_hosts>"
        }
      }
      auth = {
        <authentication_method>
      }
      topic_names = ["<list_of_topic_names>"]
      <extended_endpoint_settings>
    }
  }
}

For more information, see the Terraform provider documentation.

• securityGroups: Security groups for network traffic, whose rules will apply to VMs and clusters without changing their settings. For more information, see Networking in Yandex Data Transfer.

• connection.onPremise: Parameters for connecting to the broker hosts:

  • brokerUrls: IP addresses or FQDNs of the broker hosts.

    If the Apache Kafka® port number differs from the standard one, specify it with a colon after the host name:

    <broker_host_IP_address_or_FQDN>:<port_number>
    

  • tlsMode: Parameters for encrypting transmitted data if required, e.g., to comply with the PCI DSS.

    • disabled: Disabled.
    • enabled: Enabled.
      • caCertificate: CA certificate.

  • subnetId: ID of the subnet hosting the broker hosts.

• auth: Authentication method used to connect to broker hosts:

  • sasl: SASL authentication. The following parameters are required:

    • user: Name of the account Data Transfer will use to connect to the topic.
    • password.raw: Password for the account in text form.
    • mechanism: Hashing mechanism.

  • noAuth: Without authentication.

• topicNames: Names of the topics to connect to.

• transformer: Transformation rules:

• parser: Conversion rules which depend on the selected parser:

  • audit_trails_v1_parser: Audit Trails log parser.

  • cloud_logging_parser: Cloud Logging log parser.

  • json_parser or tskv_parser: JSON and TSKV parsers, respectively.

    Both parsers use the same attributes to define parameters:

    • data_schema: Data schema represented either as a JSON object with a schema description or as a list of fields:

      • fields: Schema represented as a JSON object. The object contains an array of JSON objects that describe individual columns.
      • json_fields: Schema represented as a list of fields.

    • null_keys_allowed: Set to true to allow the null value in key columns.

    • add_rest_column: Set to true to add the fields missing in the schema to the _rest column.

    • unescape_string_values: Set to true to remove quotation marks from string variables. Otherwise, the string field values will remain unchanged.

• transformer: Transformation rules:

  • cloudFunction: ID of the function created in Cloud Functions.

  • serviceAccountId: ID of the service account you will use to invoke the processing function.

  • numberOfRetries: Number of attempts to invoke the processing function.

  • bufferSize: Buffer size, in bytes. When the buffer is full, data will be transferred to the processing function.

    The maximum buffer size is 3.5 MB. For more information about restrictions that apply to functions in Cloud Functions, see the relevant section.

  • bufferFlushInterval: Interval to wait before transferring stream data to the processing function, in seconds.

    Note

    If the buffer is full or the flush interval has expired, the data is transferred to the processing function.

  • invocationTimeout: Allowed timeout of a processing function response, in seconds.

  Warning

  Append s to the bufferFlushInterval and invocationTimeout values, e.g., 10s.

• parser: Conversion rules which depend on the selected parser:

  • auditTrailsV1Parser: Audit Trails log parser.

  • cloudLoggingParser: Cloud Logging log parser.

  • jsonParser or tskvParser: JSON and TSKV parsers, respectively.

    Both parsers share the same parameter fields:

    • dataSchema: Data schema represented either as a JSON object with a schema description or as a list of fields:

      • fields: Schema represented as a JSON object. The object contains an array of JSON objects that describe individual columns.

      • jsonFields: Schema represented as a list of fields.

    • nullKeysAllowed: Set to true to allow the null value in key columns.

    • addRestColumn: Set to true to add the fields missing in the schema to the _rest column.

    • unescapeStringValues: Set to true to remove quotation marks from string variables. Otherwise, the string field values will remain unchanged.