Systems in streaming pipelines can use different representations for schema and data types.
Schema mapping is required to align congruent types in a pipeline. For example, to send data from a CDC-enabled {cass-reg} table to an {pulsar-reg} topic, you must define schema mapping between the {cass-short} table and the {pulsar-short} topic. If the schema includes more complex data types, like maps and nested data structures, schema management becomes more difficult.
Schema management is a complicated, tedious, and error-prone tasks that requires you to understand and translate multiple sets of schema rules.
Instead, you can use the {gpt-schema-translator} to save time and reduce schema mapping toil.
This tool uses generative AI,based on the GPT-4 model, to automatically generate schema mappings between {product} ({pulsar-short} topics) and the {astra-db} sink connector ({cass-short} tables).
-
An {product} tenant with a namespace and topic.
-
An {astra-db} database with a keyspace and table.
-
An {astra-db} sink connector. The {gpt-schema-translator} is available for the {astra-db} sink connector only.
This example uses a JSON schema for a {pulsar-short} topic, and a CQL schema for an {astra-db} table. The {gpt-schema-translator} generates a mapping between the two schemas that the {astra-db} sink connector can use to write data from the {pulsar-short} topic to the {astra-db} table.
-
In the {astra-ui-link} header, click [grip], and then select Streaming.
-
Click the name of your tenant.
-
On the Namespaces and Topics tab, find the topic that you want to map to a table, and then click Create mapping.
-
Select your database’s keyspace, and then enter the name of the table that you want to map to.
-
Click Generate Mapping.
If this button isn’t available, the GPT schema translator doesn’t have an available schema for the topic or schema mapping isn’t available for the selected topic and table.
Table 1. {cass-short}-to-{pulsar-short} schema mapping example {cass-short} table schema {pulsar-short} JSON schema Generated mapping { "primaryKey": { "partitionKey": [ "id" ] }, "columnDefinitions": [ { "name": "id", "typeDefinition": "uuid", "static": false }, { "name": "file1", "typeDefinition": "text", "static": false }, { "name": "file2", "typeDefinition": "text", "static": false }, { "name": "file3", "typeDefinition": "text", "static": false } ] }{ "type": "record", "name": "sample.schema", "namespace": "default", "fields": [ { "name": "file1", "type": [ "null", "string" ], "default": null }, { "name": "file2", "type": [ "null", "string" ], "default": null }, { "name": "file3", "type": [ "string", "null" ], "default": "dfdf" } ] }id=key, file1=value.file1, file2=value.file2, file3=value.file3 -
Save the mapping configuration.
-
After configuring your {astra-db} sink connector for the given topic and table, messages should flow between the two schemas without error. You can check the {astra-ui} logs to confirm that the data is flowing into your table without errors.
This example demonstrates how you can generates schema mapping in real time.
-
The Data Generator source connector generates data for a {pulsar-short} topic with an AVRO schema.
AVRO schema example"pulsar_topic_schema": { "person": { "type": "record", "name": "Person", "namespace": "org.apache.pulsar.io.datagenerator", "fields": [ { "name": "address", "type": [ "null", { "type": "record", "name": "Address", "namespace": "org.apache.pulsar.io.datagenerator.Person", "fields": [ { "name": "apartmentNumber", "type": [ "null", "string" ], "default": null }, { "name": "city", "type": [ "null", "string" ], "default": null }, { "name": "postalCode", "type": [ "null", "string" ], "default": null }, { "name": "street", "type": [ "null", "string" ], "default": null }, { "name": "streetNumber", "type": [ "null", "string" ], "default": null } ] } ], "default": null }, { "name": "age", "type": [ "null", "int" ], "default": null }, { "name": "company", "type": [ "null", { "type": "record", "name": "Company", "namespace": "org.apache.pulsar.io.datagenerator.Person", "fields": [ { "name": "domain", "type": [ "null", "string" ], "default": null }, { "name": "email", "type": [ "null", "string" ], "default": null }, { "name": "name", "type": [ "null", "string" ], "default": null }, { "name": "vatIdentificationNumber", "type": [ "null", "string" ], "default": null } ] } ], "default": null }, { "name": "companyEmail", "type": [ "null", "string" ], "default": null }, { "name": "dateOfBirth", "type": { "type": "long", "logicalType": "timestamp-millis" } }, { "name": "email", "type": [ "null", "string" ], "default": null }, { "name": "firstName", "type": [ "null", "string" ], "default": null }, { "name": "lastName", "type": [ "null", "string" ], "default": null }, { "name": "middleName", "type": [ "null", "string" ], "default": null }, { "name": "nationalIdentificationNumber", "type": [ "null", "string" ], "default": null }, { "name": "nationalIdentityCardNumber", "type": [ "null", "string" ], "default": null }, { "name": "passportNumber", "type": [ "null", "string" ], "default": null }, { "name": "password", "type": [ "null", "string" ], "default": null }, { "name": "sex", "type": [ "null", { "type": "enum", "name": "Sex", "namespace": "org.apache.pulsar.io.datagenerator.Person", "symbols": [ "MALE", "FEMALE" ] } ], "default": null }, { "name": "telephoneNumber", "type": [ "null", "string" ], "default": null }, { "name": "username", "type": [ "null", "string" ], "default": null } ] }, } -
The {astra-db} sink connector writes data to the {cass-short} table with a CQL schema.
CQL schema example"cassandra_table_schemas": { "person": { "primaryKey": { "partitionKey": [ "passportnumber" ] }, "columnDefinitions": [ { "name": "passportnumber", "typeDefinition": "text", "static": false }, { "name": "age", "typeDefinition": "varint", "static": false }, { "name": "firstname", "typeDefinition": "text", "static": false }, { "name": "lastname", "typeDefinition": "text", "static": false } ] }, }
-
In the {astra-ui}, create a mapping for the tenant. When a topic schema is available to the {gpt-schema-translator}, click Generate Mapping.
The {gpt-schema-translator} generates an AVRO-to-CQL schema mapping while messages are processed.
passportnumber=value.passportNumber, age=value.age, firstname=value.firstName, lastname=value.lastNameNotice that the
firstnamevalue becamefirstNamebecause the {pulsar-short} topic AVRO schema superseded the {cass-short} table schema.
If you don’t declare a schema in the {pulsar-short} topic, the schema translator can generate a default schema mapping based on the values of your {cass-short} table schema, without using GPT.
When you create the mapping in the {astra-ui}, you can click Generate Mapping to create a generic {pulsar-short} topic schema based on your {cass-short} table schema. If schema mapping isn’t possible for the selected table and topic, the Generate Mapping button isn’t available.
For example, assume you have the following {cass-short} table schema:
{
"primaryKey": {
"partitionKey": [
"passportnumber"
]
},
"columnDefinitions": [
{
"name": "passportnumber",
"typeDefinition": "text",
"static": false
},
{
"name": "age",
"typeDefinition": "varint",
"static": false
},
{
"name": "firstname",
"typeDefinition": "text",
"static": false
},
{
"name": "lastname",
"typeDefinition": "text",
"static": false
}
]
}The schema translator would generate the following {pulsar-short} JSON schema mapping based on the given {cass-short} table schema:
passportnumber=value.passportnumber, age=value.age, firstname=value.firstname, lastname=value.lastname