nfarah86
commented
1 year ago cc @yihua to help with this issue.
Open joeytman opened 1 year ago
nfarah86
commented
1 year ago cc @yihua to help with this issue.
yihua
commented
1 year ago Hi @joeytman thanks for the details. @rmahindra123 could you help answer the question around the Debezium source?
joeytman
commented
1 year ago 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 😄).
codope
commented
1 year ago Tracking https://issues.apache.org/jira/browse/HUDI-5877 to followup on the fix.
Tips before filing an issue
Have you gone through our FAQs? Yes
If you have triaged this as a bug, then file an issue directly. Unknown yet if it's a bug or user error
Describe the problem you faced
I'm trying to set up a DeltaStreamer job that consumes Avro changelog from Kafka and writes to s3. The changelog is created by Debezium from Vitess binlog, writing schemas to the confluent-api-compatible apicurio registry. I've gone ahead and implemented a VitessDebeziumSource and VitessDebeziumAvroPayload and can see the envelope is deserialized correctly and the flattened dataset looks good from logs within
VitessDebeziumSource.processDataset. However, writes are later failing when trying to deserialize records here (invoked here).To Reproduce
Steps to reproduce the behavior:
VitesssDebeziumSource.javaandVitessDebeziumAvroPayload.java. My implementation of each is:VitessDebeziumSource:VitessDebeziumAvroPayload:Launch DeltaStreamer job via EMR CLI w/
Observe deserialization error (details below)
Expected behavior
In RFC-39, its stated that:
From my understanding of this paragraph, at the point in the write operation we're failing at, we should be inferring the schema from the record which has already been deserialized and flattened via
VitessDebeziumSource. However, adding some logs to thebytesToAvromethod shows that thewriterSchemaandreaderSchemaare set to the schema of the debezium envelope, rather than the schema of the record we're writing. This goes against my understanding of how the flow should work, as it seems like it's using the target schema returned from theSchemaRegistryProviderrather than inferring the schema.I also tried implementing DebeziumSchemaRegistryProvider with a new
getTargetSchemaas mentioned as an alternative in the RFC, and was able to see that the schemas passed tobytesToAvrowere now set to the formatted schema produced byDebeziumSchemaRegistryProvider, confirming that the schema inference flow described by the RFC is not being triggered.Is there some config that needs to be set so that Deltastreamer will write with schema of the record produced by VitessDebeziumSource ? Or am I misunderstanding how this is supposed to work?
Environment Description
Hudi version : 0.12.1 (also tried on 0.13.0 and 0.14.0 and got the same error)
Spark version : 3.3.0-amzn-1
Hive version : Hive 3.1.3-amzn-2
Hadoop version : Hadoop 3.3.3-amzn-1
EMR version : emr-6.9.0
Storage (HDFS/S3/GCS..) : S3
Running on Docker? (yes/no) : No
Additional context
Debezium Envelope Schema (from schema registry):
From within the
VitessDebeziumSourcelogs we can see that the packets from kafka are ingested, deserialized, and flattened properly:But from logs within
avroToByteswe can see that thewriterSchemaandreaderSchemaare still set to the Debezium Envelope schema, rather than inferring the schema from the flattened dataset:Stacktrace Driver:
Executor logs shows many stacktraces, with two primary variations of errors.
Negative Length error:
String too long error:
Given that we’ve already deserialized the records once at this point, I don't think it's an issue with the data itself, despite what the errors may imply. I think the error is arising from the mismatch between the schema used to deserialize and the actual bytes, but let me know if you think something else could be going on.