apache / apache/hudi

[SUPPORT] hudi deltastreamer jsonkafka source schema registry fail

Open
#9,132 8 comments 0 reactions 0 assignees View on GitHub
area:ingest priority:high
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.