Skip to content

Repository files navigation

Flink Utils

Various tools and improvements for Flink

Formats

json-schema (Json With Schema Registry)

'json-schema' format has been created in order to support Schema Registry Json messages.
This format is based on the Apicurio Json SerDe
All the Apicurio configs can be used with this format.
Connector kafka will supply the default json-schema.apicurio.registry.artifact.artifact-id equals to topic.name-value, but it can be changed in the DDL.

This format has the same configuration as Json Format with addition of all the Apicurio configs and the following ones:

ParameterDescriptionExampleDefault
json-schema.schema-registry.schema-nametitle field in Json Schema definition. By default equals to the artifact-id in Upper Camel CaseSomeSchemaNameTopicNameValue

Default Apicurio configs for this format:

ParameterDescriptionExampleDefault
json-schema.apicurio.registry.as-confluentUse Confluent Id formatfalsetrue
json-schema.apicurio.registry.headers.enabledBoolean to indicate whether serde classes should pass Global Id information via message headers instead of in the message payload.falsetrue
json-schema.apicurio.registry.auto-registerBoolean to indicate whether serializer classes should attempt to create an artifact in the registry.falsetrue
json-schema.apicurio.registry.use-idConfigures the serdes to use the specified IdOption as the identifier for the artifacts. Instructs the serializer to write the specified id into the kafka records and instructs the deserializer to read and use the specified id from the kafka records (to find the schema).globalIdcontentId

Schema Registry URL should be configured via json-schema.apicurio.registry.url parameter.

Keep in mind that the URL should contain only https://schema-registry.host.name value and should NOT contain /apis/ccompat7 part of it

avro-schema (Avro With Schema Registry)

'avro-schema' format has been created in order to support Schema Registry Avro messages.
This format is based on the Apicurio Avro SerDe
All the Apicurio configs can be used with this format.
Connector kafka will supply the default avro-schema.apicurio.registry.artifact.artifact-id equals to topic.name-value, but it can be changed in the DDL.

This format has all the Apicurio configs and the following ones:

ParameterDescriptionExampleDefault
avro-schema.schema-registry.schema-namename field in Avro Schema definition. By default equals to the artifact-id in Upper Camel CaseSomeSchemaNameTopicNameValue
avro-schema.schema-registry.schema-namespacenamespace field in Avro Schema definition. By default equals to com.exness.sdp.flink.custom.namespacecom.exness.sdp.flink

Default Apicurio configs for this format:

ParameterDescriptionExampleDefault
avro-schema.apicurio.registry.as-confluentUse Confluent Id formatfalsetrue
avro-schema.apicurio.registry.headers.enabledBoolean to indicate whether serde classes should pass Global Id information via message headers instead of in the message payload.falsefalse
avro-schema.apicurio.registry.auto-registerBoolean to indicate whether serializer classes should attempt to create an artifact in the registry.falsetrue
avro-schema.apicurio.registry.use-idConfigures the serdes to use the specified IdOption as the identifier for the artifacts. Instructs the serializer to write the specified id into the kafka records and instructs the deserializer to read and use the specified id from the kafka records (to find the schema).globalIdcontentId

Schema Registry URL should be configured via avro-schema.apicurio.registry.url parameter.

Keep in mind that the URL should contain only https://schema-registry.host.name value and should NOT contain /apis/ccompat7 part of it

About

Flink Connectors, Formats and other plugins

Resources

Stars

3 stars

Watchers

3 watching

Forks

Releases

Packages

Contributors

Languages