我让kafka在confluent cloud上运行，在那里我可以用node.js客户端生成数据，数据作为字符串发送，我在confluent cloud中得到以下字段。

然后，我创建了一个 ElasticsearchSink Connector 并将其连接到ElasticSearch云。如果我在ElasticSearch中没有创建任何Map，那么数据的传输将是成功的，格式如下所示。

"_source" : {
          "booked" : false,
          "phone_number" : "919191919191",
          "location" : {
            "lon" : 60.23,
            "lat" : 78.233
          }
        }

现在的问题是，如果我想运行任何 geo queries 它不允许我，并给出以下错误：

"root_cause" : [
      {
        "type" : "query_shard_exception",
        "reason" : "failed to find geo_point field [location]",
        "index_uuid" : "C8Xxu9QlTMKN4Lk1LjpOmQ",
        "index" : "locations"
      }

原因是动态Map不支持geo\u字段。因此，当我尝试为ElasticSearch创建自定义Map，同时创建索引时，如下所示：

PUT /locations
{
  "mappings": {
    "properties": {
      "phone_number": {
        "type": "text"
      },
      "booked": {
        "type": "boolean"
      },
      "location": {
        "type": "geo_point"
        }
      }
  }
}

然后confluent连接器失败并显示以下错误：

There is a mapping collision in your index: Can't merge a non object mapping with an object mapping.

我也试过了 booked 作为一个 text 但事情似乎并不顺利。我没有在合流云上强制任何模式。下面是汇合云的一些基本配置。

如何强制Map以便可以运行 geo queries ElasticSearch？
更新：这个问题之所以持续存在，主要是因为发送给Kafka的数据格式不同

{
    "phone_number": "919191919191",
    "location": {
            "lat": 78.233,
            "lon": 60.23
    },
    "booked": false,
}

{
    "phone_number": "+919191919190",
    "location": " 78.233, 60.23",
    "booked": false,
}

两种格式都无法Map到中的上述定义的Map ElasticSearch 以及 connector sink 显示以下错误：

Received Illegal Argument Exception from Elasticsearch: One of your fields' type does not match the mapped type in Elasticsearch