[SUPPORT] Deltastreamer transform cannot add fields

[liujinhui1994]

Describe the problem you faced

source.avsc

{
"type": "record",
"name": "t3_app_td_ad_info",
"fields": [{
"name": "dataId",
"type": "string"
},
{
"name": "collectTime",
"type": "string"
},
{
"name": "clickTime",
"type": "string"
},
{
"name": "spreadUrl",
"type": ["null", "string"],
"default": null
},{
"name": "spreadName",
"type": ["null", "string"],
"default": null
},
{
"name": "ua",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetName",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetDesc",
"type": ["null", "string"],
"default": null
}
]
}

target.avsc

{
"type": "record",
"name": "t3_app_td_ad_info",
"fields": [{
"name": "dataId",
"type": "string"
},
{
"name": "collectTime",
"type": "string"
},
{
"name": "clickTime",
"type": "string"
},
{
"name": "ds",
"type": ["null", "string"],
"default": null
},
{
"name": "spreadUrl",
"type": ["null", "string"],
"default": null
},{
"name": "spreadName",
"type": ["null", "string"],
"default": null
},
{
"name": "ua",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetName",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetDesc",
"type": ["null", "string"],
"default": null
}
]
}

Use deltastreamer to get data from Kafka, define the source and target schemas as follows. Use a custom transform to add a new field named ds and type string. Found that an exception is thrown and cannot be written normally

Stacktrace

20/10/10 17:36:32 ERROR io.HoodieWriteHandle: Error writing record HoodieRecord{key=HoodieKey { recordKey=915FABFAD0A796B26B0E40E7602460C3 partitionPath=2020/10/10}, currentLocation='null', newLocation='null'}
java.lang.ArrayIndexOutOfBoundsException: 10
at org.apache.avro.io.parsing.Symbol$Alternative.getSymbol(Symbol.java:436)
at org.apache.avro.io.ResolvingDecoder.doAction(ResolvingDecoder.java:290)
at org.apache.avro.io.parsing.Parser.advance(Parser.java:88)
at org.apache.avro.io.ResolvingDecoder.readIndex(ResolvingDecoder.java:267)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:179)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
at org.apache.avro.generic.GenericDatumReader.readField(GenericDatumReader.java:232)
at org.apache.avro.generic.GenericDatumReader.readRecord(GenericDatumReader.java:222)
at org.apache.avro.generic.GenericDatumReader.readWithoutConversion(GenericDatumReader.java:175)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:153)
at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:145)
at org.apache.hudi.avro.HoodieAvroUtils.bytesToAvro(HoodieAvroUtils.java:144)
at org.apache.hudi.avro.HoodieAvroUtils.bytesToAvro(HoodieAvroUtils.java:131)
at org.apache.hudi.common.model.OverwriteWithLatestAvroPayload.getInsertValue(OverwriteWithLatestAvroPayload.java:70)
at org.apache.hudi.execution.LazyInsertIterable$HoodieInsertValueGenResult.(LazyInsertIterable.java:92)
at org.apache.hudi.execution.LazyInsertIterable.lambda$getTransformFunction$0(LazyInsertIterable.java:105)
at org.apache.hudi.common.util.queue.BoundedInMemoryQueue.insertRecord(BoundedInMemoryQueue.java:190)
at org.apache.hudi.common.util.queue.IteratorBasedQueueProducer.produce(IteratorBasedQueueProducer.java:46)
at org.apache.hudi.common.util.queue.BoundedInMemoryExecutor.lambda$null$0(BoundedInMemoryExecutor.java:92)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

[liujinhui1994]

@bhasudha can you help me thanks

[bvaradar]

@liujinhui1994 : I see that the new column column is added to the middle of the schema (not at the end). Are you doing the same thing with transformer ? You need to make sure the field offsets is consistent between the Rows generated and the avro schema.

[liujinhui1994]

source.avsc
{
"type": "record",
"name": "t3_app_td_ad_info",
"fields": [{
"name": "dataId",
"type": "string"
},
{
"name": "collectTime",
"type": "string"
},
{
"name": "clickTime",
"type": "string"
},
{
"name": "spreadUrl",
"type": ["null", "string"],
"default": null
},{
"name": "spreadName",
"type": ["null", "string"],
"default": null
},
{
"name": "ua",
"type": ["null", "string"],
"default": null
},
{
"name": "uid",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetName",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetDesc",
"type": ["null", "string"],
"default": null
}
]
}

target.avsc
{
"type": "record",
"name": "t3_app_td_ad_info",
"fields": [{
"name": "dataId",
"type": "string"
},
{
"name": "collectTime",
"type": "string"
},
{
"name": "clickTime",
"type": "string"
},
{
"name": "spreadUrl",
"type": ["null", "string"],
"default": null
},{
"name": "spreadName",
"type": ["null", "string"],
"default": null
},
{
"name": "ua",
"type": ["null", "string"],
"default": null
},
{
"name": "uid",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetName",
"type": ["null", "string"],
"default": null
},
{
"name": "adnetDesc",
"type": ["null", "string"],
"default": null
},
{
"name": "ds",
"type": "string"
}
]
}

[liujinhui1994]

[liujinhui1994]

@bvaradar I am using transform and want to add a new ds field. I now re-create a new hudi table, and add the ds field to the end of target.avsc according to your suggestion, but the same error is still reported.

[liujinhui1994]

[liujinhui1994]

Please help me to find out what went wrong, it has troubled me for a long time. Thank you very much for your help @bvaradar

[bvaradar]

1. I am not able to pinpoint the issue rightaway but let me engage in debugging this with you. Couple of things :

2. Can you make ds field and any additional fields you are adding nullable
   {
   "name": "ds",
   "type": ["null", "string"],
   "default": null
   }

3. In transformer implementation, after you constructed the dataset "fi", can you call printSchema() and add the output here.

If possible, It would make things very easy, if you can construct some form of test that I can use it to repro the issue and debug here.

[liujinhui1994]

1. I have now changed all fields to “ type”：[“ null”，“ string”]，“ default”：null

printSchema（）
root
|-- dataId: string (nullable = true)
|-- collectTime: string (nullable = true)
|-- clickTime: string (nullable = true)
|-- spreadUrl: string (nullable = true)
|-- spreadName: string (nullable = true)
|-- ua: string (nullable = true)
|-- uid: string (nullable = true)
|-- adnetName: string (nullable = true)
|-- adnetDesc: string (nullable = true)
|-- ds: string (nullable = true

@bvaradar

[liujinhui1994]

This is the data I printed in the transform

1. This is before adding the ds field

2. This is after adding the ds field

[bvaradar]

I think this is due to the way spark deduces avro schema from Transformer ROw (in a way similar to #2149 (comment))

You can try changing the SchemaProvider to do the below steps where the target schema is recreated using Spark-avro. This would make it consistent with Transformer generated DF.

Schema newSchema = AvroConversionUtils.convertStructTypeToAvroSchema(

•    AvroConversionUtils.convertAvroSchemaToStructType(schema), RowBasedSchemaProvider.HOODIE_RECORD_STRUCT_NAME,
  

•    RowBasedSchemaProvider.HOODIE_RECORD_NAMESPACE);
  

If you are using master branch, we have added support for plugging in SchemaPostProcessor (using config: hoodie.deltastreamer.schemaprovider.schema_post_processor=org.apache.hudi.utilities.schema.SparkAvroPostProcessor

where you can implement the processSchema() method to do the above transformation.

package org.apache.hudi.utilities.schema;

import org.apache.hudi.AvroConversionUtils;
import org.apache.hudi.common.config.TypedProperties;

import org.apache.avro.Schema;
import org.apache.spark.api.java.JavaSparkContext;

public class SparkAvroPostProcessor extends SchemaPostProcessor {

  protected SparkAvroPostProcessor(TypedProperties props, JavaSparkContext jssc) {
    super(props, jssc);
  }

  @Override
  public Schema processSchema(Schema schema) {
    return AvroConversionUtils.convertStructTypeToAvroSchema(
        AvroConversionUtils.convertAvroSchemaToStructType(schema), RowBasedSchemaProvider.HOODIE_RECORD_STRUCT_NAME, 
        RowBasedSchemaProvider.HOODIE_RECORD_NAMESPACE);
  }
}```


If this works, I will open a PR (Jira: https://issues.apache.org/jira/browse/HUDI-1343)

[liujinhui1994]

@bvaradar Okay, I will be experimenting and I would be happy to complete this PR if allowed, thanks

[bvaradar]

@liujinhui1994 : Did this work ?

[liujinhui1994]

Can work, but if the default value is not null, it will not work
{
"name": "adnetDesc",
"type": ["null", "long"],
"default": -1
}
@bvaradar