Transferring data from an Apache Kafka® source endpoint

1. If not planning to use Cloud Interconnect or VPN for connections to an external cluster, make such cluster accessible from the Internet from IP addresses used by Data Transfer.

   For details on linking your network up with external resources, see this concept.

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. Optionally, to log in with a username and password, 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 the configuration file example:

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 = ["<topic_name_list>"]
      <endpoint_advanced_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 the data to transfer, if required, e.g., for compliance with the PCI DSS requirements.

    • disabled: Disabled.
    • enabled: Enabled.

      • ca_certificate: CA certificate.

        Warning

        If no certificate is added, the transfer may fail with an error.

  • subnet_id: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them.

• 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 the configuration file example:

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_broker_host_FQDNs>"]
          subnet_id  = "<ID_of_subnet_with_broker_hosts>"
        }
      }
      auth = {
        <authentication_method>
      }
      topic_names = ["<topic_name_list>"]
      <endpoint_advanced_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 the data to transfer, if required, e.g., for compliance with the PCI DSS requirements.

    • disabled: Disabled.
    • enabled: Enabled.

      • caCertificate: CA certificate.

        Warning

        If no certificate is added, the transfer may fail with an error.

  • subnetId: ID of the subnet hosting the broker hosts. The transfer will use this subnet to access them.

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