joeytman commented on issue #7973:
URL: https://github.com/apache/hudi/issues/7973#issuecomment-1439400942
Hello again, I have an update. I was able to work around the issue and get
DeltaStreamer working by defining a `DebeziumSchemaRegistryProvider` as follows:
```
package org.apache.hudi.utilities.schema;
import org.apache.avro.Schema;
import org.apache.avro.SchemaBuilder;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.debezium.DebeziumConstants;
import org.apache.spark.api.java.JavaSparkContext;
import java.util.Arrays;
/**
* Obtains latest schema from the Confluent/Kafka schema-registry, with
Debezium Envelope format
* <p>
* https://github.com/confluentinc/schema-registry
*/
public class DebeziumSchemaRegistryProvider extends SchemaRegistryProvider {
public DebeziumSchemaRegistryProvider(TypedProperties props,
JavaSparkContext jssc) {
super(props, jssc);
}
/**
* Debezium target schema is a nested structure with many metadata
fields. This will
* flatten the schema structure and only require necessary metadata
information
* @return
*/
@Override
public Schema getTargetSchema() {
Schema registrySchema = super.getTargetSchema();
Schema.Field opField =
registrySchema.getField(DebeziumConstants.INCOMING_OP_FIELD);
Schema.Field dataField =
registrySchema.getField(DebeziumConstants.INCOMING_AFTER_FIELD);
// TODO: use INCOMING_SOURCE_TS_MS_FIELD instead
// naively using INCOMING_SOURCE_TS_MS_FIELD results in NPE, but we
can work around it
// since INCOMING_TS_MS_FIELD has nullable version of
INCOMING_SOURCE_TS_MS_FIELD schema
Schema.Field tsField =
registrySchema.getField(DebeziumConstants.INCOMING_TS_MS_FIELD);
// Initialize with metadata columns which must match the fields
included in VitessDebeziumSource.
// Note: ALL fields are nullable after being processed by
VitessDebeziumSource
// so if you want to include a field that is NOT nullable in the
schema registry,
// you must call `setSchemaNullable` on the schema. If it's already
nullable, just use `type(field.schema())`
SchemaBuilder.FieldAssembler<Schema> payloadFieldAssembler =
SchemaBuilder.builder()
.record("formatted_debezium_payload")
.fields()
.name(DebeziumConstants.FLATTENED_OP_COL_NAME).type(setSchemaNullable(opField.schema())).withDefault(null)
.name(DebeziumConstants.FLATTENED_TS_COL_NAME).type(tsField.schema()).withDefault(null);
// Add data columns to schema
dataField.schema()
.getTypes()
// "after" field is a union with data schema and null
schema, so we need to extract only the data schema portion
.get(dataField.schema().getIndexNamed(registrySchema.getNamespace() + ".Value"))
.getFields()
.forEach(field -> {
payloadFieldAssembler.name(field.name()).type(setSchemaNullable(field.schema())).withDefault(null);
});
return payloadFieldAssembler.endRecord();
}
/**
* @param toBecomeNullable source Schema (non-nullable) that needs to
become nullable
* @return Schema of UNION type, like: [null, toBecomeNullable]
*/
private Schema setSchemaNullable(Schema toBecomeNullable) {
return
Schema.createUnion(Arrays.asList(Schema.create(Schema.Type.NULL),
toBecomeNullable));
}
}
```
I still think the issue should remain open, since the RFC explicitly calls
this approach out as something we _don't_ have to do. But I wanted to share
this workaround to confirm that everything else is working besides the schema
inference that's called out in the RFC (and help anyone else with a similar
issue of course 😄).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
