[SUPPORT] hudi deltastreamer jsonkafka source schema registry fail
- Dominant language
- Java
- Stars
- 6.2k
- Forks
- 2.5k
- Avg merge
- 2d 8h
- Merged PRs (30d)
- 111
Description
when trying to read data from Kafka which is stored as Json from schema registry. using below configuration. pls tell me where do I'm wrong with the configuration
- --checkpoint
- "OCR.VEHICLE.REGISTRATION,0:10"
- --schemaprovider-class
- org.apache.hudi.utilities.schema.SchemaRegistryProvider
- --source-class
- org.apache.hudi.utilities.sources.JsonKafkaSource
- --hoodie-conf
- "hoodie.deltastreamer.schemaprovider.registry.schemaconverter=org.apache.hudi.utilities.schema.converter.JsonToAvroSchemaConverter"
- --table-type
- COPY_ON_WRITE
- --source-ordering-field
- plate
- --target-base-path
- hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest
- --target-table
- ocr_vehicle_registration_cow_latest
- --op
- INSERT
- --props
- file:///var/hoodie/ws/properties/ocr_vehicle_registration_cow_latest.properties
- --min-sync-interval-seconds
- "1"
- --continuous
- --enable-hive-sync
- --source-limit
- "1000"
this is my property file
hoodie.datasource.write.recordkey.field=plate
hoodie.deltastreamer.schemaprovider.registry.url=http://confluent-schema-registry.kafka.svc:8081/subjects/OCR.VEHICLE.REGISTRATION-value/versions/latest
hoodie.deltastreamer.source.kafka.topic=OCR.VEHICLE.REGISTRATION
hoodie.datasource.hive_sync.table=ocr_vehicle_registration_cow_latest
hoodie.datasource.hive_sync.database=raw
hoodie.datasource.write.keygenerator.class=org.apache.hudi.keygen.NonpartitionedKeyGenerator
hoodie.datasource.hive_sync.partition_extractor_class=org.apache.hudi.hive.NonPartitionedExtractor
this is my schema registry
```{
"type": "object",
"additionalProperties": false,
"title": "ocr vehicle registration",
"description": "ocr vehicle registration",
"properties": {
"filename": {
"type": "string"
},
"type": {
"type": "number"
},
"name": {
"type": "string"
},
"address": {
"type": "string"
},
"brand": {
"type": "string"
},
"model": {
"type": "string"
},
"engine": {
"type": "string"
},
"chassis": {
"type": "string"
},
"color": {
"type": "string"
},
"sit": {
"type": "number"
},
"type_": {
"type": "string"
},
"capacity": {
"type": "string"
},
"day": {
"type": "number"
},
"month": {
"type": "number"
},
"year": {
"type": "number"
},
"plate": {
"type": "string"
},
"expired_date": {
"type": "string"
}
}
}
```
**Environment Description**
* Hudi version : 0.13.0
* Spark version : 3.2.1
* Hive version : 3.1.0
* Hadoop version : 3.1.1.3.1
* Storage (HDFS/S3/GCS..) : HDFS
* Running on Docker? (yes/no) : no
**Additional context**
Add any other context about the problem here.
**Stacktrace**
```
23/07/06 06:49:21 INFO HoodieTableMetaClient: Initializing hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest as hoodie table hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest
23/07/06 06:49:22 INFO HoodieTableMetaClient: Loading HoodieTableMetaClient from hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest
23/07/06 06:49:22 INFO HoodieTableConfig: Loading table properties from hdfs:/dwh/raw/ocr/ocr_vehicle_registration_cow_latest/.hoodie/hoodie.properties
23/07/06 06:49:23 INFO HoodieTableMetaClient: Finished Loading Table of type COPY_ON_WRITE(version=1, baseFileFormat=PARQUET) from hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest
23/07/06 06:49:23 INFO HoodieTableMetaClient: Finished initializing Table of type COPY_ON_WRITE from hdfs:///dwh/raw/ocr/ocr_vehicle_registration_cow_latest
23/07/06 06:49:23 INFO SparkUI: Stopped Spark web UI at http://ocr-vehicle-registration-src-9b7f828929f5e5ad-driver-svc.spark.svc:8090
23/07/06 06:49:23 INFO KubernetesClusterSchedulerBackend: Shutting down all executors
23/07/06 06:49:23 INFO KubernetesClusterSchedulerBackend$KubernetesDriverEndpoint: Asking each executor to shut down
23/07/06 06:49:23 WARN ExecutorPodsWatchSnapshotSource: Kubernetes client has been closed.
23/07/06 06:49:23 INFO MapOutputTrackerMasterEndpoint: MapOutputTrackerMasterEndpoint stopped!
23/07/06 06:49:23 INFO MemoryStore: MemoryStore cleared
23/07/06 06:49:23 INFO BlockManager: BlockManager stopped
23/07/06 06:49:23 INFO BlockManagerMaster: BlockManagerMaster stopped
23/07/06 06:49:23 INFO OutputCommitCoordinator$OutputCommitCoordinatorEndpoint: OutputCommitCoordinator stopped!
23/07/06 06:49:23 INFO SparkContext: Successfully stopped SparkContext
Exception in thread "main" org.apache.avro.SchemaParseException: Type not supported: object
at org.apache.avro.Schema.parse(Schema.java:1734)
at org.apache.avro.Schema$Parser.parse(Schema.java:1430)
at org.apache.avro.Schema$Parser.parse(Schema.java:1418)
at org.apache.hudi.utilities.schema.SchemaRegistryProvider.getSchema(SchemaRegistryProvider.java:100)
at org.apache.hudi.utilities.schema.SchemaRegistryProvider.getSourceSchema(SchemaRegistryProvider.java:107)
at org.apache.hudi.utilities.schema.SchemaProviderWithPostProcessor.getSourceSchema(SchemaProviderWithPostProcessor.java:42)
at org.apache.hudi.utilities.deltastreamer.DeltaSync.registerAvroSchemas(DeltaSync.java:911)
at org.apache.hudi.utilities.deltastreamer.DeltaSync.(DeltaSync.java:243)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer$DeltaSyncService.(HoodieDeltaStreamer.java:680)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.(HoodieDeltaStreamer.java:148)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.(HoodieDeltaStreamer.java:121)
at org.apache.hudi.utilities.deltastreamer.HoodieDeltaStreamer.main(HoodieDeltaStreamer.java:573)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.apache.spark.deploy.JavaMainApplication.start(SparkApplication.scala:52)
at org.apache.spark.deploy.SparkSubmit.org$apache$spark$deploy$SparkSubmit$$runMain(SparkSubmit.scala:955)
at org.apache.spark.deploy.SparkSubmit.doRunMain$1(SparkSubmit.scala:180)
at org.apache.spark.deploy.SparkSubmit.submit(SparkSubmit.scala:203)
at org.apache.spark.deploy.SparkSubmit.doSubmit(SparkSubmit.scala:90)
at org.apache.spark.deploy.SparkSubmit$$anon$2.doSubmit(SparkSubmit.scala:1043)
at org.apache.spark.deploy.SparkSubmit$.main(SparkSubmit.scala:1052)
at org.apache.spark.deploy.SparkSubmit.main(SparkSubmit.scala)
23/07/06 06:49:23 INFO ShutdownHookManager: Shutdown hook called
23/07/06 06:49:23 INFO ShutdownHookManager: Deleting directory /var/data/spark-a32e287f-3ad6-4bb9-8b35-af6bb8445b85/spark-fb2e8bd9-b281-4391-87d3-75f552ae9e7f
23/07/06 06:49:23 INFO ShutdownHookManager: Deleting directory /tmp/spark-65b0ea58-5e2b-47a3-97fc-d573497d4115```
Contributor guide
No contributing guide indexed for this repository
Research direction
Start at SchemaRegistryProvider.getSchema and the JsonToAvroSchemaConverter configuration shown in the issue, then reproduce the failure with Hudi 0.13.0, Spark 3.2.1, and the supplied schema registry settings. Compare the registry response with the schema format expected by the provider; done means DeltaStreamer accepts the schema and starts without the Avro SchemaParseException.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka, spark
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 35/100