我想设置卡夫卡德鲁伊摄入,但即使是配置在common.runtime.properties和加入后德鲁伊 - 卡夫卡索引服务,它仍然给我的错误。这个你能帮我吗。我的数据是CSV格式。
{
"type": "kafka",
"spec": {
"dataSchema": {
"dataSource": "london_crime_by_lsoa",
"parser": {
"type": "string",
"parseSpec": {
"format": "csv",
"dimensionsSpec": {
"dimensions": [
"lsoa_code",
"borough",
"major_category",
"minor_category",
{
"name": "value",
"type": "long"
},
{
"name": "year",
"type": "long"
},
{
"name": "month",
"type": "long"
}
]
},
"timestampSpec": {
"column": "year",
"format": "auto"
},
"columns": [
"lsoa_code",
"borough",
"major_category",
"minor_category",
"value",
"year",
"month"
]
}
},
"metricsSpec": [],
"granularitySpec": {
"type": "uniform",
"segmentGranularity": "year",
"queryGranularity": "NONE",
"rollup": false
}
},
"ioConfig": {
"topic": "london_crime_by_lsoa",
"taskDuration": "PT10M",
"useEarliestOffset": "true",
"consumerProperties": {
"bootstrap.servers": "localhost:9092"
}
},
"tuningConfig": {
"type": "kafka",
"maxRowsPerSegment": 500000
}
}
}
执行这个命令之后:
curl -XPOST -H'Content-Type: application/json' -d @quickstart/tutorial/crime_supervisor.json http://localhost:8090/druid/indexer/v1/supervisor
我得到这个错误:
{"error":"Instantiation of [simple type, class org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorSpec] value failed: dataSchema"}
我认为这是你在你的JSON指定规格的方式有问题。您可以直接在您的JSON而不是作为dataSchema
的子属性指定spec
。
下面是你应该遵循以下格式:
{
"type": "kafka",
"dataSchema": {},
"tuningConfig": {},
"ioConfig": {}
}