DataWorks:Kafka

Resource estimation for real-time read

When using a subscription Serverless resource group, estimate resources before starting a task to avoid failures from insufficient capacity.

Per-topic base allocation: 1 CU per topic

Traffic-based allocation:

Cluster slot usage limits:

Offline read from a single table

If both parameter.groupId and parameter.kafkaConfig.group.id are configured, parameter.groupId takes precedence.

Real-time write to a single table

Data deduplication is not supported. If a task restarts after an offset reset or a failover, duplicate records may be written.

Real-time write for an entire database

Key assignment by primary key configuration:

Strict ordering under cluster exceptions:

If the Kafka cluster returns exceptions and you need strict ordering for records with the same primary key, add the following to the extended parameters of the Kafka data source:

{"max.in.flight.requests.per.connection":1,"buffer.memory": 100554432}

For the format of messages written to Kafka—including heartbeat messages and DDL change records—see Appendix: Message format.

Read data

DataWorks can parse Kafka data in JSON format. The table below shows how each field is handled.

Write data

DataWorks supports writing to Kafka in JSON or text format. The format behavior varies by sync task type.

Configure an offline sync task for a single table

• For the setup procedure, see Configure an offline synchronization task in the codeless UI and Configure an offline synchronization task in the code editor.

• For the full parameter list and a script demo, see Appendix: Script demos and parameter descriptions.

Configure a real-time sync task for a single table or entire database

See Configure a real-time synchronization task in DataStudio, Configure a real-time synchronization task in Data Integration, and Configure a synchronization task for an entire database.

SSL

Set Special Authentication Method to SSL or SASL_SSL to enable SSL authentication. You must upload a client truststore certificate file and enter the truststore passphrase.

Truststore certificate by cluster type:

• Alibaba Cloud Kafka: See SSL certificate algorithm upgrade instructions to download the correct certificate. The truststore passphrase is KafkaOnsClient.

• EMR cluster: See Use SSL encryption for Kafka connections to download the certificate and retrieve the passphrase.

• Self-managed cluster: Upload the correct truststore certificate and enter the corresponding passphrase.

The keystore certificate file, keystore passphrase, and SSL passphrase are only required when bidirectional SSL authentication is enabled (ssl.client.auth=required in the server's server.properties file). See Use SSL encryption for Kafka connections for details.

GSSAPI

Set Sasl Mechanism to GSSAPI when the Kafka cluster uses Kerberos authentication. Upload three files—a JAAS configuration file, a Kerberos configuration file, and a Keytab file—and configure DNS/HOST settings for the exclusive resource group.

KafkaClient {
   com.sun.security.auth.module.Krb5LoginModule required
   useKeyTab=true
   keyTab="<path-to-keytab>"
   storeKey=true
   serviceName="kafka-server"
   principal="<client-principal>@<REALM>";
};

[libdefaults]
  default_realm = <REALM>

[realms]
  <REALM> = {
    kdc = <kdc-ip>:<port>
  }

klist -ket ./client.keytab

Keytab name: FILE:client.keytab
KVNO Timestamp           Principal
---- ------------------- ------------------------------------------------------
   7 2018-07-30T10:19:16 te**@**.com (des-cbc-md5)

PLAIN

Set Sasl Mechanism to PLAIN when the Kafka cluster uses username/password authentication. The JAAS file must start with KafkaClient, with all configuration items enclosed in braces {}.

Format rules:

• The first line inside the braces specifies the login module class.

• Each configuration item uses key=value format.

• All items except the last must not end with a semicolon.

• The last item must end with a semicolon, and a semicolon must follow the closing brace }.

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="<your-username>"
  password="<your-password>";
};

Reader script demo

{
    "type": "job",
    "steps": [
        {
            "stepType": "kafka",
            "parameter": {
                "server": "host:9093",
                "column": [
                    "__key__",
                    "__value__",
                    "__partition__",
                    "__offset__",
                    "__timestamp__",
                    "'123'",
                    "event_id",
                    "tag.desc"
                ],
                "kafkaConfig": {
                    "group.id": "demo_test"
                },
                "topic": "topicName",
                "keyType": "ByteArray",
                "valueType": "ByteArray",
                "beginDateTime": "20190416000000",
                "endDateTime": "20190416000006",
                "skipExceedRecord": "true"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {
                "print": false,
                "fieldDelimiter": ","
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "version": "2.0",
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    },
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "throttle": true,
            "concurrent": 1,
            "mbps": "12"
        }
    }
}

Reader script parameters

kafkaConfig parameters

Use kafkaConfig to pass extended configuration to the KafkaConsumer client.

Writer script demo

{
  "type": "job",
  "version": "2.0",
  "steps": [
    {
      "stepType": "stream",
      "parameter": {},
      "name": "Reader",
      "category": "reader"
    },
    {
      "stepType": "Kafka",
      "parameter": {
          "server": "ip:9092",
          "keyIndex": 0,
          "valueIndex": 1,
          "keyType": "Integer",
          "valueType": "Short",
          "topic": "t08",
          "batchSize": 1024
        },
      "name": "Writer",
      "category": "writer"
    }
  ],
  "setting": {
      "errorLimit": {
        "record": "0"
      },
      "speed": {
          "throttle": true,
          "concurrent": 1,
          "mbps": "12"
      }
   },
  "order": {
    "hops": [
        {
            "from": "Reader",
            "to": "Writer"
        }
      ]
    }
}

Writer script parameters